From d280db514ee7e7053d7deb051bb88c8d44f9062c Mon Sep 17 00:00:00 2001 From: Patrick Buckley Date: Fri, 17 Jul 2026 03:54:55 -0700 Subject: [PATCH] =?UTF-8?q?fix(compaction):=20review=20round=209=20?= =?UTF-8?q?=E2=80=94=20drain-exit=20ownership,=20missed-edge=20settle,=20p?= =?UTF-8?q?re-turn=20hook=20guard?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three point-guards from the ceiling round (no primitive took a hit; correctness yield halved at identical review sensitivity): - The drain's clean-exit wake moved OUT of the function-level try: it runs after the drain has already retired its slot, so a raise out of the wake (the dispatcher re-raises Thread.start failures) could reach the last-resort handler and clear a slot this thread no longer owned — nulling a successor drain's live registration and letting two drains service one list. The wake now runs post-try under its own guard (mirroring _retry_pending_wake), only on the clean-exit path, and the last-resort slot-clear is identity-guarded like every sibling exit seam. The except arm needed a function-local threading import: the module-top import is TYPE_CHECKING-only, so the guard would have NameErrored inside the handler with strict mypy fully green. - The shared settle helper promotes a non-deferred chip that binds onto an already-idle pane: its only sweep fired mid-POST (unbound then) and no message_dispatched ever comes for non-deferred sends, so the chip stayed a permanently retractable "queued" bubble for a delivered message. Keyed on post-bind chip state (also catching a raced folded settle bind just reconciled) and skipping dismiss-in-flight chips — the sweep's own aria-busy discipline. Pinned behaviorally: the helper now executes under node (a 4-row missed-edge matrix), possible since the consumer-less window bridge is gone. - _claim_generation's on_generation_claimed emission is call-guarded: it sits on send()'s pre-turn path, before the user turn is appended and before the fatal handler's coverage, so a raising override degrades to a lost latch-break instead of silently dropping every user message on that session. Cleanups: /command's transport catch and status-less non-2xx bodies are loud now (threading {ok, status} through the parse — deliberately no throw-on-!ok pre-gate, since the busy and error arms ride 409/503); PENDING_SENDS_MAX lives in workstream.py and ChatSession._QUEUE_MAX aliases it (one backpressure bound, structurally incapable of diverging); the send handler's not-ok arm uses _queue_full_response(); the dead window.createQueueController bridge is deleted and the file header's consumer map corrected. --- tests/test_cooperative_compaction.py | 28 ++++ tests/test_interactive_pane_js.py | 93 ++++++++++- tests/test_server_authz.py | 150 +++++++++++++++++- .../console/static/coordinator/coordinator.js | 1 + turnstone/core/session.py | 23 ++- turnstone/core/session_routes.py | 95 ++++++----- turnstone/core/workstream.py | 11 ++ turnstone/shared_static/composer_queue.js | 52 ++++-- turnstone/shared_static/interactive.js | 25 ++- 9 files changed, 412 insertions(+), 66 deletions(-) diff --git a/tests/test_cooperative_compaction.py b/tests/test_cooperative_compaction.py index d5849699..8f628fe9 100644 --- a/tests/test_cooperative_compaction.py +++ b/tests/test_cooperative_compaction.py @@ -2170,6 +2170,34 @@ class TestPreHookUICompat: ) +class TestClaimGenerationHookGuard: + """_claim_generation's on_generation_claimed emission sits on send()'s + PRE-turn path — before the user turn is appended, before + _record_fatal_error's coverage — so a raising override must degrade + to a lost latch-break, never to a silently dropped user message.""" + + def test_raising_claim_hook_does_not_abort_the_claim(self, session): + calls: list[int] = [] + + def _boom(gen: int) -> None: + calls.append(gen) + raise RuntimeError("broadcast backend down") + + session.ui = SimpleNamespace( + on_thinking_start=lambda: None, + on_thinking_stop=lambda: None, + on_error=lambda _m: None, + on_generation_claimed=_boom, + ) + before = session._generation + claimed = session._claim_generation() + assert claimed == before + 1 # claim-state writes stayed infallible + assert session._generation == claimed + assert calls == [claimed] # the hook WAS attempted, then contained + # And again — every claim survives, not just the first. + assert session._claim_generation() == claimed + 1 + + class TestCompactionNoticeStamp: """_compaction_event is the single display-policy site: failed ends carry ``notice`` — renderers show the message iff it is true, instead diff --git a/tests/test_interactive_pane_js.py b/tests/test_interactive_pane_js.py index e94dd9b1..1d3dc6c0 100644 --- a/tests/test_interactive_pane_js.py +++ b/tests/test_interactive_pane_js.py @@ -13,6 +13,8 @@ from __future__ import annotations import re from pathlib import Path +import pytest + _ROOT = Path(__file__).resolve().parent.parent _INTERACTIVE = _ROOT / "turnstone/shared_static/interactive.js" _COMPOSER = _ROOT / "turnstone/shared_static/composer.js" @@ -738,6 +740,12 @@ def test_deferred_send_settle_protocol_pins() -> None: assert composer_queue.count("ctx.optimisticEl.remove()") >= 2, ( "both the retro-convert and queue_full arms must clear the optimistic bubble" ) + # The missed-edge settle: a non-deferred chip binding onto an + # already-idle pane missed its only sweep — the post-bind promote + # (keyed on POST-bind chip state, honoring the aria-busy + # dismiss-in-flight discipline) is what settles it. + assert "!ctx.paneIsBusy()" in composer_queue + assert 'queuedEl.hasAttribute("aria-busy")' in composer_queue # Both panes route their parsed /send response through the helper and # consume the pane-tier settle event; the busy stamp is centralized in # each pane's setBusy (source defaults to "server" — only the send @@ -745,12 +753,91 @@ def test_deferred_send_settle_protocol_pins() -> None: for name, src in (("interactive.js", interactive), ("coordinator.js", coordinator)): assert "settleSendResponse(" in src, f"{name}: settle matrix must be the shared helper" assert "busyIsOptimistic" in src, name + assert "paneIsBusy" in src, f"{name}: the missed-edge settle needs the live flag" assert 'setBusy(true, "optimistic")' in src, f"{name}: optimistic flip must stamp" assert "parsePriority(" in src, f"{name}: shared !!! parse" assert 'case "message_dispatched"' in src, f"{name}: settle event not consumed" assert "settleDeferred(" in src, name - # /command's degraded statuses are all surfaced: busy, running (the - # backstop answer), and error (503 — the worker never spawned; silence - # here reads as success). + # /command's degraded outcomes are ALL surfaced: busy, running (the + # backstop answer), error (503 — the worker never spawned), the + # status-less non-2xx arm (404 / proxy 502), and the transport catch — + # silence at any of them reads as success. assert 'body.status === "running"' in interactive assert 'body.status === "error"' in interactive + assert "Command failed (HTTP " in interactive + assert '"Command failed: " + err.message' in interactive + + +def test_settle_send_response_missed_edge_behavior(tmp_path) -> None: + """Execute the shared settle helper under node and pin the + missed-edge matrix behaviorally (not just textually): a non-deferred + chip binding onto an idle pane promotes; a busy pane, a deferred + chip, and a dismiss-in-flight chip do not.""" + import shutil + import subprocess + + if shutil.which("node") is None: + pytest.skip("node binary not available on PATH") + helper = _ROOT / "turnstone/shared_static/composer_queue.js" + script = tmp_path / "settle_harness.mjs" + script.write_text( + f'const {{ settleSendResponse }} = await import("file://{helper}");\n' + + """ +function makeEl(over) { + const el = { + isConnected: true, + classList: { contains: (c) => c === "msg-queued" }, + dataset: {}, + hasAttribute: () => false, + }; + return Object.assign(el, over || {}); +} +function run(queuedEl, paneBusy, data) { + const calls = []; + const queue = { + bind: (el, id, opts) => { + calls.push("bind"); + // Mirror the real bind: stamp the deferred flag from opts. + if (opts && opts.deferred) el.dataset.deferred = "1"; + }, + promote: () => calls.push("promote"), + remove: () => calls.push("remove"), + addQueuedMessage: () => makeEl(), + }; + settleSendResponse(queue, data, { + queuedEl, + optimisticEl: null, + isBusy: true, + displayText: "t", + priority: "notice", + setBusy: () => {}, + busyIsOptimistic: () => false, + paneIsBusy: () => paneBusy, + renderError: () => {}, + consumeAttachments: () => {}, + }); + return calls; +} +const queued = { status: "queued", msg_id: "m1" }; +let c = run(makeEl(), false, queued); +if (!(c.includes("bind") && c.includes("promote"))) + throw new Error("missed-edge chip must promote: " + c); +c = run(makeEl(), true, queued); +if (c.includes("promote")) throw new Error("busy pane must not promote: " + c); +c = run(makeEl(), false, { status: "queued", msg_id: "m1", deferred: true }); +if (c.includes("promote")) + throw new Error("deferred chip is message_dispatched's: " + c); +c = run(makeEl({ hasAttribute: (a) => a === "aria-busy" }), false, queued); +if (c.includes("promote")) + throw new Error("dismiss-in-flight chip must be left to its DELETE verdict: " + c); +console.log("settle matrix OK"); +""", + encoding="utf-8", + ) + proc = subprocess.run( + ["node", str(script)], + capture_output=True, + text=True, + timeout=15, + ) + assert proc.returncode == 0, f"settle harness failed:\n{proc.stderr}\n{proc.stdout}" diff --git a/tests/test_server_authz.py b/tests/test_server_authz.py index 35619a43..f81775b1 100644 --- a/tests/test_server_authz.py +++ b/tests/test_server_authz.py @@ -2176,11 +2176,11 @@ class TestCompactCommandDispatch: assert [s[0] for s in ws.session.sends] == ["first"] def test_deferred_send_list_saturation_returns_queue_full(self, app_client): - """The deferred list is bounded (_PENDING_SENDS_MAX = 10, mirroring - the interjection queue's cap): the 11th pending send answers - queue_full instead of silently pinning message + attachment bytes - per entry for a whole command window and then costing one - unattended turn each.""" + """The deferred list is bounded (workstream.PENDING_SENDS_MAX — the + shared constant ChatSession._QUEUE_MAX aliases): the 11th pending + send answers queue_full instead of silently pinning message + + attachment bytes per entry for a whole command window and then + costing one unattended turn each.""" client, mgr = app_client ws_id = self._create_ws(client) ws = mgr.get(ws_id) @@ -2292,6 +2292,146 @@ class TestCompactCommandDispatch: assert resp.json()["status"] == "error" assert ws.session.commands == [] # the command never ran + def test_drain_exit_wake_raise_is_contained(self, app_client, monkeypatch, caplog): + """A raising drain-exit wake (session_worker.send re-raises + Thread.start failures) must stay inside the wake's own guard — + it runs AFTER the drain retired its slot, so reaching the + last-resort handler would clear a slot this thread no longer + owns (a successor drain's live registration).""" + import logging + + from turnstone.core import idle_nudge_watcher + + def _raising_gate(ws, *, trigger="unspecified"): + if trigger == "drain-exit": + raise RuntimeError("can't start new thread") + return False + + monkeypatch.setattr(idle_nudge_watcher, "wake_workstream_if_pending", _raising_gate) + client, mgr = app_client + ws_id = self._create_ws(client) + ws = mgr.get(ws_id) + assert ws is not None + gate = threading.Event() + ws.session.compact_gate = gate + client.post( + "/v1/api/command", + json={"command": "/compact", "ws_id": ws_id}, + headers=_auth("user-1"), + ) + client.post( + f"/v1/api/workstreams/{ws_id}/send", + json={"message": "dispatch me"}, + headers=_auth("user-1"), + ) + with caplog.at_level(logging.WARNING, logger="turnstone.core.session_routes"): + gate.set() + wait_until(lambda: ws.session.sends, timeout=8.0) + wait_until(lambda: self._drain_idle(ws), timeout=8.0) + wait_until( + lambda: any("drain_exit_wake_failed" in r.message for r in caplog.records), + timeout=8.0, + ) + # The raise was contained by the wake's own guard — never the + # last-resort handler (whose log would be a false failure for a + # drain that exited cleanly). + assert not any("pending_drain_failed" in r.message for r in caplog.records) + # The seam stays fully serviceable: another defer cycle works. + gate2 = threading.Event() + ws.session.compact_gate = gate2 + client.post( + "/v1/api/command", + json={"command": "/compact", "ws_id": ws_id}, + headers=_auth("user-1"), + ) + client.post( + f"/v1/api/workstreams/{ws_id}/send", + json={"message": "and me"}, + headers=_auth("user-1"), + ) + gate2.set() + wait_until(lambda: len(ws.session.sends) == 2, timeout=8.0) + wait_until(lambda: self._drain_idle(ws), timeout=8.0) + + def test_closed_drain_exit_runs_no_wake(self, app_client, monkeypatch): + """The drain-exit wake backstop belongs to the CLEAN exit only: a + closed workstream is torn down, and firing the gate there would + be work on a corpse. (The clean-exit case is pinned by + test_drain_exit_re_arms_wake_gate_after_pure_retraction.)""" + from turnstone.core import idle_nudge_watcher + + triggers: list[str] = [] + monkeypatch.setattr( + idle_nudge_watcher, + "wake_workstream_if_pending", + lambda ws, *, trigger="unspecified": (triggers.append(trigger), False)[1], + ) + client, mgr = app_client + ws_id = self._create_ws(client) + ws = mgr.get(ws_id) + assert ws is not None + gate = threading.Event() + ws.session.compact_gate = gate + client.post( + "/v1/api/command", + json={"command": "/compact", "ws_id": ws_id}, + headers=_auth("user-1"), + ) + client.post( + f"/v1/api/workstreams/{ws_id}/send", + json={"message": "dies with the ws"}, + headers=_auth("user-1"), + ) + with ws._lock: + ws._closed = True # tombstone, as SessionManager.close sets it + gate.set() + wait_until(lambda: self._drain_idle(ws), timeout=8.0) + assert "drain-exit" not in triggers + assert ws.session.sends == [] + + def test_drain_last_resort_clear_is_identity_guarded(self, app_client, monkeypatch, caplog): + """Drive the drain's OUTER except (loop machinery failing — here + the closed-arm log call) and prove the last-resort slot-clear + executes its identity guard: the slot is this drain's own, so it + clears; a successor's registration would be left alone (the + guard compares thread identity, the sibling exit-seam pattern).""" + import logging + + from turnstone.core import session_routes + + client, mgr = app_client + ws_id = self._create_ws(client) + ws = mgr.get(ws_id) + assert ws is not None + gate = threading.Event() + ws.session.compact_gate = gate + client.post( + "/v1/api/command", + json={"command": "/compact", "ws_id": ws_id}, + headers=_auth("user-1"), + ) + client.post( + f"/v1/api/workstreams/{ws_id}/send", + json={"message": "stranded by the crash"}, + headers=_auth("user-1"), + ) + + def _broken_warning(*_a, **_kw): + raise RuntimeError("log backend down") + + monkeypatch.setattr(session_routes.log, "warning", _broken_warning) + with ws._lock: + ws._closed = True # closed arm with a pending entry → log.warning + with caplog.at_level(logging.ERROR, logger="turnstone.core.session_routes"): + gate.set() + wait_until( + lambda: any("pending_drain_failed" in r.message for r in caplog.records), + timeout=8.0, + ) + # The identity-guarded clear ran for the drain's OWN slot. + wait_until(lambda: ws._pending_drain is None, timeout=8.0) + assert ws.session.sends == [] + def test_exit_command_emits_ended_info_and_never_answers(self, app_client): """should_exit commands shut the session down — the worker emits the ended notice and never launches an answering turn (sends diff --git a/turnstone/console/static/coordinator/coordinator.js b/turnstone/console/static/coordinator/coordinator.js index 6974827e..e6da753e 100644 --- a/turnstone/console/static/coordinator/coordinator.js +++ b/turnstone/console/static/coordinator/coordinator.js @@ -2076,6 +2076,7 @@ function createCoordinatorPane(root, wsId, opts) { priority, setBusy: (b) => setBusy(b), busyIsOptimistic: () => busy && busySource === "optimistic", + paneIsBusy: () => busy, renderError: (msg) => appendText("error", msg, { label: "error" }), consumeAttachments: (attached, droppedIds) => attachments.consume(attached, droppedIds), diff --git a/turnstone/core/session.py b/turnstone/core/session.py index 5d69fd9b..50415312 100644 --- a/turnstone/core/session.py +++ b/turnstone/core/session.py @@ -204,7 +204,7 @@ from turnstone.core.trajectory import ( ) from turnstone.core.watch import WATCH_REMINDER_OPTIONAL_KEYS from turnstone.core.web import check_ssrf, fetch_with_ssrf_guard, strip_html -from turnstone.core.workstream import WorkstreamKind +from turnstone.core.workstream import PENDING_SENDS_MAX, WorkstreamKind from turnstone.prompts import ( INTERACTIVE_CONSENT_CLIENT_TYPES, ClientType, @@ -1370,7 +1370,11 @@ def _tool_turn_meta( class ChatSession: - _QUEUE_MAX = 10 + # The mid-turn interjection queue's cap — an ALIAS of the shared + # per-workstream backpressure bound (see workstream.PENDING_SENDS_MAX): + # the deferred-send list refuses at the same size, so busy-window and + # command-window sends can never silently diverge on saturation. + _QUEUE_MAX = PENDING_SENDS_MAX def __init__( self, @@ -5409,14 +5413,23 @@ class ChatSession: told to break a stale compaction activity latch here — otherwise a force-abandoned compaction's latch suppresses the whole successor turn's pill writes, re-broadcasting "Compacting context…" through - a live turn. getattr-guarded like ``on_aux_usage``: minimal UI - stubs predate the hook and must not crash a claim. + a live turn. getattr-guarded like ``on_aux_usage`` (minimal UI + stubs predate the hook) AND call-guarded: this emission sits on + send()'s PRE-turn path — before the user turn is appended, before + ``_record_fatal_error``'s coverage — so a raising override + (``_broadcast_activity`` is a documented override seam) must + degrade to a lost latch-break, never to a silently dropped user + message. Same policy as ``_compaction_event``'s dispatch tail; + the claim-state writes above stay infallible either way. """ self._generation += 1 self._cancel_event = threading.Event() release = getattr(self.ui, "on_generation_claimed", None) if release is not None: - release(self._generation) + try: + release(self._generation) + except Exception: + log.debug("ui.on_generation_claimed raised; claim proceeds", exc_info=True) return self._generation def _consume_cancel(self, my_generation: int) -> bool: diff --git a/turnstone/core/session_routes.py b/turnstone/core/session_routes.py index c12369da..7418227e 100644 --- a/turnstone/core/session_routes.py +++ b/turnstone/core/session_routes.py @@ -40,7 +40,7 @@ from starlette.routing import Route from turnstone.core.log import get_logger from turnstone.core.session_ui_base import AutoApproveReason -from turnstone.core.workstream import _PendingSend +from turnstone.core.workstream import PENDING_SENDS_MAX, _PendingSend if TYPE_CHECKING: import threading @@ -3954,20 +3954,15 @@ def make_detail_handler(cfg: SessionEndpointConfig) -> Handler: # Deferred sends (command windows / order barrier) # --------------------------------------------------------------------------- -# The dataclass and the order-barrier predicate live with the Workstream -# fields they annotate (turnstone.core.workstream): _PendingSend carries -# the "drain not alive ⇒ nothing claimed" invariant (imported at module -# top), and Workstream.send_barrier_active() is the ONE definition of -# the two-term barrier every dispatch surface consults. - -# Saturation bound for ws._pending_sends — restores the backpressure -# contract the interjection queue (ChatSession._QUEUE_MAX, same value) -# enforced for every busy-window send before the defer seam replaced it. -# Each accepted entry pins its full message text plus materialized -# attachment bytes for the length of a command window and then costs one -# unattended turn, so acceptance must be bounded; callers get the -# standard retryable ``queue_full``. -_PENDING_SENDS_MAX = 10 +# The dataclass, the order-barrier predicate, and the saturation bound +# all live with the Workstream fields they annotate +# (turnstone.core.workstream, imported at module top): _PendingSend +# carries the "drain not alive ⇒ nothing claimed" invariant, +# Workstream.send_barrier_active() is the ONE definition of the two-term +# barrier every dispatch surface consults, and PENDING_SENDS_MAX is the +# shared backpressure bound ChatSession._QUEUE_MAX aliases — one +# constant, structurally incapable of diverging between the interjection +# queue and the deferred list. def _make_drain_thread(ws: Workstream) -> threading.Thread: @@ -4220,10 +4215,18 @@ def _drain_pending_sends(ws: Workstream) -> None: dispatch, it's a cheap no-op re-check after the last turn's own exit backstop. """ + # Function-local imports (file style): ``threading`` is NEEDED here — + # the module-top import is TYPE_CHECKING-only, and the except arm's + # identity guard below would otherwise NameError at runtime inside + # the last-resort handler (masking the original exception and leaving + # the slot permanently held — the exact wedge the handler prevents); + # mypy can't catch that because the type-only import satisfies it. + import threading import time from turnstone.core.idle_nudge_watcher import wake_workstream_if_pending + clean_exit = False try: while True: with ws._lock: @@ -4242,7 +4245,8 @@ def _drain_pending_sends(ws: Workstream) -> None: return if not pending: ws._pending_drain = None - break # clean exit — wake backstop below, outside the lock + clean_exit = True + break # clean exit — wake backstop AFTER the try block entry = pending[0] if ws._worker_running and ws.worker_kind == "command": # The park, relocated server-side: the poll cadence @@ -4322,7 +4326,6 @@ def _drain_pending_sends(ws: Workstream) -> None: # Dispatched: fresh spawn (empty outcome) or interjection # fallback (msg_id preserved) — this entry is done. The # settle event (message_dispatched) fired inside the attempt. - wake_workstream_if_pending(ws, trigger="drain-exit") except Exception: # Never die holding the single-flight slot — a wedged drain would # strand every future deferred send for this workstream. With the @@ -4334,7 +4337,32 @@ def _drain_pending_sends(ws: Workstream) -> None: # the single spawn site is what makes single-flight structural). log.exception("ws.send.pending_drain_failed ws=%s", ws.id[:8] if ws.id else "") with ws._lock: - ws._pending_drain = None + # Identity-guarded, like every sibling exit seam: on paths + # that already RELEASED the slot before raising, an + # unconditional clear here would null a SUCCESSOR drain's + # live registration (two drains servicing one list — FIFO + # inversion, and the barrier reads inactive while the + # survivor holds a claimed entry). The guard makes any + # future post-release statement inside the try safe by + # construction. + if ws._pending_drain is threading.current_thread(): + ws._pending_drain = None + return + # Clean-exit wake backstop — AFTER the try/except, deliberately: this + # runs once the drain has already retired its slot, so a raise out of + # the wake (session_worker.send re-raises Thread.start failures) must + # not reach the last-resort handler above and mutate state this + # thread no longer owns. Own guard, mirroring _retry_pending_wake's + # discipline around the same gate. Not run on the closed arm (the + # workstream is torn down) nor after the except (entries remain, the + # barrier still holds — the gate would just yield). + if clean_exit: + try: + wake_workstream_if_pending(ws, trigger="drain-exit") + except Exception: + log.warning( + "ws.send.drain_exit_wake_failed ws=%s", ws.id[:8] if ws.id else "", exc_info=True + ) def make_send_handler(cfg: SessionEndpointConfig) -> Handler: @@ -4386,7 +4414,7 @@ def make_send_handler(cfg: SessionEndpointConfig) -> Handler: "dropped_attachment_ids"}`` — the send was refused with retry-shortly semantics: the live worker's interjection queue is at capacity, the deferred-send list hit its saturation bound - (``_PENDING_SENDS_MAX`` — the same backpressure contract), or the + (``PENDING_SENDS_MAX`` — the shared backpressure bound), or the drain thread could not be started under resource exhaustion (the entry is rolled back, never phantom-parked). Reservations released; caller should retry. The ``attached_ids`` list is @@ -4546,15 +4574,14 @@ def make_send_handler(cfg: SessionEndpointConfig) -> Handler: # answer for a workstream whose next resolution 404s # would promise a dispatch that can never happen. return JSONResponse({"error": cfg.not_found_label}, status_code=404) - if len(ws._pending_sends) >= _PENDING_SENDS_MAX: - # Saturation backpressure — restores the contract the - # interjection queue enforced for every busy-window - # send before the defer seam replaced it - # (ChatSession._QUEUE_MAX, session.py): without a - # bound, each acked entry pins its message text plus - # materialized attachment bytes for a whole command - # window and then costs one unattended turn — an - # automated caller could OOM the node with 200s. + if len(ws._pending_sends) >= PENDING_SENDS_MAX: + # Saturation backpressure — the SHARED bound + # (workstream.PENDING_SENDS_MAX, which the + # interjection queue's _QUEUE_MAX aliases): without + # a bound, each acked entry pins its message text + # plus materialized attachment bytes for a whole + # command window and then costs one unattended turn + # — an automated caller could OOM the node with 200s. # len() deliberately counts retract-marked husks # awaiting the drain's loop-top purge (DELETE only # marks): a live-only count would let park/retract @@ -4697,19 +4724,11 @@ def make_send_handler(cfg: SessionEndpointConfig) -> Handler: # resolution 404s. Mirror the resolution miss instead. return JSONResponse({"error": cfg.not_found_label}, status_code=404) # queue.Full or session-disappeared race — surface as - # queue_full so clients retry rather than 500 (the same shape - # _defer_send answers for its saturation cap and drain-spawn - # failure). ``attached_ids`` + # queue_full so clients retry rather than 500. ``attached_ids`` # is always empty on this path (the dispatch never took # ownership); the empty arrays preserve the response-shape # guarantee so SDK consumers don't branch on status. - return JSONResponse( - { - "status": "queue_full", - "attached_ids": [], - "dropped_attachment_ids": list(requested_ids), - } - ) + return _queue_full_response() if queue_outcome.get("rejected") == "attachments_busy": # Attachments can't ride a queued user turn (see diff --git a/turnstone/core/workstream.py b/turnstone/core/workstream.py index 904b9e66..7102f1cd 100644 --- a/turnstone/core/workstream.py +++ b/turnstone/core/workstream.py @@ -108,6 +108,17 @@ BULK_CLOSE_STATE_VALUES: frozenset[str] = frozenset( # Deferred sends # --------------------------------------------------------------------------- +# Per-workstream backpressure bound shared by BOTH message-acceptance +# surfaces: the deferred-send list below refuses appends at this size +# (the 11th answers the retryable ``queue_full``), and +# ``ChatSession._QUEUE_MAX`` aliases it for the mid-turn interjection +# queue — one constant, so busy-window and command-window sends can +# never silently diverge on saturation behavior. Each accepted deferred +# entry pins its full message text plus materialized attachment bytes +# for the length of a command window and then costs one unattended +# turn, so acceptance must be bounded. +PENDING_SENDS_MAX = 10 + @dataclass class _PendingSend: diff --git a/turnstone/shared_static/composer_queue.js b/turnstone/shared_static/composer_queue.js index ed8ea7dc..0d421ff6 100644 --- a/turnstone/shared_static/composer_queue.js +++ b/turnstone/shared_static/composer_queue.js @@ -1,7 +1,7 @@ /* composer_queue.js — shared optimistic-queue UI for the chat composer. * * Used by both: - * - turnstone/ui/static/app.js (interactive Pane) + * - turnstone/shared_static/interactive.js (interactive Pane) * - turnstone/console/static/coordinator/coordinator.js (coord IIFE) * * What this owns: @@ -444,8 +444,11 @@ export function createQueueController(opts) { // Unbound chips (no msg_id — the POST round-trip is still in // flight): a quick command window can close inside the round-trip, // firing this edge before bind() could stamp the deferred flag. - // Every response arm (bind/promote/remove/catch) settles unbound - // chips, so the sweep never needs them. + // Every response arm settles unbound chips — including the edge + // this sweep just consumed without them: settleSendResponse's + // post-bind missed-edge promote catches a non-deferred chip whose + // busy→idle edge passed mid-round-trip — so the sweep never needs + // them. if (!el.dataset.msgId) return; _promote(el); }); @@ -518,6 +521,8 @@ export function parsePriority(text) { // busyIsOptimistic(): true iff the pane's CURRENT busy came from this // send flow's optimistic flip and no server state event // has since asserted it (see the panes' busySource stamp) +// paneIsBusy(): the pane's LIVE busy flag (not the send-time snapshot) +// — drives the missed-edge settle below // renderError(msg): pane error row // consumeAttachments(attached_ids, dropped_ids): composer chip sync // @@ -528,7 +533,9 @@ export function parsePriority(text) { // delivered). For deferred sends the optimistic busy flip is then a // lie — no worker exists for this send — so busy clears under the // busyIsOptimistic guard, AFTER bind() so the false-edge idle sweep -// sees dataset.deferred and skips the new chip. +// sees dataset.deferred and skips the new chip. A non-deferred chip +// binding onto an ALREADY-idle pane missed its only sweep — the +// post-bind settle promotes it (see the inline contract). // queue_full — the send was NEVER accepted (interjection cap, deferred- // list saturation, or drain-spawn failure): remove the optimistic // bubble too — leaving it renders loss as delivery — and restore busy @@ -551,11 +558,39 @@ export function settleSendResponse(queue, data, ctx) { attachedCount: (data.attached_ids || []).length, }); if (data.deferred && ctx.busyIsOptimistic()) ctx.setBusy(false); + // Missed-edge settle: if the busy→idle edge already passed during + // the POST round-trip, this chip's only sweep skipped it (unbound + // then) and — for non-deferred sends — no message_dispatched will + // ever fire, so without this the pane keeps a retractable "queued" + // bubble for a delivered message forever. Keyed on the POST-BIND + // chip state, not the wire flag: bind's pre-bind-settle + // reconciliation may have just cleared a raced folded settle's + // deferred flag, and that chip needs the same catch-up. Skips + // dismiss-in-flight chips (aria-busy — _confirmDequeue's verdict + // settles those, the sweep's own discipline) and still-deferred + // chips (message_dispatched owns them). Carries the same bet the + // old edge-promote made: idle can precede the final flush; a + // DELETE racing delivery resolves via the not_found → "already + // sent" arm as ever. + if ( + queuedEl.isConnected && + queuedEl.classList.contains("msg-queued") && + !queuedEl.dataset.deferred && + !queuedEl.hasAttribute("aria-busy") && + !ctx.paneIsBusy() + ) { + queue.promote(queuedEl); + } } else { // Non-deferred queuedEl-absent: the client thought it was idle but // a live worker interjection-queued the message (SSE state_change - // race). A worker provably exists, so its idle edge will arrive — - // keep the historical busy flip. + // race) — keep the historical busy flip. KNOWN residual (the + // missed-edge sibling of the chip arm above): if that worker + // exited during the POST round-trip, no further state event + // arrives and this flip strands an idle pane busy until the next + // real activity; pre-branch behavior, preserved verbatim — + // ctx.paneIsBusy() is the instrument if a future round promotes + // this from residual to fix. ctx.setBusy(true); } ctx.consumeAttachments(data.attached_ids, data.dropped_attachment_ids); @@ -603,8 +638,3 @@ export function settleSendResponse(queue, data, ctx) { if (ctx.queuedEl) queue.promote(ctx.queuedEl); ctx.consumeAttachments(data.attached_ids, data.dropped_attachment_ids); } - -// --- Legacy window bridge --------------------------------------------------- -// Still-classic consumers reach this as a global at event/boot time (after -// this deferred module evaluated). New module code imports instead. -window.createQueueController = createQueueController; diff --git a/turnstone/shared_static/interactive.js b/turnstone/shared_static/interactive.js index b903624d..4da5987c 100644 --- a/turnstone/shared_static/interactive.js +++ b/turnstone/shared_static/interactive.js @@ -3736,12 +3736,16 @@ class Pane { body: JSON.stringify({ command: text, ws_id: this.wsId }), }) .then((r) => + // Thread {ok, status} alongside the parsed body — the loud + // arms below ride NON-2xx codes (busy = 409, error = 503), so + // an /send-style throw-on-!ok pre-gate would wrongly reroute + // them into the transport catch. r.json().then( - (b) => b || {}, - () => ({}), + (b) => ({ ok: r.ok, status: r.status, body: b || {} }), + () => ({ ok: r.ok, status: r.status, body: {} }), ), ) - .then((body) => { + .then(({ ok, status, body }) => { // /compact dispatched onto an already-busy worker reports // {status: "busy"} — surface it (the optimistic busy guard // above can lose that race). @@ -3764,9 +3768,21 @@ class Pane { this.addErrorMessage( body.error || "Command failed to start — retry shortly.", ); + } else if (!ok) { + // Status-less non-2xx (404 unknown workstream, proxy 502 + // HTML): the same silence-reads-as-success rule as the arms + // above. Rendered directly — not thrown into the catch, + // which would double-prefix the message. + this.addErrorMessage( + body.error || "Command failed (HTTP " + status + ")", + ); } }) - .catch(() => {}); + .catch((err) => { + // Transport failure (network drop, abort): the command-echo + // chip is already rendered — say the command did not run. + this.addErrorMessage("Command failed: " + err.message); + }); // Echo as a command chip, not addUserMessage — a slash command is // control-plane input, not a conversational user turn. this.addCommandEcho(text); @@ -3878,6 +3894,7 @@ class Pane { priority, setBusy: (b) => this.setBusy(b), busyIsOptimistic: () => this.busy && this.busySource === "optimistic", + paneIsBusy: () => this.busy, renderError: (msg) => this.addErrorMessage(msg), consumeAttachments: (attached, droppedIds) => this.attachments.consume(attached, droppedIds),