fix(cli): wait for the consumer when the gateway client's event queue is full

[dokterdok]

What does this PR do?

The problem. Suspend the classic CLI with Ctrl-Z while a long answer streams in many small pieces, then fg. The CLI prints Gateway disconnected; turn outcome is unknown… although the gateway is healthy and the turn completes there. The next prompt fails with sent 1000 (OK); no close frame received and never reaches the model; only the one after it works. The same queue also stops a Kanban worker whose card's turn streams more than 4,096 events.

Why.

• The gateway client keeps received event frames in a bounded asyncio.Queue(maxsize=4096). GatewayClient._read adds each one with put_nowait and treats QueueFull as a lost connection: it fails pending requests and queues the "outcome is unknown" error.
• After fg, the frames the socket buffered while the process was stopped come back one after another, and the reader's async for never yields to the event loop while frames are buffered. The CLI's pump (GatewayConnection._pump), which moves frames to the agent thread and never blocks, is ready but gets no turn. The queue fills within milliseconds and the reader fails a consumer that was never slow.
• The CLI then closes the socket, and the next prompt is sent on it during the close handshake, so it fails too.
• kanban.run subscribes the Kanban worker's socket to the card's session (run_task calls connection.resume), but the worker only polls prompt.receipt and never reads its events. A card whose turn streams more than 4,096 events fills the queue, the reader stops, and the next receipt never arrives.

The fix. One invariant: a client's own event backlog never fails its turn.

1. A full queue is not a lost connection while its consumer still drains it. When put_nowait finds the queue full, the reader now waits for room (async with asyncio.timeout(EVENT_STALL_TIMEOUT): await self.events.put(frame)). The pump empties the queue on its next turn and reading resumes. Only a consumer that takes no event for EVENT_STALL_TIMEOUT (30 s) is failed, with the same error and cleanup as before.
2. A client that never reads its events drains them. The Kanban worker client runs a task that discards its socket's events for as long as it polls the card's receipt.

Design choice.

• Park, don't drop or grow. The owner applies the same rule to its own read side (_WS_DISPATCH_QUEUE_MAX in tui_gateway/ws.py: "A full queue parks the reader"). A bigger queue only moves the threshold: the live repro streamed 150,000 deltas. An unbounded one gives up the memory bound.
• The bound stays. The queue still holds at most 4,096 frames. While the reader waits, websockets stops reading from the socket once its own buffer passes max_queue (frames already parsed from the last read stay in memory), so the rest of the backlog stays in the kernel and in the owner's bounded per-viewer fanout. If that fills, the owner sends its session.replay_gap notice: the gateway-side path that fix(gateway): re-attach after session.replay_gap instead of failing the turn #140389 handles.
• A stalled consumer still fails, and nothing hangs. 30 s is the owner's own send deadline (_WS_SEND_DEADLINE_S): a viewer that can't take a frame for that long is gone. The connection can end sooner: with socket reads paused, the client's 20 s keepalive ping can time out. Either way pending requests end, with their own timeout or the usual disconnect error.
• asyncio.timeout, not wait_for. On Python 3.11, wait_for returns normally when it is cancelled just as the inner put completes, so a client closing at that moment would wait on the open socket forever.
• Why not just yield once (await asyncio.sleep(0) before put_nowait): that would fix the CLI's pump, which drains everything in one turn. It wouldn't fix a consumer that awaits per frame, such as the ACP adapter writing each update to the editor; a short write stall would still fail its turn.
• Drain in the Kanban worker rather than stop subscribing it. The subscription is the owner's (run_task); the worker is the side that ignores the events, so it discards them.

Before / after:

Case	Before	After
Ctrl-Z during a 150,000-delta reply, ~14 s, fg	"outcome is unknown"; the next prompt fails; the one after works	the answer shows once, complete; the next prompt answers
Burst larger than 4,096 frames, consumer draining	turn failed	every frame delivered, in order
Kanban card whose turn streams more than 4,096 events	the worker's next receipt never arrives	the worker settles the card
Consumer takes no event at all	failed as the 4,097th frame arrives	failed within 30 s, same error and cleanup; queue still bounded at 4,096
Gateway-side session.replay_gap	unchanged	unchanged (#140389 handles it)
Normal streaming, no backlog	unchanged	unchanged

Related Issue

Fixes #140597

Refs #134889

Overlap:

• fix(gateway): re-attach after session.replay_gap instead of failing the turn #140389 (draft; re-attach after a gateway-side session.replay_gap): no file overlap. It covers the owner dropping a slow viewer; this PR covers the client dropping itself first, which fix(gateway): re-attach after session.replay_gap instead of failing the turn #140389 lists under Known limits ("Client queue"). They are independent and complementary: with both, a resumed CLI reads its backlog, and if the owner's fanout overflowed meanwhile, the gap re-attaches. This commit applies cleanly on fix(gateway): re-attach after session.replay_gap instead of failing the turn #140389's head (6529f8e1de). Its tests plus fix(gateway): re-attach after session.replay_gap instead of failing the turn #140389's replay-gap and reconnect tests and the Kanban worker tests pass together (9 files, 112 passed), and the combined build passed the live retest below.

Type of Change

• 🐛 Bug fix (non-breaking change that fixes an issue)

Changes Made

• hermes_cli/gateway_client.py: EVENT_STALL_TIMEOUT (30 s); GatewayClient._read waits for room when the event queue is full, under asyncio.timeout, and treats a wait that times out like the old QueueFull.
• hermes_cli/kanban_worker_client.py: run discards the socket's events (_discard) while _submit_claim polls the card's receipt.
• tests/hermes_cli/test_gateway_client_backlog.py (new):
  • test_backlog_larger_than_the_client_queue_reaches_the_cli_intact: the real GatewayConnection pump over a GatewayClient whose socket returns 6,000 buffered frames without yielding, as a read after fg does, then message.complete. Every delta reaches the CLI's queue in order, with no loss notice;
  • test_kanban_worker_settles_a_card_whose_turn_outstreams_the_client_queue: the real worker client over a real GatewayClient; kanban.run is followed by 6,000 events, and the next receipt poll finds the card terminal;
  • test_a_consumer_that_stops_draining_is_still_failed_with_the_queue_bounded: with nothing draining, the reader still fails the pending request with the "outcome is unknown" error, and the queue holds its 4,096-frame bound with the error last;
  • test_closing_as_a_waiting_frame_lands_still_ends_the_reader: a close that lands just as the consumer makes room still ends the reader.
• tests/hermes_cli/test_kanban_worker_owner_wait.py, tests/agent/test_kanban_turn_recovery.py: their fake clients get the events queue a real one has.

How to Test

scripts/run_tests.sh tests/hermes_cli/test_gateway_client_backlog.py

1. Fail first. On main at 991f34bcbe, the backlog and Kanban tests fail. The backlog test's loss notice arrives after 4,095 deltas, and the first delta is gone (dropped to make room for the notice). The Kanban test times out waiting for a receipt the stopped reader never delivers. The bound and close tests pass on main, which has no wait. The close test fails on Python 3.11 with wait_for in place of asyncio.timeout: a standalone run of the same steps on 3.11.15 left the close waiting.
2. On this branch: 4 passed. The suites that drive the real client reader or the Kanban worker client give 413 passed, 29 skipped over 49 files: the classic CLI's connection and agent, the gateway chat view, stream-json, the ACP adapter, cron, the session owner, and the Kanban worker against a real owner. One unrelated test fails on main too, on this host: test_kanban_db.py::test_infrastructure_spawn_refusal_never_charges_the_card (the dispatcher's spawn accounting).
3. End to end (kept out of the PR): see the live retest below.
4. scripts/check --base 991f34bcbe: 12 checks ok; code health 0 blocking, 0 advisory.

Independent review: an independent review round and a final review-and-fix pass. The round's one P1 is fixed above with a regression test: on Python 3.11, wait_for can swallow a close that lands as the put completes, so __aexit__ could wait forever. The round's audit of every event consumer also found the Kanban worker site, fixed above. The final pass found no further P1/P2. It reran the 3.11 race and timeout probes on Python 3.11 to 3.14, and corrected a few claims in this description.

Known limits (each as on main):

• Any other client-side loss on a socket the owner still holds open (a consumer stalled for 30 s, a frame over the client's 8 MiB limit) still sends the next prompt into the closing socket, so that prompt fails during the close handshake. The classic CLI's pump never stalls, so this PR removes the CLI's usual route to it.
• The classic CLI's hand-off queue from the pump to the agent thread (GatewayConnection.events) is unbounded, as on main; this PR adds no buffer.
• A client that awaits an RPC answer while nothing drains its events now gets that RPC's error after up to 30 s instead of at once, when more than 4,096 events arrive ahead of the answer. Apart from the Kanban worker (fixed here), on main that happens only in short windows outside any drain loop: the CLI's and the ACP adapter's re-attach after a reconnect, before their pumps start, and the A2A forward's receipt re-poll. fix(gateway): re-attach after session.replay_gap instead of failing the turn #140389's chat view and ACP adapter await the re-attach inside their drain loops; more than 4,096 events arriving ahead of its answer would hold it until the bound (not seen in the combined live run).

Live retest (17 Oct).

What was run: a real gateway daemon and the real classic CLI (hermes --cli) in tmux, in a throwaway HOME and HERMES_HOME, against a local fake OpenAI-compatible model. The reply streamed 40 slow deltas, then 150,000 fast deltas (one character each, with a numbered line marker every 100th). The driver pressed Ctrl-Z after the slow deltas, kept the CLI stopped for about 14 s, ran fg, then sent a second prompt.

Results:

• main (991f34bcbe): the client queue overflowed 13 ms after fg, and the CLI printed "turn outcome is unknown". The second prompt failed with sent 1000 (OK); no close frame received and never reached the model. The owner logged its fanout overflow 1.2 s after fg, after the client had stopped reading.
• This branch (033482fe10): no loss; the reader waited 26 times while it read the backlog. The answer shows once and complete: all 1,500 numbered lines, unique and in order, with the slow prefix once. The owner logged the turn complete (160,650 chars) with no fanout overflow. The second prompt reached the model within the driver's 0.1 s poll, and its reply showed.
• With fix(gateway): re-attach after session.replay_gap instead of failing the turn #140389 (this commit applied on its head 6529f8e1de):
  • the same case passes the same way;
  • with 20,000 deltas of 150 chars (about 3 MB), the owner's fanout also overflowed while the CLI was stopped, so both paths ran in one session. After fg the client queue filled four times while the backlog was read, and the reader waited each time. Then the owner's session.replay_gap arrived, fix(gateway): re-attach after session.replay_gap instead of failing the turn #140389 re-attached, and the turn settled from its receipt with the exact-prefix continuation (no [output resumed]). The second prompt answered. The pane's scrollback kept the last 1,874 lines, contiguous and each once.
  • A 5.6 MB reply instead hit the 8 MiB frame limit already listed in fix(gateway): re-attach after session.replay_gap instead of failing the turn #140389's Known limits: the re-attach succeeded, but the receipt read after it was too large. The queue overflow before it was absorbed here.

Still needs a live retest:

• a real Kanban card whose turn streams more than 4,096 events (covered here by the worker client over a real GatewayClient);
• the ACP adapter and hermes chat --format stream-json with a backlog over 4,096 frames. They share the changed reader, but only the classic CLI was run live.

Checklist

Code

• I've read the Contributing Guide
• My commit messages follow Conventional Commits
• I searched for existing PRs to make sure this isn't a duplicate. Related PRs are listed under Overlap above
• My PR contains only changes related to this fix/feature (no unrelated commits)
• I've run pytest tests/ -q and all tests pass. Not ticked: I ran the related suites above, not the whole tree; CI is the full run
• python scripts/check passes
• I've added tests for my changes
• I've tested on my platform: macOS (Python 3.14; the close test's race also on 3.11)

Documentation & Housekeeping

• Documentation: N/A; the rule is in the code comments
• cli-config.yaml.example: N/A
• CONTRIBUTING.md / AGENTS.md: N/A
• Cross-platform: N/A (asyncio only, no file or process handling)
• Tool descriptions/schemas: N/A

[dokterdok]

Landed via #140968 (merged as f0892ef9b11), salvaged with my commits kept as authored. Closing.