feat(observer): add observer mode and disconnect recovery

scripts/ws_protocol_client.py

@@ -10,6 +10,7 @@ import json
 import os
 import socket
 import struct
+import threading
 import time
 from typing import Any
 
@@ -120,6 +121,131 @@ def join(client: WebSocket, role: str, name: str | None = None) -> dict[str, Any
     return client.receive_type("session")["payload"]
 
 
+def receive_view(client: WebSocket, maximum_messages: int = 800) -> dict[str, Any]:
+    parts: dict[int, dict[str, Any]] = {}
+    sequence: int | None = None
+    expected = 0
+    for _ in range(maximum_messages):
+        message = client.receive()
+        if message.get("type") != "view":
+            continue
+        payload = message["payload"]
+        if sequence != message["seq"]:
+            sequence = message["seq"]
+            expected = payload["chunks"]
+            parts = {}
+        parts[payload["chunk"]] = payload
+        if len(parts) == expected:
+            first = dict(parts[0])
+            first["cells"] = [cell for index in range(expected) for cell in parts[index]["cells"]]
+            first["objects"] = [
+                item for index in range(expected) for item in parts[index]["objects"]
+            ]
+            return first
+    raise RuntimeError("did not receive a complete view")
+
+
+def open_direction(view: dict[str, Any]) -> str:
+    passages = next(cell[1] for cell in view["cells"] if cell[0] == view["self"])
+    for bit, direction in ((1, "north"), (2, "east"), (4, "south"), (8, "west")):
+        if passages & bit:
+            return direction
+    raise AssertionError("player cell has no open passage")
+
+
+def verify_m009(host: str) -> None:
+    clients: list[WebSocket] = []
+    try:
+        player_1 = WebSocket(host)
+        player_2 = WebSocket(host)
+        clients.extend((player_1, player_2))
+        join(player_1, "player", "Наблюдаемый Один")
+        session_2 = join(player_2, "player", "Наблюдаемый Два")
+        observers = [WebSocket(host) for _ in range(4)]
+        clients.extend(observers)
+        for observer in observers:
+            join(observer, "observer")
+            observer.socket.settimeout(30.0)
+
+        observer_views: list[dict[str, Any] | None] = [None] * 4
+        failures: list[BaseException] = []
+
+        def read_observer(index: int) -> None:
+            try:
+                observers[index].receive_type("active", attempts=40)
+                observer_views[index] = receive_view(observers[index])
+            except BaseException as error:  # Preserve worker failure for the main thread.
+                failures.append(error)
+
+        readers = [threading.Thread(target=read_observer, args=(index,)) for index in range(4)]
+        for reader in readers:
+            reader.start()
+        player_1.send("start", {}, "m009-start")
+        player_1.receive_type("countdown")
+        player_2.receive_type("countdown")
+        player_1.receive_type("active")
+        player_2.receive_type("active")
+        player_1_view = receive_view(player_1)
+        player_2_view = receive_view(player_2)
+        for reader in readers:
+            reader.join(timeout=35.0)
+        assert not failures and all(not reader.is_alive() for reader in readers)
+        assert all(view is not None and len(view["cells"]) == 8192 for view in observer_views)
+        print("four-observer full-map delivery: PASS (8192 cells each)")
+
+        observers[0].send("move", {"direction": "north"}, "observer-move")
+        error = observers[0].receive_type("error")
+        assert error["payload"]["code"] == "read_only"
+        print("observer mutation authorization: PASS (read_only)")
+
+        direction = open_direction(player_1_view)
+        player_1.send("move", {"direction": direction}, "before-observer-close")
+        player_1.receive_type("ack")
+        player_1_view = receive_view(player_1)
+        observers[-1].close()
+        clients.remove(observers[-1])
+        observers.pop()
+        time.sleep(0.2)
+        player_1.send("move", {"direction": open_direction(player_1_view)},
+                      "after-observer-close")
+        player_1.receive_type("ack")
+        print("observer disconnect continuity: PASS")
+
+        preserved_cell = player_2_view["self"]
+        preserved_inventory = player_2_view.get("inventory")
+        token_2 = session_2["token"]
+        player_2.close()
+        clients.remove(player_2)
+        pause = player_1.receive_type("pause", attempts=30)["payload"]
+        assert pause["slot"] == 2
+        player_1.send("move", {"direction": "north"}, "paused-move")
+        error = player_1.receive_type("error", attempts=30)
+        assert error["payload"]["code"] == "premature"
+
+        restored = WebSocket(host)
+        clients.append(restored)
+        restored.send("restore", {"token": token_2}, "restore-player-2")
+        restored_session = restored.receive_type("session")["payload"]
+        assert restored_session["slot"] == 2 and restored_session["token"] == token_2
+        player_1.receive_type("resume", attempts=30)
+        restored_view = receive_view(restored)
+        assert restored_view["self"] == preserved_cell
+        assert restored_view.get("inventory") == preserved_inventory
+        print("player pause and reconnect: PASS (same position and inventory)")
+
+        restored.close()
+        clients.remove(restored)
+        player_1.socket.settimeout(70.0)
+        started = time.monotonic()
+        result = player_1.receive_type("match_result", attempts=60)["payload"]
+        elapsed = time.monotonic() - started
+        assert result["result"] == "draw" and elapsed >= 59.0
+        print(f"reconnect timeout draw: PASS ({elapsed:.1f} seconds)")
+    finally:
+        for client in clients:
+            client.close()
+
+
 def verify(host: str) -> None:
     clients: list[WebSocket] = []
     try:
@@ -223,8 +349,11 @@ def main() -> None:
     parser = argparse.ArgumentParser()
     parser.add_argument("host", help="maze32 IP address or hostname")
     parser.add_argument("--verify", action="store_true", help="run the Milestone 004 scenario")
+    parser.add_argument("--verify-m009", action="store_true", help="run the Milestone 009 scenario")
     arguments = parser.parse_args()
-    if arguments.verify:
+    if arguments.verify_m009:
+        verify_m009(arguments.host)
+    elif arguments.verify:
         verify(arguments.host)
     else:
         client = WebSocket(arguments.host)