fix(gateway): a restart wait cuts sockets opened before it again

e1e6f30486 kept a draining owner reachable during a graceful restart's after-turn wait so a
client can attach and answer a turn parked on approval/clarify (#139041). It also stopped the
receive-side check from cutting sockets that were ALREADY open when the wait began. A kanban
worker polling its card's receipt over one of those sockets then never reconnected to the next
owner, while the restart wait waited on that card: `hermes update` sat at the full drain cap
(the e2e-upgrade handoff lane hung 25 min and timed out; bisected to e1e6f30486, and reverting
only gateway/run_api.py turned it green).

Only a socket admitted INTO the wait (admit-time check already passed under restart_wait) now
outlives it; one opened while the owner was ready is cut with 1013 when the wait opens, exactly
as before #139041. Answering a parked prompt from a client that attaches during the wait is
unchanged.

gateway/run_api.py

@@ -279,12 +279,16 @@ class GatewayRuntimeAPI:
             return await self.app(scope, receive, send)
         original_receive = receive
         allow_draining = False
+        admitted_in_wait = False
         from gateway.drain_control import restart_wait_open
 
         async def receive_admitted():
             message = await original_receive()
+            # Only a socket admitted INTO the restart wait (to answer a parked prompt) outlives it. One
+            # opened before the wait is cut as before: its client reconnects to the next owner, and a
+            # held one (a kanban worker polling its card) would otherwise pin the drain to its cap.
             if ((descriptor['state'] != 'ready' or self.runner._draining) and not allow_draining
-                    and not restart_wait_open(self.runner)):
+                    and not (admitted_in_wait and restart_wait_open(self.runner))):
                 return {'type': 'websocket.disconnect', 'code': 1013}
             return message
 
@@ -322,10 +326,11 @@ class GatewayRuntimeAPI:
                     return
                 return await self.app(scope, receive, send)
         allow_draining = not operator
+        admitted_in_wait = restart_wait_open(self.runner)
         # A restart's after-turn wait still owns its turns: an interactive client attaches to answer
         # a parked approval/clarify (#139041); stop() closes this window.
         if ((descriptor['state'] != 'ready' or self.runner._draining) and not allow_draining
-                and not restart_wait_open(self.runner)):
+                and not admitted_in_wait):
             await ws.close(code=1013)
             return
         from gateway.session_authorities import authority_for_profile_id

tests/gateway/test_restart_wait_prompt_reach.py

@@ -234,3 +234,67 @@ def test_update_drain_hint_names_the_draining_profile(tmp_path, monkeypatch):
     work = [{"kind": "chat", "session": "s1"}]
     assert "`hermes -p work --tui`" in format_drain_report(work, remaining_s=60, home=root / "profiles" / "work")
     assert "`hermes --tui`" in format_drain_report(work, remaining_s=60, home=root)
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("opened_in_wait", [False, True])
+async def test_restart_wait_cuts_sockets_opened_before_it(tmp_path, monkeypatch, opened_in_wait):
+    """Only a socket admitted INTO the restart wait outlives it. One opened while the owner was ready
+    (a kanban worker polling its card's receipt) is cut when the wait opens, as before #139041, so its
+    client reconnects to the next owner instead of pinning the drain to its cap."""
+    import json
+
+    from gateway.run_api import GatewayRuntimeAPI
+    from gateway.runtime_bootstrap import TicketStore
+    from hermes_cli import web_server as web
+    from hermes_cli import web_server_chat
+
+    tickets = TicketStore("owner", [str(tmp_path)])
+    ticket = tickets.mint(profile_id=str(tmp_path), subject="human", purpose="interactive")
+    authority = SimpleNamespace(db=SimpleNamespace(db_path=tmp_path / "state.db"), profile_id=str(tmp_path), runner=None)
+    runner = SimpleNamespace(_draining=False, _running=True, session_runtime_descriptor={"state": "ready"},
+                             session_authority=authority, session_ticket_store=tickets)
+
+    def open_restart_wait():
+        runner._draining = True
+        runner.session_runtime_descriptor.update(state="draining", drain_reason="restart_wait")
+
+    if opened_in_wait:
+        open_restart_wait()
+    received = []
+
+    async def handle(ws, **kwargs):
+        await ws.accept(subprotocol="hermes-gateway-v1")
+        open_restart_wait()  # the wait opens while this socket is live
+        while True:
+            message = await ws.receive()
+            if message["type"] == "websocket.disconnect":
+                received.append(("disconnect", message.get("code")))
+                return
+            received.append(("text", message.get("text")))
+
+    async def fallback(*args):
+        pytest.fail("a ticketed native socket must not enter the legacy fallback")
+
+    monkeypatch.setattr("tui_gateway.ws.handle_ws", handle)
+    monkeypatch.setattr(web, "_DASHBOARD_EMBEDDED_CHAT_ENABLED", True)
+    monkeypatch.setattr(web_server_chat, "_ws_request_is_allowed", lambda ws: True)
+    frame = json.dumps({"method": "prompt.receipt"})
+    incoming = iter([{"type": "websocket.connect"}, {"type": "websocket.receive", "text": frame},
+                     {"type": "websocket.disconnect", "code": 1000}])
+
+    async def receive():
+        return next(incoming)
+
+    async def send(value):
+        pass
+
+    scope = {"type": "websocket", "path": "/api/ws", "scheme": "ws", "query_string": b"",
+             "client": ("127.0.0.1", 1234), "server": ("127.0.0.1", 9999),
+             "headers": [(b"sec-websocket-protocol", ("hermes-gateway-v1, hermes-gateway-ticket." + ticket).encode())],
+             "subprotocols": ["hermes-gateway-v1", "hermes-gateway-ticket." + ticket]}
+    await GatewayRuntimeAPI(fallback, runner, web.app)(scope, receive, send)
+    if opened_in_wait:
+        assert received == [("text", frame), ("disconnect", 1000)], received
+    else:
+        assert received == [("disconnect", 1013)], received