fix(kanban): a worker whose socket drops redials instead of ending the turn

A Kanban worker's submitter client turned a mid-turn transport loss into exit 1: the
poll on `prompt.receipt` raised `GatewayClientError`, the retry loop caught only
`GatewayUnavailableError`, and `_submit`'s `(KeyError, ValueError, OSError)` catch
swallowed it as a plain refusal. The owner's `_watch_submitter` then saw the submitter
pid gone and ended the in-flight turn with `os._exit(1)`, so the card was booked
`crashed` with no `worker_result`. On one host: 93 such runs, 80 already admitted and
bound, 0 with a result.

The admission id and session id are the durable handle: the owner keeps the admission
across connections and still answers `prompt.receipt`. Keep them, redial, and re-read
the same attempt instead of resubmitting it; bound the redials with `OWNER_WAIT_S` so a
gone owner still fails honestly. A `runtime_draining` refusal is retried in the same
loop, since nothing was submitted.

Refs #145062

hermes_cli/kanban_worker_client.py

@@ -19,10 +19,26 @@ import time
 # until an owner is ready, so waiting can never run the claim twice.
 OWNER_WAIT_S = 180.0
 _TRANSIENT_OWNER_STATES = frozenset({'draining', 'starting'})
+# An owner that is up but draining refuses in band (JSON-RPC error), before admitting anything.
+_TRANSIENT_RPC_REASONS = frozenset({'runtime_draining'})
 # Past the board lock wait: the owner's session setup and receipt before and after the write.
 ADMISSION_ANSWER_MARGIN_S = 60.0
 
 
+class _Redial(Exception):
+    """The socket died after the owner admitted the claim: reconnect and re-read the receipt.
+
+    The admission id and session id are the durable handle for the attempt (the owner keeps the
+    admission across client connections), so a dead socket is a transport event rather than a turn
+    outcome. Exiting instead would let ``_watch_submitter`` end the owner's in-flight turn.
+    """
+
+    def __init__(self, admission):
+        super().__init__('gateway connection lost after the owner admitted the claim')
+        self.admission = admission
+
+
+
 def admission_answer_budget():
     """Seconds to wait for ``kanban.run``'s answer: the longest the owner's admission write can
     wait on the board lock (``write_txn`` retries ``BEGIN IMMEDIATE`` ``_BUSY_MAX_RETRIES`` times,
@@ -33,10 +49,16 @@ def admission_answer_budget():
 
 
 async def run(params, board_db):
-    from hermes_cli.gateway_client import connect_gateway, GatewayClientError, GatewayUnavailableError
+    from hermes_cli.gateway_client import (connect_gateway, GatewayClientError, GatewayRPCError,
+                                           GatewayUnavailableError)
     if os.environ.get('HERMES_TUI_GATEWAY_URL'):
         raise GatewayClientError('Kanban requires the assigned local profile owner')
     deadline = time.monotonic() + OWNER_WAIT_S
+    # A lost socket is not a lost attempt: after the owner admits the claim, the admission id is
+    # what the next connection re-reads. Bounded once, so a flapping gateway cannot hold the worker
+    # for ever and a truly gone owner still fails honestly.
+    admission = None
+    redial_deadline = None
     while True:
         try:
             async with connect_gateway() as client:
@@ -45,9 +67,23 @@ async def run(params, board_db):
                 # reader stops, with the next receipt behind it.
                 drain = asyncio.create_task(_discard(client.events))
                 try:
-                    return await _submit_claim(client, params, board_db)
+                    return await _submit_claim(client, params, board_db, admission)
                 finally:
                     drain.cancel()
+        except _Redial as exc:
+            admission = exc.admission
+            if redial_deadline is None:
+                redial_deadline = time.monotonic() + OWNER_WAIT_S
+            if time.monotonic() >= redial_deadline:
+                # The attempt stays readable for its owner; failing honestly beats resubmitting it.
+                raise GatewayClientError('Kanban owner connection lost after admission: the attempt '
+                                         'is retained by the owner; do not resend the work') from None
+            await asyncio.sleep(1)
+        except GatewayRPCError as exc:
+            # Nothing was submitted when the owner refused the claim, so this is the host's state.
+            if admission is not None or str(exc) not in _TRANSIENT_RPC_REASONS or time.monotonic() >= deadline:
+                raise
+            await asyncio.sleep(1)
         except GatewayUnavailableError as exc:
             if exc.state not in _TRANSIENT_OWNER_STATES or time.monotonic() >= deadline:
                 raise
@@ -59,7 +95,19 @@ async def _discard(events):
         await events.get()
 
 
-async def _submit_claim(client, params, board_db):
+async def _receipt(client, session_id, admission_id):
+    """One ``prompt.receipt`` read. A transport loss becomes a redial; an owner's answer does not."""
+    from hermes_cli.gateway_client import GatewayClientError, GatewayRPCError
+    from websockets.exceptions import ConnectionClosed
+    try:
+        return await client.rpc('prompt.receipt', session_id=session_id, admission_id=admission_id)
+    except GatewayRPCError:
+        raise
+    except (GatewayClientError, ConnectionClosed, OSError, TimeoutError):
+        raise _Redial((session_id, admission_id)) from None
+
+
+async def _submit_claim(client, params, board_db, admission=None):
     from hermes_cli.gateway_client import GatewayClientError
     db = str(await asyncio.to_thread(Path(board_db).resolve))
     # The owner answers only after it checks the claim and writes its ``owner_admitted`` marker
@@ -69,12 +117,17 @@ async def _submit_claim(client, params, board_db):
     # admission was left ``unknown``). The wait is still finite: once the marker commits,
     # ``enforce_max_runtime`` and TTL reclaim skip the run (``owner_reclaim_paused``), so an owner
     # that stalls after admitting is bounded only here. A hang-up fails the wait sooner.
-    accepted = await client.rpc('kanban.run', _timeout=admission_answer_budget(), **dict(params, db=db))
-    sid, receipt = accepted['session_id'], accepted['receipt']
-    print(f'Session: {sid}', file=sys.stderr)
+    if admission is None:
+        accepted = await client.rpc('kanban.run', _timeout=admission_answer_budget(), **dict(params, db=db))
+        sid, receipt = accepted['session_id'], accepted['receipt']
+        print(f'Session: {sid}', file=sys.stderr)
+        admission = (sid, receipt['admission_id'])
+    else:
+        # A redial re-reads the attempt the owner already holds; it never submits the claim again.
+        receipt = await _receipt(client, *admission)
     while receipt['status'] not in {'terminal', 'unknown'}:
         await asyncio.sleep(.2)
-        receipt = await client.rpc('prompt.receipt', session_id=sid, admission_id=receipt['admission_id'])
+        receipt = await _receipt(client, *admission)
     if receipt['status'] == 'unknown':
         raise GatewayClientError('Kanban execution unknown; retained by owner')
     from gateway.session_kanban import worker_result

tests/hermes_cli/test_kanban_worker_socket_recovery.py

@@ -0,0 +1,132 @@
+"""A Kanban worker whose gateway socket dies mid-turn redials and re-reads the same receipt.
+
+A dead socket is a transport event, not a turn outcome: the owner keeps the admission and answers
+``prompt.receipt`` over the next connection, so exiting 1 there loses work that was already running
+(``_watch_submitter`` then ends the owner's in-flight turn). Measured before the fix: 93 runs booked
+``crashed`` on a gateway disconnect, 80 of them already admitted and bound, 0 with a
+``worker_result``. An owner's own refusal is still an answer, and a genuine refusal still fails.
+"""
+
+import asyncio
+from contextlib import asynccontextmanager
+
+import pytest
+
+from hermes_cli import gateway_client, kanban_worker_client
+
+
+class _Owner:
+    """Scripted owner: ``connects`` holds one entry per dial (an exception fails the dial), and
+    ``receipts`` holds one entry per ``prompt.receipt`` read (an exception is a transport loss).
+    ``submit_failures`` holds entries consumed by ``kanban.run``."""
+
+    def __init__(self, connects=None, receipts=None, submit_failures=None):
+        self.connects = list(connects or [])
+        self.receipts = list(receipts or [])
+        self.submit_failures = list(submit_failures or [])
+        self.calls = []
+        self.dials = 0
+
+    def connect(self):
+        owner = self
+
+        @asynccontextmanager
+        async def _connect():
+            owner.dials += 1
+            step = owner.connects.pop(0) if owner.connects else None
+            if isinstance(step, Exception):
+                raise step
+
+            class _Client:
+                def __init__(self):
+                    self.events = asyncio.Queue()
+
+                async def rpc(self, method, **params):
+                    owner.calls.append(method)
+                    if method == "kanban.run":
+                        failure = owner.submit_failures.pop(0) if owner.submit_failures else None
+                        if failure is not None:
+                            raise failure
+                        return {"session_id": "s1", "receipt": {"status": "pending", "admission_id": "a1"}}
+                    status = owner.receipts.pop(0) if owner.receipts else "terminal"
+                    if isinstance(status, Exception):
+                        raise status
+                    return {"status": status}
+
+            yield _Client()
+
+        return _connect
+
+
+def _disconnect():
+    return gateway_client.GatewayClientError(
+        "Gateway disconnected; turn outcome is unknown. Resume the printed session ID")
+
+
+@pytest.fixture(autouse=True)
+def _worker_env(monkeypatch, tmp_path):
+    monkeypatch.delenv("HERMES_TUI_GATEWAY_URL", raising=False)
+    monkeypatch.setattr(kanban_worker_client.asyncio, "sleep", _no_sleep)
+    from gateway import session_kanban
+    monkeypatch.setattr(session_kanban, "worker_result",
+                        lambda *_: {"exit_code": 0, "last_output": "card closed"})
+    return tmp_path
+
+
+async def _no_sleep(_seconds):
+    return None
+
+
+@pytest.mark.asyncio
+async def test_a_socket_lost_mid_turn_redials_and_settles_the_same_attempt(monkeypatch, _worker_env):
+    owner = _Owner(connects=[None, None],
+                   receipts=["pending", _disconnect(), "terminal"])
+    monkeypatch.setattr(gateway_client, "connect_gateway", owner.connect())
+
+    assert await kanban_worker_client.run({"task_id": "t1"}, str(_worker_env / "kanban.db")) == 0
+    # The claim is submitted once: the redial re-reads the receipt, it does not resubmit.
+    assert owner.calls.count("kanban.run") == 1
+    assert owner.dials == 2
+
+
+@pytest.mark.asyncio
+async def test_a_draining_owner_is_retried_before_anything_is_admitted(monkeypatch, _worker_env):
+    owner = _Owner(connects=[None, None],
+                   submit_failures=[gateway_client.GatewayRPCError("runtime_draining")],
+                   receipts=["terminal"])
+    monkeypatch.setattr(gateway_client, "connect_gateway", owner.connect())
+
+    assert await kanban_worker_client.run({"task_id": "t1"}, str(_worker_env / "kanban.db")) == 0
+    assert owner.calls.count("kanban.run") == 2
+
+
+@pytest.mark.asyncio
+async def test_an_owner_refusal_is_not_a_redial(monkeypatch, _worker_env):
+    owner = _Owner(connects=[None, None],
+                   submit_failures=[gateway_client.GatewayRPCError("claim_mismatch")])
+    monkeypatch.setattr(gateway_client, "connect_gateway", owner.connect())
+
+    with pytest.raises(gateway_client.GatewayRPCError):
+        await kanban_worker_client.run({"task_id": "t1"}, str(_worker_env / "kanban.db"))
+    assert owner.dials == 1
+
+
+@pytest.mark.asyncio
+async def test_a_gateway_that_never_comes_back_fails_honestly(monkeypatch, _worker_env):
+    owner = _Owner(connects=[None, gateway_client.GatewayUnavailableError("no owner", "inaccessible")],
+                   receipts=["pending", _disconnect()])
+    monkeypatch.setattr(gateway_client, "connect_gateway", owner.connect())
+
+    with pytest.raises(gateway_client.GatewayClientError):
+        await kanban_worker_client.run({"task_id": "t1"}, str(_worker_env / "kanban.db"))
+
+
+@pytest.mark.asyncio
+async def test_the_redial_budget_bounds_a_flapping_gateway(monkeypatch, _worker_env):
+    monkeypatch.setattr(kanban_worker_client, "OWNER_WAIT_S", 0.0)
+    owner = _Owner(connects=[None, None], receipts=["pending", _disconnect(), "terminal"])
+    monkeypatch.setattr(gateway_client, "connect_gateway", owner.connect())
+
+    with pytest.raises(gateway_client.GatewayClientError):
+        await kanban_worker_client.run({"task_id": "t1"}, str(_worker_env / "kanban.db"))
+    assert owner.calls.count("kanban.run") == 1