fix(hosted-room): peer client resolves a member's unknown run instead of wedging

Mechanism: e3dcd1ea20 made run_projection report a paused (restart-lost)
admission as status "unknown". PeerRunsHTTPClient._poll_receipt only knows the
main-era run states, so every room-side poll raised "peer returned a mismatched
run status": _wait_for_terminal, _inspect_session and Stop never settled, and
nothing called resolve-unknown, so the member's later turns queued behind the
unknown head forever.

Fix: when the poll body reports "unknown" for this run, POST
/v1/runs/{id}/resolve-unknown with the admission_id and execution_generation
from that same body (room grants carry `stop`, the route's permission; an
exact retry replays the receipt), then cache the run as terminal
"interrupted" with "member gateway restarted", which is what the room saw
before e3dcd1ea20. Genuinely unrecognised statuses still hit the mismatch
guard. api-server.md notes that older room homes do not know the status.

Test (red on r13-base2: PeerRunsHTTPError mismatched run status):
tests/gateway/test_peer_client_unknown_run.py runs the real _poll_receipt over
real /v1/runs routes, a real SessionAuthority + SQLite and a real owner
restart; the member's queued follower then runs exactly once.

Review: dokterdok N85

tests/gateway/test_peer_client_unknown_run.py

@@ -0,0 +1,37 @@
+"""A hosted-room peer whose member gateway restarted mid-turn reports the run ``unknown``: the
+room's poll resolves that exact admission and settles the run interrupted, and the member's
+queued follower then runs once (the room no longer wedges on "mismatched run status")."""
+import asyncio
+
+import pytest
+
+from tests.gateway.test_api_unknown_resolution import (
+    _AUTH_HEADERS, _close_fixture, _restarted_api_runs)
+from tui_gateway.hosted_room_peer_http import PeerRunsHTTPClient, PeerRunsHTTPError
+
+
+@pytest.mark.asyncio
+async def test_peer_poll_resolves_unknown_member_run(tmp_path, monkeypatch):
+    client, adapters, store, owner, unknown, _follower, executed, tasks = (
+        await _restarted_api_runs(tmp_path, monkeypatch))
+    peer, loop = PeerRunsHTTPClient(base_url="http://127.0.0.1:9", api_key="k" * 32), asyncio.get_running_loop()
+
+    def forward(path, *, method="GET", body=None, headers=None, room_grant=None):
+        async def call():
+            response = await client.request(method, path, json=body, headers=_AUTH_HEADERS)
+            payload = await response.json()
+            if response.status >= 400:
+                raise PeerRunsHTTPError(str(payload), status_code=response.status)
+            return payload
+        return asyncio.run_coroutine_threadsafe(call(), loop).result(10)
+
+    monkeypatch.setattr(peer, "_request", forward)
+    record = {"run_id": unknown["request_id"], "task_id": "room-task", "execution_generation": 1}
+    try:
+        status = await asyncio.to_thread(peer._poll_receipt, record, grant="room-grant-token")
+        assert status["status"] == "interrupted" and status["run_id"] == unknown["request_id"]
+        assert ("room-task", 1) in peer._terminal_receipts
+        await asyncio.wait_for(owner.sessions["api-session"].task, 10)
+        assert executed == ["FOLLOWER"]
+    finally:
+        await _close_fixture(client, adapters, store, tasks)

tui_gateway/hosted_room_peer_http.py

@@ -472,6 +472,8 @@ class PeerRunsHTTPClient:
         entry = {"delay": delay, "next_poll_at": now + delay}
         try:
             full = self._request(_run_path(record), room_grant=self._require_room_grant(grant))
+            if full.get("status") == "unknown" and str(full.get("run_id") or "") == run_id:
+                full = self._resolve_unknown(record, full, grant)
             status = {key: full[key] for key in _RUN_STATUS_KEYS if key in full}
             if (
                 str(status.get("run_id") or "") != run_id
@@ -487,6 +489,14 @@ class PeerRunsHTTPClient:
                 (str(record["task_id"]), int(record["execution_generation"])))
         return status
 
+    def _resolve_unknown(self, record: Mapping[str, Any], full: Mapping[str, Any], grant: str) -> dict[str, Any]:
+        """A member gateway that restarted mid-turn reports the run ``unknown`` and holds its later
+        turns behind it. The room cannot resume that lost execution: acknowledge exactly the polled
+        admission (``stop`` grant; an exact retry replays the receipt) and settle it interrupted."""
+        self._request(_run_path(record, "resolve-unknown"), method="POST", room_grant=grant, body={
+            "admission_id": full.get("admission_id"), "execution_generation": full.get("execution_generation")})
+        return {"run_id": full["run_id"], "status": "interrupted", "error": "member gateway restarted"}
+
     def history(
         self, *, room_id: str, profile: str, session_id: str, grant: str,
     ) -> Sequence[Mapping[str, Any]]: