my-sea: async witness — spectators see the owner's draw land live via WS, no refresh — TDD

The bud watching @owner's sea now sees each card appear in real time instead of
having to refresh. Follows the epic RoomConsumer broadcast pattern (view ->
group_send -> consumer handler -> send_json), keyed on the owner (mysea_<owner>)
since my-sea has no Room.

- apps/gameboard/consumers.py: MySeaSpectateConsumer — read-only WS; membership
  gate matches voice (owner OR present invitee: ACCEPTED + deposited + not
  left). Relays a `sea_draw` event carrying the owner's full hand.
- apps/gameboard/routing.py + core/asgi.py: ws/my-sea-spectate/<owner_id>/.
- gameboard/views.py: _notify_sea_draw(owner_id, hand) — best-effort, guarded
  group_send so a down/missing channel layer can't break the solo draw. Fired
  from my_sea_lock (both the create + the mid-draw-upsert branch) and from
  my_sea_delete (empty hand -> clears the spectators' cross).
- my_sea_visit.html: a WS listener fills the cross live — SeaDeal.register(card,
  '.sea-pos-'+pos, isLevity) reuses _fillSlot (incl. the --rank-long squeeze) +
  seeds the slot clickable into the stage; a DEL re-empties cleared slots.
  Capped reconnect for transient blips.

Tests: 5 channels ITs (owner/present-invitee connect + receive; unauth /
stranger / accepted-not-present rejected); +2 view ITs (lock broadcasts owner+
hand; lock still 200s when the broadcast raises). Client fill needs live
two-party verification on staging (Redis up).

Code architected by Disco DeDisco <discodedisco@outlook.com>
Git commit message Co-Authored-By:
Claude Opus 4.8 (1M context) <noreply@anthropic.com>

src/apps/gameboard/tests/integrated/test_spectate_consumer.py

@@ -0,0 +1,81 @@
+"""Channels ITs for MySeaSpectateConsumer — the async-witness WS that pushes
+the owner's draw to watching invitees (user-spec 2026-05-29).
+
+Same shape as the voice consumer tests: scope injected directly, in-memory
+channel layer, `@tag("channels")` (excluded from the default run).
+"""
+
+from channels.db import database_sync_to_async
+from channels.testing.websocket import WebsocketCommunicator
+from django.contrib.auth.models import AnonymousUser
+from django.test import TransactionTestCase, override_settings, tag
+from django.utils import timezone
+
+from apps.gameboard.consumers import MySeaSpectateConsumer
+from apps.gameboard.models import SeaInvite
+from apps.lyric.models import User
+
+
+TEST_CHANNEL_LAYERS = {
+    "default": {"BACKEND": "channels.layers.InMemoryChannelLayer"},
+}
+
+
+@tag("channels")
+@override_settings(CHANNEL_LAYERS=TEST_CHANNEL_LAYERS)
+class MySeaSpectateConsumerTest(TransactionTestCase):
+    def setUp(self):
+        self.owner = User.objects.create(email="owner@test.io", username="discoman")
+        self.bud = User.objects.create(email="bud@test.io", username="budster")
+
+    def _comm(self, user):
+        comm = WebsocketCommunicator(
+            MySeaSpectateConsumer.as_asgi(),
+            f"/ws/my-sea-spectate/{self.owner.id}/",
+        )
+        comm.scope["user"] = user
+        comm.scope["url_route"] = {"kwargs": {"owner_id": self.owner.id}}
+        return comm
+
+    def _invite(self, present):
+        return SeaInvite.objects.create(
+            owner=self.owner, invitee=self.bud, invitee_email=self.bud.email,
+            status=SeaInvite.ACCEPTED, accepted_at=timezone.now(),
+            token_deposited_at=timezone.now() if present else None,
+        )
+
+    async def test_owner_can_watch(self):
+        comm = self._comm(self.owner)
+        connected, _ = await comm.connect()
+        self.assertTrue(connected)
+        await comm.disconnect()
+
+    async def test_unauthenticated_is_rejected(self):
+        comm = self._comm(AnonymousUser())
+        connected, _ = await comm.connect()
+        self.assertFalse(connected)
+
+    async def test_stranger_is_rejected(self):
+        comm = self._comm(self.bud)   # no invite
+        connected, _ = await comm.connect()
+        self.assertFalse(connected)
+
+    async def test_accepted_but_not_present_is_rejected(self):
+        await database_sync_to_async(self._invite)(present=False)
+        comm = self._comm(self.bud)
+        connected, _ = await comm.connect()
+        self.assertFalse(connected)
+
+    async def test_present_invitee_receives_the_owners_hand(self):
+        await database_sync_to_async(self._invite)(present=True)
+        comm = self._comm(self.bud)
+        connected, _ = await comm.connect()
+        self.assertTrue(connected)
+        from channels.layers import get_channel_layer
+        hand = [{"position": "lay", "card_id": 1,
+                 "reversed": False, "polarity": "gravity"}]
+        await get_channel_layer().group_send(
+            f"mysea_{self.owner.id}", {"type": "sea_draw", "hand": hand})
+        msg = await comm.receive_json_from()
+        self.assertEqual(msg, {"type": "sea_draw", "hand": hand})
+        await comm.disconnect()

src/apps/gameboard/tests/integrated/test_views.py

@@ -1860,6 +1860,31 @@ class MySeaLockHandViewTest(TestCase):
         response = self.client.get(self.url)
         self.assertEqual(response.status_code, 405)
 
+    def test_lock_broadcasts_hand_to_spectators(self):
+        # The owner's draw pushes the full hand to watching invitees (async
+        # witness, 2026-05-29) — my_sea_lock calls _notify_sea_draw(owner, hand).
+        import json
+        from unittest.mock import patch
+        payload = self._build_payload()
+        with patch("apps.gameboard.views._notify_sea_draw") as mock_notify:
+            self.client.post(self.url, data=json.dumps(payload),
+                             content_type="application/json")
+        mock_notify.assert_called_once()
+        owner_id, hand = mock_notify.call_args[0]
+        self.assertEqual(owner_id, self.user.id)
+        self.assertEqual(hand, payload["hand"])
+
+    def test_lock_succeeds_even_when_the_broadcast_fails(self):
+        # The broadcast is best-effort — a down/missing channel layer must NOT
+        # break the solo draw save (guarded in _notify_sea_draw).
+        import json
+        from unittest.mock import patch
+        with patch("channels.layers.get_channel_layer",
+                   side_effect=Exception("redis down")):
+            resp = self.client.post(self.url, data=json.dumps(self._build_payload()),
+                                    content_type="application/json")
+        self.assertEqual(resp.status_code, 200)
+
     def test_lock_post_creates_my_sea_draw_for_user(self):
         import json
         from apps.gameboard.models import MySeaDraw