diff --git a/CHANGELOG.md b/CHANGELOG.md index 2949399c..7763076d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -175,7 +175,8 @@ Earlier stable lines (`stable/1.6`, `stable/1.5`) are frozen. ### Fixed -- **A failed worker-thread spawn no longer wedges the workstream.** If +- **A failed worker-thread spawn no longer wedges the workstream — at + either spawn site — and never masquerades as success.** If `Thread.start()` itself raised (thread exhaustion, out-of-memory), the dispatcher had already claimed the worker slot but the flag's only clearer lived in the never-started thread — the workstream looked idle @@ -183,7 +184,14 @@ Earlier stable lines (`stable/1.6`, `stable/1.5`) are frozen. didn't exist, until an operator force-cancel. The claim is now rolled back under the lock and the error propagates, so the workstream is dispatchable again as soon as resources recover. Affected every - dispatch path (sends, wakes, retries, deferred-send drain, init). + dispatch path (sends, wakes, retries, deferred-send drain, init). The + same failure at the deferred-send drain's own spawn rolls back the + just-accepted entry and answers the retryable `queue_full` (previously + a 500 landed *after* the entry was registered — an invisible, + unretractable phantom that later dispatched as duplicate turns), and a + `/command` whose worker never spawned now answers **503** + `{"status": "error"}` instead of the generic 200 ok that told SDK + callers their `/clear` or `/resume` had applied. - **Manual `/compact` from the web UI: no phantom user turn, no frozen server, cancellable.** A slash command typed into the web composer no @@ -225,17 +233,28 @@ Earlier stable lines (`stable/1.6`, `stable/1.5`) are frozen. retractable until dispatch via the same `DELETE .../send` used for queued interjections (node-local, in-memory — the API reference documents the at-most-once durability contract). Deferred responses - carry `"deferred": true`, the pending list is the **order authority** + carry `"deferred": true`; the pending list is the **order authority** (a fresh send — or a coordinator dispatch, or a queued-nudge wake — - lines up behind acknowledged entries instead of overtaking them, and - the wake gate re-arms at the drain's exit even when everything pending - was retracted), a dispatch crash re-queues the entry instead of eating - an acknowledged message, and each dispatch emits a pane-tier + lines up behind acknowledged entries instead of overtaking them, with + the two-term barrier defined once on the workstream so the wake gate + also honors a claimed entry whose dispatch is mid-flight, and the gate + re-arms at the drain's exit even when everything pending was + retracted); acceptance is **bounded** (10 pending per workstream — the + interjection queue's own backpressure contract; the 11th answers the + retryable `queue_full` instead of pinning attachment bytes without + limit and then running one unattended turn per entry); a dispatch + crash re-queues the entry instead of eating an acknowledged message, + and a drain thread that fails to *start* rolls the acceptance back and + answers `queue_full` rather than parking a phantom the client can + neither see nor retract; each dispatch emits a pane-tier `message_dispatched` event (`folded: true` for interjection fold-ins) so queued-bubble UI keeps its retract affordance exactly until the message truly leaves — including when the send was accepted by a pane that believed the workstream idle, which now renders a real queued - chip instead of a sent-looking bubble. + chip instead of a sent-looking bubble, releases the composer (a + deferred send has no running worker to wait on), and cleans up fully + when the send is refused or the chip retracted instead of stranding + the pane in Stop mode. A `/compact` raced against an in-flight turn is refused with an explicit busy response. Every other slash command runs through the same worker slot too — mutual exclusion against sends, a running compaction, diff --git a/docs/api-reference.md b/docs/api-reference.md index 45f33959..8e9bfac9 100644 --- a/docs/api-reference.md +++ b/docs/api-reference.md @@ -807,8 +807,12 @@ Sends a user message to a workstream. Spawns a daemon worker thread that calls list (a command window holds the slot, or earlier deferred sends are pending) and dispatched as its own full-fidelity send afterwards; see the defer contract under `POST /v1/api/command`. -- `{"status": "queue_full", ...}` — the live worker's queue is at capacity; - retry shortly. +- `{"status": "queue_full", ...}` — 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 (10 pending — the same + backpressure contract), or the deferred-send drain could not be started + under resource exhaustion (the message was **not** accepted; nothing is + parked). - `{"status": "attachments_busy", ...}` — attachments can't ride a queued turn; the staged uploads survive for a retry once the worker idles. @@ -940,11 +944,12 @@ or `{"status": "running"}` as above. **Error responses:** -| Status | Body | Condition | -|--------|------------------------------------|----------------------------------| -| 400 | `{"error": "Empty command"}` | Command is empty | -| 404 | `{"error": "Unknown workstream"}` | `ws_id` not found | -| 409 | `{"status": "busy", "error": ...}` | A turn/command holds the worker | +| Status | Body | Condition | +|--------|-------------------------------------|--------------------------------------------------| +| 400 | `{"error": "Empty command"}` | Command is empty | +| 404 | `{"error": "Unknown workstream"}` | `ws_id` not found | +| 409 | `{"status": "busy", "error": ...}` | A turn/command holds the worker | +| 503 | `{"status": "error", "error": ...}` | The command worker could not be started (resource exhaustion) — the command did **not** run; retry shortly | --- diff --git a/sdk/typescript/openapi-server.json b/sdk/typescript/openapi-server.json index 185edf28..0934efb6 100644 --- a/sdk/typescript/openapi-server.json +++ b/sdk/typescript/openapi-server.json @@ -410,6 +410,16 @@ } } } + }, + "503": { + "description": "Error 503", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ErrorResponse" + } + } + } } } } diff --git a/tests/test_cooperative_compaction.py b/tests/test_cooperative_compaction.py index 4d5295d8..d5849699 100644 --- a/tests/test_cooperative_compaction.py +++ b/tests/test_cooperative_compaction.py @@ -2098,6 +2098,77 @@ class TestPreHookUICompat: assert _coerce_event_id(None) is None assert _coerce_event_id("46") is None + def test_raising_on_error_still_emits_exactly_one_failed_end(self, session): + """A raising on_error (bounded listener queue.Full — the duck-typed + embedder class) must not void the exactly-one-end contract: the + guard in _compaction_bailed lets the end emit run, so no pane is + left holding a frozen progress bar.""" + ends: list[dict] = [] + + def _boom(_msg: str) -> None: + raise RuntimeError("listener queue full") + + session.ui = SimpleNamespace( + on_thinking_start=lambda: None, + on_thinking_stop=lambda: None, + on_error=_boom, + on_compaction=lambda payload: ( + ends.append(payload) if payload.get("phase") == "end" else None + ), + ) + result = session._compaction_bailed( + "error", "summarizer exploded", trigger="manual", my_generation=0 + ) + assert result is False # handled bail, no propagation + assert len(ends) == 1 + assert ends[0]["ok"] is False and ends[0]["reason"] == "error" + + def test_raising_lifecycle_hook_never_propagates(self, session): + """_compaction_event is the single raise-proofing site for every + lifecycle emission: a raising on_compaction degrades to a lost + render (marker id None), never to a propagating exception — a + raising failed-END re-created the frozen-bar hole one call deeper, + and a raising SUCCESS end after the committed swap made the + wrapper backstop fabricate a failed end for a compaction that + succeeded.""" + + def _boom(_payload: dict) -> int: + raise RuntimeError("hook broken") + + session.ui = SimpleNamespace( + on_thinking_start=lambda: None, + on_thinking_stop=lambda: None, + on_error=lambda _m: None, + on_compaction=_boom, + ) + # Success end: swallowed, marker id lost, nothing fabricated. + assert ( + session._compaction_event( + 0, {"phase": "end", "ok": True, "trigger": "manual", "summary": "s"} + ) + is None + ) + # Failed end via the bail path: still returns False, no propagation + # (pre-guard, the raise re-entered _compaction_bailed through the + # wrapper backstop and no end was ever emitted on either pass). + assert ( + session._compaction_bailed("error", "boom", trigger="manual", my_generation=0) is False + ) + # The duck-type fallback route is guarded by the same try: a + # raising on_info must not escape either. + session.ui = SimpleNamespace( + on_thinking_start=lambda: None, + on_thinking_stop=lambda: None, + on_error=lambda _m: None, + on_info=_boom, + ) + assert ( + session._compaction_event( + 0, {"phase": "end", "ok": True, "trigger": "manual", "summary": "s"} + ) + is None + ) + class TestCompactionNoticeStamp: """_compaction_event is the single display-policy site: failed ends diff --git a/tests/test_coordinator_page.py b/tests/test_coordinator_page.py index 4f202a72..8211d72d 100644 --- a/tests/test_coordinator_page.py +++ b/tests/test_coordinator_page.py @@ -687,7 +687,14 @@ def test_coordinator_js_gates_send_on_cross_user_busy(): assert "actingUserId !== me" in coord_js assert "composer.setSendBlocked(" in coord_js assert "function reconcileSendBlock()" in coord_js - # reactive 409 fallback + # reactive 409 fallback — the pane converts the 409 body at the fetch + # stage; the status ARM itself lives in the shared settle helper + # (composer_queue.settleSendResponse) with the rest of the response + # matrix, one implementation for both panes. assert "r.status === 409" in coord_js assert 'status: "cross_user_interjection"' in coord_js - assert 'data.status === "cross_user_interjection"' in coord_js + assert "settleSendResponse(" in coord_js + helper = ( + Path(__file__).resolve().parents[1] / "turnstone/shared_static/composer_queue.js" + ).read_text(encoding="utf-8") + assert 'status === "cross_user_interjection"' in helper diff --git a/tests/test_idle_nudge_watcher.py b/tests/test_idle_nudge_watcher.py index c853b632..4ab9602b 100644 --- a/tests/test_idle_nudge_watcher.py +++ b/tests/test_idle_nudge_watcher.py @@ -11,6 +11,7 @@ from __future__ import annotations import contextlib import logging import threading +from types import SimpleNamespace from typing import Any from unittest.mock import patch @@ -39,10 +40,17 @@ class _FakeWorkstream: self._worker_running = False self._closed = False self.worker_thread: Any = None - # Deferred /send entries — the wake gate yields while any are - # pending (order barrier); empty is the default every other test + # Deferred /send entries — the wake gate yields while the order + # barrier holds; empty/None is the default every other test # assumes. self._pending_sends: list[Any] = [] + self._pending_drain: Any = None + + def send_barrier_active(self) -> bool: + # Mirrors Workstream.send_barrier_active — the gate calls the + # METHOD, so the stub must carry the same two-term pair. + drain = self._pending_drain + return bool(self._pending_sends) or (drain is not None and drain.is_alive()) class _FakeManager: @@ -211,8 +219,18 @@ class TestWakeWorkstreamIfPending: with patch("turnstone.core.session_worker.send", return_value=True) as mock_send: assert wake_workstream_if_pending(ws, trigger="worker-exit") is False assert mock_send.call_count == 0 - # Barrier cleared (the drain retired) — the same call dispatches. + # CLAIMED-entry window: list empty but the drain is alive (an + # acked entry was popped, its dispatch in flight). The one-term + # list check let a wake jump the acknowledged send here — the + # barrier's drain-alive term must hold the yield. ws._pending_sends.clear() + ws._pending_drain = SimpleNamespace(is_alive=lambda: True) + with patch("turnstone.core.session_worker.send", return_value=True) as mock_send: + assert wake_workstream_if_pending(ws, trigger="worker-exit") is False + assert mock_send.call_count == 0 + # Barrier fully cleared (the drain retired) — the same call + # dispatches. + ws._pending_drain = None with patch("turnstone.core.session_worker.send", return_value=True) as mock_send: assert wake_workstream_if_pending(ws, trigger="drain-exit") is True assert mock_send.call_count == 1 diff --git a/tests/test_interactive_pane_js.py b/tests/test_interactive_pane_js.py index 849a561f..e94dd9b1 100644 --- a/tests/test_interactive_pane_js.py +++ b/tests/test_interactive_pane_js.py @@ -398,17 +398,21 @@ def test_pane_gates_send_on_cross_user_busy() -> None: # ...and drives the composer's hard block, re-run on every busy edge. assert "this.composer.setSendBlocked(" in body stripped = _strip_comments(body) - setbusy = stripped.index("setBusy(b) {") - assert "this._reconcileSendBlock();" in stripped[setbusy : setbusy + 600] + setbusy = stripped.index("setBusy(b, source) {") + assert "this._reconcileSendBlock();" in stripped[setbusy : setbusy + 800] def test_pane_handles_cross_user_409() -> None: """The reactive fallback: a 409 (button not yet disabled) surfaces a clean - message, not the generic 'Connection error' catch.""" + message, not the generic 'Connection error' catch. The pane converts + the 409 body at the fetch stage; the status ARM itself lives in the + shared settle helper (composer_queue.settleSendResponse) with the rest + of the response matrix.""" body = _INTERACTIVE.read_text(encoding="utf-8") assert "r.status === 409" in body assert 'status: "cross_user_interjection"' in body - assert 'data.status === "cross_user_interjection"' in body + helper = (_ROOT / "turnstone/shared_static/composer_queue.js").read_text(encoding="utf-8") + assert 'status === "cross_user_interjection"' in helper def test_sync_approval_state_prunes_orphan_cycles() -> None: @@ -713,16 +717,40 @@ def test_deferred_send_settle_protocol_pins() -> None: # so the SSE settle can beat the POST response's bind(): the controller # parks chip-absent settles and bind() reconciles them — without this a # raced chip stays flagged deferred and the idle sweep skips it forever. + # Expiry is TTL-based: a size cap evicted exactly this tab's raced + # settle when a window closed with a burst of deferred sends (ours + # parks FIRST, the foreign settles behind it overflow the cap). assert "_preBindSettles" in composer_queue assert "_preBindSettles.has(msgId)" in composer_queue - # Both panes pass the response through the options seam and consume the - # pane-tier settle event. + assert "PRE_BIND_SETTLE_TTL_MS" in composer_queue + assert "_preBindSettles.size" not in composer_queue, "size-cap eviction must stay dead" + # The full send-response settle matrix lives ONCE, in the shared + # helper — retro-convert (a parked, still-retractable message must not + # render as a sent bubble), the deferred busy-undo, and the queue_full + # idle-pane cleanup (bubble removed + busy restored: the refusal can + # now fire with no worker and no drain alive, so no state event would + # ever unstick the composer). + assert "export function settleSendResponse(queue, data, ctx)" in composer_queue + assert "!queuedEl && data.deferred" in composer_queue + assert "deferred: !!data.deferred" in composer_queue + assert "attachedCount: (data.attached_ids || []).length" in composer_queue + assert "ctx.busyIsOptimistic()" in composer_queue + assert composer_queue.count("ctx.optimisticEl.remove()") >= 2, ( + "both the retro-convert and queue_full arms must clear the optimistic bubble" + ) + # 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 + # flow's optimistic flip may ever be undone). for name, src in (("interactive.js", interactive), ("coordinator.js", coordinator)): - assert "deferred: !!data.deferred" in src, f"{name}: bind must carry the deferred flag" - assert "attachedCount: (data.attached_ids || []).length" in src, name + assert "settleSendResponse(" in src, f"{name}: settle matrix must be the shared helper" + assert "busyIsOptimistic" in src, name + 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 - # queuedEl-absent + deferred: the idle-thinking pane must render a - # real queued chip (retro-convert), not leave a sent-looking bubble - # for a parked, still-retractable message. - assert "!queuedEl && data.deferred" in src, f"{name}: idle-pane defer must chip" + # /command's degraded statuses are all surfaced: busy, running (the + # backstop answer), and error (503 — the worker never spawned; silence + # here reads as success). + assert 'body.status === "running"' in interactive + assert 'body.status === "error"' in interactive diff --git a/tests/test_server_attachments_endpoints.py b/tests/test_server_attachments_endpoints.py index 56657a0f..0ffa7602 100644 --- a/tests/test_server_attachments_endpoints.py +++ b/tests/test_server_attachments_endpoints.py @@ -509,6 +509,10 @@ class TestSendMessageAttachments: # pending list. ws._pending_sends = [] ws._pending_drain = None + # The /send route consults the order barrier as a METHOD — a bare + # Mock attr would be a truthy callable result and defer every send + # behind a phantom barrier (same class as the fields above). + ws.send_barrier_active = lambda: False ws._lock = threading.RLock() mgr.get.return_value = ws return captured, session @@ -712,6 +716,10 @@ class TestQueuedSendWithAttachments: # bare Mock attr would defer every send behind a phantom list. ws._pending_sends = [] ws._pending_drain = None + # The /send route consults the order barrier as a METHOD — a bare + # Mock attr would be a truthy callable result and defer every send + # behind a phantom barrier (same class as the fields above). + ws.send_barrier_active = lambda: False ws._lock = threading.RLock() mgr.get.return_value = ws return captured @@ -780,6 +788,10 @@ class TestBusyWorkerAttachments: # bare Mock attr would defer every send behind a phantom list. ws._pending_sends = [] ws._pending_drain = None + # The /send route consults the order barrier as a METHOD — a bare + # Mock attr would be a truthy callable result and defer every send + # behind a phantom barrier (same class as the fields above). + ws.send_barrier_active = lambda: False ws._lock = threading.RLock() mgr.get.return_value = ws return ws, session diff --git a/tests/test_server_authz.py b/tests/test_server_authz.py index 975e58e4..35619a43 100644 --- a/tests/test_server_authz.py +++ b/tests/test_server_authz.py @@ -1261,6 +1261,19 @@ class TestInteractiveEventsLifted: # --------------------------------------------------------------------------- +def test_command_backstop_sits_under_console_proxy_timeout() -> None: + """The /command ``running`` answer only ever traverses a proxied pane + while the node-side completion backstop is STRICTLY under the console + proxy's client timeout — two constants in two processes whose + inequality used to live in a comment. This is the enforcement: edit + either constant past the other and this fails before a proxied + deployment silently loses the degraded answer again.""" + from turnstone.console.server import _PROXY_CLIENT_TIMEOUT_S + from turnstone.server import _COMMAND_RESPONSE_BACKSTOP_S + + assert _COMMAND_RESPONSE_BACKSTOP_S < _PROXY_CLIENT_TIMEOUT_S + + class TestCompactCommandDispatch: """Manual /compact runs on the workstream's worker slot: the event loop stays free to stream the compaction progress events, and a concurrent @@ -2162,6 +2175,123 @@ 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.""" + 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"), + ) + for i in range(10): + r = client.post( + f"/v1/api/workstreams/{ws_id}/send", + json={"message": f"m{i}"}, + headers=_auth("user-1"), + ) + assert r.json()["status"] == "queued", i + r11 = client.post( + f"/v1/api/workstreams/{ws_id}/send", + json={"message": "one too many"}, + headers=_auth("user-1"), + ) + body = r11.json() + assert body["status"] == "queue_full" + assert "msg_id" not in body # never acknowledged + with ws._lock: + assert len(ws._pending_sends) == 10 + gate.set() + wait_until(lambda: len(ws.session.sends) == 10, timeout=8.0) + assert [s[0] for s in ws.session.sends] == [f"m{i}" for i in range(10)] + wait_until(lambda: self._drain_idle(ws), timeout=8.0) + + def test_drain_spawn_failure_rolls_back_and_answers_queue_full(self, app_client, monkeypatch): + """Thread.start failing at the drain-spawn site must not leave a + phantom acknowledged-nowhere entry (the 500-after-registration + class): entry and slot roll back under the same lock, the client + gets the retryable queue_full, and a retry once resources return + dispatches exactly once — no duplicate turns.""" + 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"), + ) + + class _ExhaustedThread(threading.Thread): + def start(self) -> None: + raise RuntimeError("can't start new thread") + + real_factory = session_routes._make_drain_thread + monkeypatch.setattr( + session_routes, + "_make_drain_thread", + lambda _ws: _ExhaustedThread(target=lambda: None, daemon=True), + ) + r = client.post( + f"/v1/api/workstreams/{ws_id}/send", + json={"message": "refused under exhaustion"}, + headers=_auth("user-1"), + ) + assert r.json()["status"] == "queue_full" + with ws._lock: + assert ws._pending_sends == [] # rolled back — no phantom + assert ws._pending_drain is None + # Resources recover: a plain retry defers and dispatches ONCE. + monkeypatch.setattr(session_routes, "_make_drain_thread", real_factory) + r2 = client.post( + f"/v1/api/workstreams/{ws_id}/send", + json={"message": "the retry"}, + headers=_auth("user-1"), + ) + assert r2.json()["status"] == "queued" + assert r2.json()["deferred"] is True + gate.set() + wait_until(lambda: ws.session.sends, timeout=8.0) + wait_until(lambda: self._drain_idle(ws), timeout=8.0) + assert [s[0] for s in ws.session.sends] == ["the retry"] + + def test_command_spawn_failure_answers_503_not_ok(self, app_client, monkeypatch): + """A command worker that never spawned must not answer 200 ok — the + endpoint's generic catch-all used to swallow the dispatcher's + Thread.start re-raise, telling SDK callers their /clear ran while + the context stayed un-cleared.""" + from turnstone.core import session_worker + + client, mgr = app_client + ws_id = self._create_ws(client) + ws = mgr.get(ws_id) + assert ws is not None + + def _exhausted(*_a, **_kw): + raise RuntimeError("can't start new thread") + + monkeypatch.setattr(session_worker, "send", _exhausted) + resp = client.post( + "/v1/api/command", + json={"command": "/clear", "ws_id": ws_id}, + headers=_auth("user-1"), + ) + assert resp.status_code == 503 + assert resp.json()["status"] == "error" + assert ws.session.commands == [] # the command never ran + 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/tests/test_session_worker.py b/tests/test_session_worker.py index 5fc69309..ff001796 100644 --- a/tests/test_session_worker.py +++ b/tests/test_session_worker.py @@ -146,6 +146,29 @@ def test_enqueue_unexpected_exception_returns_false_logged() -> None: assert ws._worker_running is True +def test_send_barrier_active_truth_table() -> None: + """The two-term order barrier (Workstream.send_barrier_active): list + non-empty OR drain alive. The drain-alive term covers the CLAIMED- + entry window (entry popped, dispatch in flight) — the _PendingSend + invariant "drain not alive ⇒ nothing claimed" is what makes the pair + exhaustive; a one-term copy at any consumer re-opens the wake-jumps- + an-acked-send hole.""" + from types import SimpleNamespace + + ws = _make_ws() + assert ws.send_barrier_active() is False # empty list, no drain + ws._pending_sends.append(object()) # type: ignore[arg-type] + assert ws.send_barrier_active() is True # list term + ws._pending_drain = SimpleNamespace(is_alive=lambda: True) # type: ignore[assignment] + assert ws.send_barrier_active() is True # both terms + ws._pending_sends.clear() + assert ws.send_barrier_active() is True # drain-alive term alone (claimed window) + ws._pending_drain = SimpleNamespace(is_alive=lambda: False) # type: ignore[assignment] + assert ws.send_barrier_active() is False # dead drain, empty list + ws._pending_drain = None + assert ws.send_barrier_active() is False + + def test_spawn_failure_releases_slot_and_reraises(monkeypatch) -> None: """``Thread.start`` raising (thread exhaustion, MemoryError) must not wedge the slot: the claim ``(worker_thread, _worker_running)`` taken diff --git a/turnstone/api/server_spec.py b/turnstone/api/server_spec.py index 7d121ddc..8df956ae 100644 --- a/turnstone/api/server_spec.py +++ b/turnstone/api/server_spec.py @@ -136,7 +136,10 @@ SERVER_ENDPOINTS: list[EndpointSpec] = [ "Execute a slash command", request_model=CommandRequest, response_model=StatusResponse, - error_codes=[400, 404, 409], + # 409 = worker slot busy (deliberate loud refusal); 503 = the + # command worker could not be started (thread exhaustion — retry + # shortly; the command did NOT run). + error_codes=[400, 404, 409, 503], tags=["Chat"], ), EndpointSpec( diff --git a/turnstone/cli.py b/turnstone/cli.py index 800f5a80..82f89cc3 100644 --- a/turnstone/cli.py +++ b/turnstone/cli.py @@ -17,7 +17,6 @@ import threading from typing import TYPE_CHECKING, Any from turnstone.core.adapters.interactive_adapter import InteractiveAdapter -from turnstone.core.compaction_render import render_compaction_event_as_info from turnstone.core.judge import JudgeConfig from turnstone.core.session import ChatSession, SessionUI from turnstone.core.session_manager import SessionManager @@ -329,17 +328,12 @@ class TerminalUI(SessionUI): sys.stdout.write(f"{YELLOW}[{label}]{RESET} {content}\n") sys.stdout.flush() - def on_compaction(self, payload: dict[str, Any]) -> int | None: - """Render compaction lifecycle events as the terminal's classic text - lines — the notice, ``part k/N`` progress, and the token-delta + - boxed-summary result the CLI printed before these became structured - events (the web UI renders the same payloads as a progress card). - Shared with :meth:`ChatSession._compaction_event`'s duck-typed - fallback so both render identically; failed-end suppression rides - the emitter-stamped ``notice`` field (single policy site). - """ - render_compaction_event_as_info(payload, self.on_info) - return None + # No on_compaction override: TerminalUI subclasses SessionUI + # explicitly, and the inherited protocol default body IS the + # terminal's classic rendering (render_compaction_event_as_info via + # on_info — see SessionUI.on_compaction). A byte-identical override + # here silently forked the policy site: a future change to the + # default stopped applying to the CLI. def on_state_change(self, state: str) -> None: pass # base TerminalUI ignores state changes diff --git a/turnstone/console/coordinator_adapter.py b/turnstone/console/coordinator_adapter.py index e3bcd5ad..f5fc328a 100644 --- a/turnstone/console/coordinator_adapter.py +++ b/turnstone/console/coordinator_adapter.py @@ -393,17 +393,17 @@ class CoordinatorAdapter: # Order-barrier yield, mirroring the /send route's pre-check: once # deferred sends are pending (or a claimed entry's dispatch is in - # flight — the drain-alive term), a spawn here would overtake - # messages already acknowledged "queued". Refuse via the return - # value — the adapter's documented backpressure surface, which the - # sole call site reports as queue_full/undelivered — NEVER by - # raising queue.Full from the body: only enqueue closures may (the - # dispatcher catches it there); a body-raise would escape into the - # create handler as a crash. Unreachable today (that caller - # dispatches on a freshly created workstream, which cannot have - # pending sends); the guard exists for future dispatch callers. - drain_t = ws._pending_drain - if ws._pending_sends or (drain_t is not None and drain_t.is_alive()): + # flight — the drain-alive term the shared predicate carries), a + # spawn here would overtake messages already acknowledged + # "queued". Refuse via the return value — the adapter's + # documented backpressure surface, which the sole call site + # reports as queue_full/undelivered — NEVER by raising queue.Full + # from the body: only enqueue closures may (the dispatcher + # catches it there); a body-raise would escape into the create + # handler as a crash. Unreachable today (that caller dispatches + # on a freshly created workstream, which cannot have pending + # sends); the guard exists for future dispatch callers. + if ws.send_barrier_active(): log.warning( "coord_adapter.send_refused_pending_sends ws=%s count=%d", ws.id[:8], diff --git a/turnstone/console/server.py b/turnstone/console/server.py index 7fddfbde..4139d4cb 100644 --- a/turnstone/console/server.py +++ b/turnstone/console/server.py @@ -588,6 +588,14 @@ _CONSOLE_PROXY_STYLE = ( _VALID_NODE_ID = re.compile(r"^[a-zA-Z0-9._-]+$") _VALID_WS_ID_RE = re.compile(r"^[a-f0-9]{1,64}$") +# Client timeout for the REST proxy pool (BOTH constructions: startup and +# the mTLS re-create). Node endpoints that answer degraded-but-in-time +# responses size their backstops strictly UNDER this bound — e.g. the +# quick-command ``running`` answer (turnstone/server.py +# _COMMAND_RESPONSE_BACKSTOP_S); a test pins the inequality. The SSE +# proxy client's granular Timeout is a separate contract. +_PROXY_CLIENT_TIMEOUT_S = 30 + _PROXY_JWT_EXPIRY_SECONDS = 300 # 5 min — ample for any request round-trip @@ -5322,7 +5330,7 @@ async def _lifespan(app: Starlette) -> AsyncGenerator[None, None]: _proxy_verify: Any = _proxy_ssl if _proxy_ssl else True app.state.proxy_client = httpx.AsyncClient( - timeout=30, + timeout=_PROXY_CLIENT_TIMEOUT_S, limits=httpx.Limits( max_connections=fan_out + 50, max_keepalive_connections=min(fan_out // 4, 100), @@ -5464,7 +5472,7 @@ async def _lifespan(app: Starlette) -> AsyncGenerator[None, None]: await app.state.proxy_client.aclose() await app.state.proxy_sse_client.aclose() app.state.proxy_client = httpx.AsyncClient( - timeout=30, + timeout=_PROXY_CLIENT_TIMEOUT_S, limits=httpx.Limits( max_connections=app.state.fan_out_limit + 50, max_keepalive_connections=min(app.state.fan_out_limit // 4, 100), diff --git a/turnstone/console/static/coordinator/coordinator.js b/turnstone/console/static/coordinator/coordinator.js index 1911f25a..6974827e 100644 --- a/turnstone/console/static/coordinator/coordinator.js +++ b/turnstone/console/static/coordinator/coordinator.js @@ -44,6 +44,11 @@ import { indexLabel, } from "/shared/conversation.js"; import { redactCredentials } from "/shared/redact_credentials.js"; +import { + createQueueController, + parsePriority, + settleSendResponse, +} from "/shared/composer_queue.js"; import { OVERFLOW_TRIP_COUNT, OVERFLOW_TRIP_WINDOW_MS, @@ -304,6 +309,9 @@ function createCoordinatorPane(root, wsId, opts) { }, }); let busy = false; + // Provenance of the current busy=true (see setBusy): "server" | + // "optimistic" | null when idle. + let busySource = null; // Acting user (turn initiator) of the in-flight turn, from state_change // events; drives the shared-workstream cross-user send gate. Carries the // owner id even single-user (the gate just no-ops — it equals this viewer); @@ -1871,8 +1879,15 @@ function createCoordinatorPane(root, wsId, opts) { // caller relies on. queue.onIdleEdge runs only on the actual edge // (it's the heavier work — querySelectorAll-driven promote sweep // plus the cancel-timer cleanup wired via the onIdle hook above). - function setBusy(b) { + function setBusy(b, source) { const next = !!b; + // Who asserted busy: "server" (default — state events and every + // existing/future writer) or "optimistic" (ONLY coordSend's pre-POST + // flip). The deferred/queue_full settle arms may clear busy solely + // while it is still this send's own optimistic flip — a server- + // stamped busy is a real turn and must never be clobbered. + // Centralized HERE so an unstamped future writer fails safe. + busySource = next ? source || "server" : null; composer.setBusy(next); // Greys out the per-message edit/rewind/retry buttons while a generation // is in flight (CSS: [data-busy="true"] .msg-action-btn). @@ -1957,20 +1972,18 @@ function createCoordinatorPane(root, wsId, opts) { let queuedEl = null; let optimisticEl = null; - // Server re-parses the !!! prefix to set queue priority — the - // optimistic bubble strips it for display. Parsed outside the busy - // branch: the queued+deferred response arm needs it too (an idle - // pane's send can defer behind a command window / pending list). - let displayText = trimmed; - let priority = "notice"; - if (trimmed.startsWith("!!!")) { - displayText = trimmed.slice(3).trimStart(); - priority = "important"; - } - if (busy) { + const isBusy = busy; + // Display-only strip of the !!! prefix (the server re-parses it + // authoritatively); shared parse so the settle helper's retro-convert + // renders the same chip either pane would have built pre-POST. + const { displayText, priority } = parsePriority(trimmed); + if (isBusy) { queuedEl = queue.addQueuedMessage(displayText, priority); } else { - setBusy(true); + // "optimistic": no server state event asserted this — the settle + // arms may undo it if the send turns out deferred/refused (see + // setBusy's busySource contract). + setBusy(true, "optimistic"); // snap.attachments carries the chip metadata (kind + filename) // for every stable chip the composer holds; pass it through so // the optimistic user bubble shows the same pill cluster the @@ -2051,71 +2064,22 @@ function createCoordinatorPane(root, wsId, opts) { return r.json(); }) .then((data) => { - if (data && data.status === "queued" && data.msg_id) { - // queuedEl-present: bind() handles the known races. queuedEl- - // absent + deferred: the pane thought it was idle but the send - // parked on the server's deferred list (command window / order - // barrier) — a plain sent bubble would present a parked, - // still-retractable, restart-droppable message as delivered, so - // replace the optimistic bubble with a real queued chip. - // queuedEl-absent + undeferred: plain interjection into a live - // turn the client hadn't seen yet (SSE state_change race); keep - // the historical small-UX-gap behavior (busy flip only). - if (!queuedEl && data.deferred) { - if (optimisticEl && optimisticEl.isConnected) optimisticEl.remove(); - queuedEl = queue.addQueuedMessage(displayText, priority); - } - if (queuedEl) { - queue.bind(queuedEl, data.msg_id, { - deferred: !!data.deferred, - attachedCount: (data.attached_ids || []).length, - }); - } else setBusy(true); - attachments.consume(data.attached_ids, data.dropped_attachment_ids); - } else if (data && data.status === "busy") { - if (queuedEl) queue.remove(queuedEl); - appendText("error", "Server is busy. Please wait.", { - label: "error", - }); - if (!queuedEl) setBusy(false); - } else if (data && data.status === "queue_full") { - if (queuedEl) queue.remove(queuedEl); - appendText("error", "Message queue full. Please wait.", { - label: "error", - }); - } else if (data && data.status === "attachments_busy") { - // Attachments can't ride a queued user turn — server held - // the reservations long enough to bounce the request and - // released them. Chips stay in the composer; user retries - // once the assistant finishes. - if (queuedEl) queue.remove(queuedEl); - appendText( - "error", - "Attachments can't be sent while the assistant is working. Send a text-only message now, or wait and resend with attachments.", - { label: "error" }, - ); - } else if (data && data.status === "cross_user_interjection") { - // Another participant's turn is in flight; the server refused the - // interjection so it can't run under their credentials or be - // misattributed. Reactive fallback for the click-beats-event race - // (the send gate normally disables the button first). - if (queuedEl) queue.remove(queuedEl); - appendText( - "error", - data.error || - "Another participant's turn is in progress. Wait for it to finish, then send your message.", - { label: "error" }, - ); - if (!queuedEl) setBusy(false); - } else { - // Unknown / "ok" status (stale-busy race): settle the optimistic - // bubble so a pre-bind × can't strand it in the dismissing state. - if (queuedEl) queue.promote(queuedEl); - attachments.consume( - data && data.attached_ids, - data && data.dropped_attachment_ids, - ); - } + // The full status dispatch (queued/retro-convert, busy, + // queue_full, attachments_busy, cross_user, unknown-ok) lives in + // the shared helper — ONE settle matrix for both panes; see + // settleSendResponse's contract for the arm semantics. + settleSendResponse(queue, data || {}, { + queuedEl, + optimisticEl, + isBusy, + displayText, + priority, + setBusy: (b) => setBusy(b), + busyIsOptimistic: () => busy && busySource === "optimistic", + renderError: (msg) => appendText("error", msg, { label: "error" }), + consumeAttachments: (attached, droppedIds) => + attachments.consume(attached, droppedIds), + }); }) .catch((e) => { if (queuedEl) queue.remove(queuedEl); diff --git a/turnstone/core/idle_nudge_watcher.py b/turnstone/core/idle_nudge_watcher.py index d0a95bad..b56eba2f 100644 --- a/turnstone/core/idle_nudge_watcher.py +++ b/turnstone/core/idle_nudge_watcher.py @@ -75,10 +75,12 @@ def wake_workstream_if_pending(ws: Workstream, *, trigger: str = "unspecified") queue at its own seams (``ATTENTION``/``THINKING``/``RUNNING`` all imply a live worker), and ``ERROR`` stays parked for the operator rather than burning inference unattended. - * ``ws._pending_sends`` non-empty — deferred sends hold the order - barrier; the wake yields and is re-armed by the deferred turns' + * ``ws.send_barrier_active()`` — deferred sends hold the order + barrier (pending entries, or a claimed entry's dispatch in + flight); the wake yields and is re-armed by the deferred turns' exit backstops (or the drain's clean exit when everything was - retracted). See the inline comment for the staleness argument. + retracted). See the predicate's docstring for the staleness + argument. * nothing gate-eligible under ``WAKE_PENDING`` — tool-only/quiet entries belong to the next tool-result seam, not a synthetic empty user turn (``deliver_wake_nudge_from_queue`` would no-op on them). @@ -105,19 +107,21 @@ def wake_workstream_if_pending(ws: Workstream, *, trigger: str = "unspecified") session = ws.session if session is None or ws._closed or ws.state is not WorkstreamState.IDLE: return False - if ws._pending_sends: - # Order-barrier yield: deferred sends (acknowledged "queued" - # during a command window — see _PendingSend) are older than any - # nudge, and a wake worker claiming the slot would push them - # behind its whole turn. Lockless peek, benign both ways: a - # stale non-empty skips once more (the next exit backstop - # converges), a stale empty means the concurrent defer holds no - # order contract against this wake anyway. Convergence is - # structural — every path that clears the barrier re-runs this - # gate: each deferred turn's exit via ``_retry_pending_wake``, - # and the drain's own clean exit (trigger="drain-exit"), which - # covers a list that empties by pure retraction and so never - # runs a turn. + if ws.send_barrier_active(): + # Order-barrier yield: deferred sends (acknowledged "queued" — + # see _PendingSend) are older than any nudge, and a wake worker + # claiming the slot would push them behind its whole turn. The + # shared predicate carries BOTH terms — the pending list AND the + # drain-alive clause covering a CLAIMED entry (popped, dispatch + # in flight but not yet holding the slot); checking the list + # alone let a wake jump an acknowledged send in exactly that + # window. Lockless call, benign both ways (staleness ruling in + # the predicate's docstring). Convergence is structural — + # every path that clears the barrier re-runs this gate: each + # deferred turn's exit via ``_retry_pending_wake``, and the + # drain's own clean exit (trigger="drain-exit"), which covers a + # list that empties by pure retraction and so never runs a + # turn. log.info("nudge_wake.yielded_to_pending_sends ws=%s trigger=%s", ws.id[:8], trigger) return False nudge_queue = getattr(session, "_nudge_queue", None) diff --git a/turnstone/core/session.py b/turnstone/core/session.py index 4fb53bbc..5d69fd9b 100644 --- a/turnstone/core/session.py +++ b/turnstone/core/session.py @@ -7861,27 +7861,43 @@ class ChatSession: # converge on render_compaction_event_as_info, policy # single-sited. emit = getattr(self.ui, "on_compaction", None) - if emit is None: - # Pre-hook UIs get the classic info lines back (an - # auto-compaction must never swap history with zero - # announcement — the pre-1.8 lines reached every UI - # unconditionally). Invoked for superseded events too: a - # superseded OK end announces a swap that really committed, - # and failed-end staleness is already encoded in ``notice``. - # ``on_info`` is getattr-guarded like the hook itself — the - # never-crash property is the floor; a UI with neither hook - # keeps compacting silently. Deliberately NOT dual-emitted - # for hook-aware UIs or SSE: pre-1.8 SSE/SDK clients that - # ignore unknown `compaction` events lose these lines — a - # documented 1.8 breaking change (CHANGELOG); dual emission - # would double-render on every current client. - info = getattr(self.ui, "on_info", None) - if info is not None: - from turnstone.core.compaction_render import render_compaction_event_as_info + try: + if emit is None: + # Pre-hook UIs get the classic info lines back (an + # auto-compaction must never swap history with zero + # announcement — the pre-1.8 lines reached every UI + # unconditionally). Invoked for superseded events too: a + # superseded OK end announces a swap that really committed, + # and failed-end staleness is already encoded in ``notice``. + # ``on_info`` is getattr-guarded like the hook itself — the + # never-crash property is the floor; a UI with neither hook + # keeps compacting silently. Deliberately NOT dual-emitted + # for hook-aware UIs or SSE: pre-1.8 SSE/SDK clients that + # ignore unknown `compaction` events lose these lines — a + # documented 1.8 breaking change (CHANGELOG); dual emission + # would double-render on every current client. + info = getattr(self.ui, "on_info", None) + if info is not None: + from turnstone.core.compaction_render import render_compaction_event_as_info - render_compaction_event_as_info(event, info) + render_compaction_event_as_info(event, info) + return None + result = emit(event) + except Exception: + # The single raise-proofing site for EVERY lifecycle emission + # (both compat routes, all phases): a raising duck-typed hook + # must degrade to a lost render, never to a lost EVENT — + # unguarded, a raising failed-END emit voided the + # exactly-one-end contract through the wrapper backstop + # (frozen progress bar on every pane), and a raising SUCCESS + # end after the committed swap made the backstop fabricate a + # failed end + red row for a compaction that succeeded. The + # cost on raise is only the marker-stamp id — the receiving + # hook was broken anyway. Same policy as this method's own + # getattr guard ("must not wedge every long session") and the + # same shape as _emit_send_ui. + log.debug("compaction lifecycle hook raised; event dropped for this UI", exc_info=True) return None - result = emit(event) # Duck-typed hooks aren't bound to the protocol's return type; the # marker-stamp consumer needs int-or-None, nothing else (and a # hook returning True must not stamp a bool — see _coerce_event_id). @@ -7914,7 +7930,17 @@ class ChatSession: Handled bails never propagate, so they always emit. """ if reason == "error" and emit_error: - self.ui.on_error(message) + try: + # Guarded like _record_fatal_error's on_error: a raising + # duck-typed hook (bounded listener queue.Full) must not + # escape BEFORE the end emit below — that voided the + # exactly-one-end contract on both bail passes and left + # every pane a frozen progress bar. Order kept + # (on_error first): the panes render the red row from it + # and treat the end event as card-teardown only. + self.ui.on_error(message) + except Exception: + log.debug("ui.on_error failed during compaction bail", exc_info=True) self._compaction_event( my_generation, {"phase": "end", "ok": False, "reason": reason, "message": message, "trigger": trigger}, diff --git a/turnstone/core/session_routes.py b/turnstone/core/session_routes.py index 30b340b4..c12369da 100644 --- a/turnstone/core/session_routes.py +++ b/turnstone/core/session_routes.py @@ -40,8 +40,11 @@ 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 if TYPE_CHECKING: + import threading + from starlette.background import BackgroundTask from starlette.requests import Request from starlette.responses import Response @@ -3948,54 +3951,40 @@ def make_detail_handler(cfg: SessionEndpointConfig) -> Handler: # --------------------------------------------------------------------------- -# Deferred sends (command windows) +# 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. -@dataclass -class _PendingSend: - """One send deferred during a command window. +# 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 - A /send that lands while a slash-command worker holds the slot - (``ws.worker_kind == "command"`` — a manual /compact can hold it for - minutes) is answered ``{"status": "queued", "msg_id": ...}`` - immediately and registered here; :func:`_drain_pending_sends` - dispatches it full-fidelity when the window closes. This replaces - the parked-POST design, which encoded "client disconnected" as - "message retracted" — true only for the web composers' ✕-abort; - every bounded caller (the coordinator client and the console proxy - at timeout=30, SDKs, curl) times out instead, and its message was - deliberately dropped for the whole window. - ``attempt`` is the prebuilt one-session-capture dispatch closure - (:func:`_make_dispatch_attempt` with ``defer_fidelity=True``), so - the drain stays endpoint-agnostic — everything kind-specific - (attachments, spawn metrics, UI hooks) was captured at defer time. - ``retracted`` is flipped under ``ws._lock`` by the DELETE dequeue - fall-through; the drain never dispatches a retracted entry. +def _make_drain_thread(ws: Workstream) -> threading.Thread: + """Construct (never start) the pending-send drain thread for *ws*. - Durability contract (documented in the API reference): node-local - and in-memory, the interjection queue's lifetime — entries die with - the workstream or the process, so "queued" is at-most-once intake, - not durable acceptance. - - Invariant both the route's order barrier and the drain depend on: - **drain not alive ⇒ nothing claimed.** The drain pops an entry only - while it lives and re-inserts it at head on ANY non-dispatch outcome - (rejection or crash), so a dead/absent drain means every accepted - entry is on this list — the route's barrier term pair - (``_pending_sends`` non-empty OR drain alive) therefore covers the - claimed-entry window with no third state. - - (No ``priority`` field: dispatch is strictly FIFO — arrival order is - the contract — and the queued response/event use the route's parsed - local. A deferred entry's ``!!!`` prefix still reaches the model: - the full text dispatches as an ordinary send.) + A module-level seam so tests can inject spawn failure without + touching the global ``threading`` module; ``_defer_send`` owns the + slot write, the ``start()`` call, and the rollback discipline. """ + import threading # matches the file's handler-scope import style - msg_id: str - attempt: Callable[[ChatSession], tuple[bool, dict[str, Any]]] - retracted: bool = False + return threading.Thread( + target=_drain_pending_sends, + args=(ws,), + name=f"pending-drain-{ws.id[:8]}", + daemon=True, + ) def _emit_send_ui(ws: Workstream, ui: Any, hook_name: str, *args: Any) -> None: @@ -4394,15 +4383,19 @@ def make_send_handler(cfg: SessionEndpointConfig) -> Handler: emits the pane-tier ``message_dispatched`` settle event (see :func:`_make_dispatch_attempt`). - 200 ``{"status": "queue_full", "attached_ids", - "dropped_attachment_ids"}`` — live worker's queue at - capacity; reservations released. Caller should retry. The - ``attached_ids`` list is always empty here (the dispatch - didn't take ownership of any reservations). + "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 + 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 + always empty here (the dispatch didn't take ownership of any + reservations). - 4xx / 500 — auth / not-found / no-session per the usual :class:`SessionEndpointConfig` semantics. """ import asyncio - import threading import uuid from turnstone.core.tool_advisory import parse_priority @@ -4529,59 +4522,131 @@ def make_send_handler(cfg: SessionEndpointConfig) -> Handler: # command-window trigger re-acquires for its append, which is safe # because that rejection was reported under the lock the attempt # itself held. - def _defer_send(*, require_barrier: bool) -> Response | None: - # ``send_id`` is minted only when attachments are enabled — - # the deferred entry needs a truthy id regardless: it is the - # client's dismiss/bind handle and the drain's - # queue_msg_id/send_id thread. - pending_msg_id = send_id or uuid.uuid4().hex - cleaned_display, pending_priority = parse_priority(message) - entry = _PendingSend( - msg_id=pending_msg_id, - attempt=_make_dispatch_attempt( - ws, - cfg, - ui, - message=message, - resolved_atts=resolved_atts, - ordered_taken=ordered_taken, - send_id=pending_msg_id, - acting_uid=acting_uid, - request=None, - defer_fidelity=True, - ), + def _queue_full_response() -> Response: + # Shared refusal shape (see the not-ok arm below for the + # rationale): retry-shortly semantics, no ownership taken. + return JSONResponse( + { + "status": "queue_full", + "attached_ids": [], + "dropped_attachment_ids": list(requested_ids), + } ) + + def _defer_send(*, require_barrier: bool) -> Response | None: with ws._lock: - if require_barrier: - drain_t = ws._pending_drain - if not ws._pending_sends and not (drain_t is not None and drain_t.is_alive()): - return None # no barrier — caller dispatches directly + # Probe FIRST, before constructing anything: in the + # overwhelmingly common no-barrier case this costs two + # field reads instead of a discarded closure tree + + # parse_priority per ordinary send. + if require_barrier and not ws.send_barrier_active(): + return None # no barrier — caller dispatches directly if ws._closed: # Mirror the dispatch-refusal 404 below: a "queued" # 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. + # len() deliberately counts retract-marked husks + # awaiting the drain's loop-top purge (DELETE only + # marks): a live-only count would let park/retract + # churn re-open the unbounded-growth hole. Transient + # over-refusal self-heals at the next purge. + return _queue_full_response() + # ``send_id`` is minted only when attachments are enabled + # — the deferred entry needs a truthy id regardless: it + # is the client's dismiss/bind handle and the drain's + # queue_msg_id/send_id thread. Constructed INSIDE the + # lock: closure creation is microsecond-cheap (the same + # argument session_worker.send makes for Thread()), and + # it keeps probe→append atomic. + pending_msg_id = send_id or uuid.uuid4().hex + cleaned_display, pending_priority = parse_priority(message) + entry = _PendingSend( + msg_id=pending_msg_id, + attempt=_make_dispatch_attempt( + ws, + cfg, + ui, + message=message, + resolved_atts=resolved_atts, + ordered_taken=ordered_taken, + send_id=pending_msg_id, + acting_uid=acting_uid, + request=None, + defer_fidelity=True, + ), + ) ws._pending_sends.append(entry) drain = ws._pending_drain if drain is None or not drain.is_alive(): - # Single drain-spawn site — also the recovery path for - # a drain that died in its last-resort handler. - t = threading.Thread( - target=_drain_pending_sends, - args=(ws,), - name=f"pending-drain-{ws.id[:8]}", - daemon=True, - ) + # Single drain-spawn site — also the recovery path + # for a drain that died in its last-resort handler. + # ``t.start()`` stays INSIDE this lock acquisition, + # deliberately unlike session_worker.send's + # outside-lock start (d3028234): that site's readers + # gate on the ``_worker_running`` flag, which is + # valid before start — this slot's only liveness + # signal is ``Thread.is_alive()``, which reads False + # for a constructed-but-unstarted thread, so an + # outside-lock start would let a concurrent defer's + # eligibility check see the pending drain as dead + # and spawn a SECOND dispatcher (FIFO inversion; the + # drain's exit slot-clears are single-flight-only). + # Holding the lock through start makes that state + # unobservable and keeps single-flight structural; + # the cost is thread-spawn latency (~100µs) on a + # cold path. + t = _make_drain_thread(ws) ws._pending_drain = t - t.start() - if cfg.emit_message_queued and hasattr(ui, "_enqueue"): - ui._enqueue( + try: + t.start() + except Exception: + # Thread creation failed (exhaustion, + # MemoryError). Roll back BOTH writes — the + # lock was held throughout, so the entry is + # provably the tail and the slot is provably + # ``t`` — and refuse with queue_full: the SDK's + # existing retry-shortly vocabulary. Never a + # 500 after registration (a phantom entry the + # client can't retract that dispatches later as + # a duplicate), and never a queued ack (it would + # promise a dispatch whose only revival trigger + # is a FUTURE send). Entries acked by earlier + # successful defers stay parked under the + # next-send-re-ensures policy — no respawn + # attempt here under the same exhaustion that + # just failed. + ws._pending_sends.pop() + ws._pending_drain = None + log.exception( + "ws.send.pending_drain_spawn_failed ws=%s — send refused", + ws.id[:8], + ) + return _queue_full_response() + # Best-effort ack event — a raising UI hook must not convert + # an ACCEPTED deferred send into a 500 (the client would + # retry and deliver twice); same never-mask-acceptance rule + # as message_dispatched. + if cfg.emit_message_queued: + _emit_send_ui( + ws, + ui, + "_enqueue", { "type": "message_queued", "message": cleaned_display, "priority": pending_priority, "msg_id": pending_msg_id, - } + }, ) return JSONResponse( { @@ -4632,7 +4697,9 @@ 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. ``attached_ids`` + # queue_full so clients retry rather than 500 (the same shape + # _defer_send answers for its saturation cap and drain-spawn + # failure). ``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. @@ -4677,15 +4744,21 @@ def make_send_handler(cfg: SessionEndpointConfig) -> Handler: dropped = [aid for aid in requested_ids if aid not in taken_set] if queue_outcome: - # Reused a live worker; ``queue_message`` succeeded. - if cfg.emit_message_queued and hasattr(ui, "_enqueue"): - ui._enqueue( + # Reused a live worker; ``queue_message`` succeeded. Best- + # effort like the defer arm's ack: the message is already + # accepted, so a raising UI hook must not 500 this into a + # client retry (duplicate delivery). + if cfg.emit_message_queued: + _emit_send_ui( + ws, + ui, + "_enqueue", { "type": "message_queued", "message": queue_outcome["cleaned"], "priority": queue_outcome["priority"], "msg_id": queue_outcome["msg_id"], - } + }, ) return JSONResponse( { diff --git a/turnstone/core/session_worker.py b/turnstone/core/session_worker.py index 3bc32ff1..76dbbf2b 100644 --- a/turnstone/core/session_worker.py +++ b/turnstone/core/session_worker.py @@ -47,7 +47,7 @@ from turnstone.core.log import get_logger if TYPE_CHECKING: from collections.abc import Callable - from turnstone.core.workstream import Workstream + from turnstone.core.workstream import WorkerKind, Workstream log = get_logger(__name__) @@ -93,7 +93,7 @@ def send( enqueue: Callable[[], None], run: Callable[[], None], thread_name: str | None = None, - worker_kind: str = "turn", + worker_kind: WorkerKind = "turn", ) -> bool: """Dispatch work onto a workstream's worker thread. diff --git a/turnstone/core/workstream.py b/turnstone/core/workstream.py index 2f55ebf9..904b9e66 100644 --- a/turnstone/core/workstream.py +++ b/turnstone/core/workstream.py @@ -13,11 +13,19 @@ import threading import time import uuid from dataclasses import dataclass, field -from typing import TYPE_CHECKING, Any +from typing import TYPE_CHECKING, Any, Literal if TYPE_CHECKING: + from collections.abc import Callable + from turnstone.core.session import ChatSession, SessionUI +# What KIND of work a worker slot holds. A Literal (not a free-form str) +# so a typo'd comparison or a new dispatch caller passing "commands" is a +# type error instead of a silently never-firing command-window guard — +# the /send defer keys on exactly these values. +WorkerKind = Literal["", "turn", "command"] + # --------------------------------------------------------------------------- # Kind enum — single source of truth for the workstream dispatch classifier @@ -96,6 +104,61 @@ BULK_CLOSE_STATE_VALUES: frozenset[str] = frozenset( ) +# --------------------------------------------------------------------------- +# Deferred sends +# --------------------------------------------------------------------------- + + +@dataclass +class _PendingSend: + """One send deferred while the /send order barrier holds. + + A /send that lands while a slash-command worker holds the slot + (``ws.worker_kind == "command"`` — a manual /compact can hold it for + minutes) or while earlier deferred entries are still pending is + answered ``{"status": "queued", "deferred": true, "msg_id": ...}`` + immediately and registered here; + :func:`turnstone.core.session_routes._drain_pending_sends` dispatches + it full-fidelity when the slot frees. This replaces the parked-POST + design, which encoded "client disconnected" as "message retracted" — + true only for the web composers' ✕-abort; every bounded caller (the + coordinator client and the console proxy at timeout=30, SDKs, curl) + times out instead, and its message was deliberately dropped for the + whole window. + + ``attempt`` is the prebuilt one-session-capture dispatch closure + (:func:`turnstone.core.session_routes._make_dispatch_attempt` with + ``defer_fidelity=True``), so the drain stays endpoint-agnostic — + everything kind-specific (attachments, spawn metrics, UI hooks) was + captured at defer time. ``retracted`` is flipped under ``ws._lock`` + by the DELETE dequeue fall-through; the drain never dispatches a + retracted entry. + + Durability contract (documented in the API reference): node-local + and in-memory, the interjection queue's lifetime — entries die with + the workstream or the process, so "queued" is at-most-once intake, + not durable acceptance. + + Invariant both the route's order barrier and the drain depend on + (and the second term of :meth:`Workstream.send_barrier_active` + exists to honor): **drain not alive ⇒ nothing claimed.** The drain + pops an entry only while it lives and re-inserts it at head on ANY + non-dispatch outcome (rejection or crash), so a dead/absent drain + means every accepted entry is on the list — the barrier term pair + (list non-empty OR drain alive) therefore covers the claimed-entry + window with no third state. + + (No ``priority`` field: dispatch is strictly FIFO — arrival order is + the contract — and the queued response/event use the route's parsed + local. A deferred entry's ``!!!`` prefix still reaches the model: + the full text dispatches as an ordinary send.) + """ + + msg_id: str + attempt: Callable[[ChatSession], tuple[bool, dict[str, Any]]] + retracted: bool = False + + # --------------------------------------------------------------------------- # Workstream dataclass # --------------------------------------------------------------------------- @@ -151,19 +214,19 @@ class Workstream: # turn-shaped (length cap, cross-user guard) and must be # unreachable during command windows — deferred entries live on # ``_pending_sends`` below. - worker_kind: str = field(default="", repr=False) - # Sends deferred during a command window (full-fidelity pending - # entries — see ``session_routes._PendingSend``), dispatched in - # arrival order by the per-workstream drain task when the window - # closes. Appends, retract-marks and claims all happen under + worker_kind: WorkerKind = field(default="", repr=False) + # Sends deferred while the order barrier holds (full-fidelity + # pending entries — see :class:`_PendingSend` above), dispatched in + # arrival order by the per-workstream drain thread when the slot + # frees. Appends, retract-marks and claims all happen under # ``_lock``. Node-local and in-memory: entries die with the # workstream or the process (the interjection queue's lifetime) — # the /send contract documents the at-most-once consequence. - _pending_sends: list[Any] = field(default_factory=list, repr=False) + _pending_sends: list[_PendingSend] = field(default_factory=list, repr=False) # Single-flight guard for the drain (a daemon ``threading.Thread`` # while one is live). Written under ``_lock``; the drain clears it # before exiting so a later deferred send starts a fresh one. - _pending_drain: Any = field(default=None, repr=False) + _pending_drain: threading.Thread | None = field(default=None, repr=False) # True once ``SessionManager.commit_create`` (or the non-deferred # path through ``SessionManager.create``) has fired the lifecycle # ``emit_created`` event for this workstream. Used by @@ -183,3 +246,35 @@ class Workstream: def __post_init__(self) -> None: if not self.name: self.name = f"ws-{self.id[:4]}" + + def send_barrier_active(self) -> bool: + """True while the /send order barrier holds — the ONE definition. + + The barrier is a two-term pair, and every dispatch surface that + must not overtake acknowledged sends consults it here (the /send + route's defer probe, ``CoordinatorAdapter.send``'s refusal, the + queued-nudge wake gate) instead of hand-copying the terms: + + * ``_pending_sends`` non-empty — acknowledged entries are waiting + (retract-marked husks count until the drain's loop-top purge: + they still occupy their arrival slot). + * the drain is alive — an entry may be CLAIMED (popped, dispatch + in flight); the :class:`_PendingSend` invariant ("drain not + alive ⇒ nothing claimed") is what makes these two terms + exhaustive, with no third state. The pair also covers the + drain-spawn window because ``_defer_send`` appends the entry + and starts the thread under one ``_lock`` acquisition — the + list term is always set before a not-yet-started drain could + be observed. + + Lockless callers (the wake gate, the coordinator adapter) get + benign staleness in both directions: a stale True skips once + more and the next barrier-clearing path re-runs the gate (every + deferred turn's exit backstop, plus the drain's own clean exit); + a stale False means the concurrent defer holds no order contract + against the caller anyway. Callers that need the answer atomic + with a mutation (the route's probe-then-append) hold ``_lock`` + around the call. + """ + drain = self._pending_drain + return bool(self._pending_sends) or (drain is not None and drain.is_alive()) diff --git a/turnstone/server.py b/turnstone/server.py index 837c1535..a07db33c 100644 --- a/turnstone/server.py +++ b/turnstone/server.py @@ -1624,6 +1624,15 @@ async def metrics_endpoint(request: Request) -> Response: return Response(content, media_type="text/plain; version=0.0.4; charset=utf-8") +# Quick-command completion backstop. MUST stay strictly below the +# console proxy's client timeout (console/server.py +# _PROXY_CLIENT_TIMEOUT_S = 30) or the degraded ``running`` answer can +# never traverse a proxied pane — the proxy aborts first, the pane's +# dispatch swallows the 5xx, and the caller sees nothing at all. The +# inequality is pinned by a test importing both constants. +_COMMAND_RESPONSE_BACKSTOP_S = 25 + + def _capture_cancel_forensics(session: Any, ui: Any, *, was_running: bool) -> dict[str, Any]: """Snapshot in-flight session state for the cancel response. @@ -1727,7 +1736,7 @@ async def command(request: Request) -> JSONResponse: session = ws.session cmd_ui = ui - busy_hit = {"hit": False} + busy_hit = False def _reject_busy() -> None: # Worker already running (a turn or another command is in @@ -1735,7 +1744,8 @@ async def command(request: Request) -> JSONResponse: # instead. Server-side mirror of the composer's client guard, # and a strict improvement for API callers: the old inline path # let /clear & co. mutate the session mid-turn. - busy_hit["hit"] = True + nonlocal busy_hit + busy_hit = True def _dispatch_command( run: Callable[[], None], thread_name: str, busy_hint: str @@ -1753,16 +1763,33 @@ async def command(request: Request) -> JSONResponse: full-fidelity sends by the pending-send drain when the window closes. """ - dispatched = session_worker.send( - ws, - enqueue=_reject_busy, - run=run, - thread_name=thread_name, - worker_kind="command", - ) + try: + dispatched = session_worker.send( + ws, + enqueue=_reject_busy, + run=run, + thread_name=thread_name, + worker_kind="command", + ) + except Exception: + # Thread.start failed (exhaustion, MemoryError) — the + # dispatcher rolled the slot claim back and re-raised. + # Answer 503, NOT the endpoint's generic 200-ok arm: a + # command worker that never spawned must be as loud as + # the busy 409 below — a status-code-only SDK caller + # would otherwise believe its /clear//name//resume + # applied and silently run against un-changed state. + log.exception("command.worker_spawn_failed ws=%s", ws.id[:8]) + return JSONResponse( + { + "status": "error", + "error": "Command worker could not be started — retry shortly.", + }, + status_code=503, + ) if not dispatched: return JSONResponse({"error": "Unknown workstream"}, status_code=404) - if busy_hit["hit"]: + if busy_hit: # 409, not 200: the refusal is deliberate (mutual exclusion # replaced the old inline mid-turn interleave) but it must # be LOUD — a status-code-only SDK caller treats a 200 as @@ -1916,14 +1943,14 @@ async def command(request: Request) -> JSONResponse: # panes. Loop-native wait (call_soon_threadsafe from the worker's # finally): a thread parked in Event.wait via to_thread would hold # a shared default-executor slot for the whole wait per wedged - # command. 25s: strictly under the console proxy's 30s client - # timeout (console/server.py proxy_client) so the degraded + # command. The bound (_COMMAND_RESPONSE_BACKSTOP_S) sits strictly + # under the console proxy's client timeout so the degraded # ``running`` answer can actually traverse a proxied pane — at 60s # the proxy aborted first, the pane's dispatch swallowed the 5xx, # and the user saw nothing at all (the same bounded-caller # reasoning that redesigned /send's command-window path). try: - async with asyncio.timeout(25): + async with asyncio.timeout(_COMMAND_RESPONSE_BACKSTOP_S): await done.wait() except TimeoutError: return JSONResponse({"status": "running"}) diff --git a/turnstone/shared_static/chat.css b/turnstone/shared_static/chat.css index 51ff22cf..58c610db 100644 --- a/turnstone/shared_static/chat.css +++ b/turnstone/shared_static/chat.css @@ -998,17 +998,20 @@ the progress bar) and the settled result (token delta + summary fold). Built by ``buildCompactionProgressCard`` / ``buildCompactionCard`` (shared conversation.js) for the interactive pane AND the coord viewer. - Magenta accent: a memory/infrastructure operation — distinct from the cyan - tool surface, the yellow operator bubbles, and the amber user turns. */ + Blue accent: a system/infrastructure-info operation — distinct from the + cyan tool surface, the yellow operator bubbles, and the amber user turns. + NOT magenta: that accent is reserved for the MCP surface (.scope-mcp and + the admin scope chips); blue's other uses are admin-surface only (model + status dots, provider chips) and never co-render with transcript cards. */ .msg.compaction-card { - border-left-color: var(--magenta); + border-left-color: var(--blue); background: var(--panel); padding: 8px 12px; margin: 6px 0; width: 100%; } .msg.compaction-card .msg-compaction-header { - color: var(--magenta); + color: var(--blue); font-family: var(--font-mono); font-size: 11px; font-weight: 600; @@ -1030,14 +1033,14 @@ height: 4px; margin-top: 8px; border-radius: 2px; - background: var(--magenta-glow); + background: var(--blue-glow); overflow: hidden; } .msg.compaction-card .msg-compaction-bar-fill { height: 100%; width: 0; border-radius: 2px; - background: var(--magenta); + background: var(--blue); transition: width 300ms ease; } /* Indeterminate state — a single-batch summarization emits no part-k/N diff --git a/turnstone/shared_static/composer_queue.js b/turnstone/shared_static/composer_queue.js index 0abff579..ed8ea7dc 100644 --- a/turnstone/shared_static/composer_queue.js +++ b/turnstone/shared_static/composer_queue.js @@ -122,10 +122,17 @@ export function createQueueController(opts) { // barrier lets a deferred entry dispatch within milliseconds of its ack, // so the SSE message_dispatched can beat the POST response's .then — // without this, the chip would stay flagged deferred and the idle sweep - // would skip it forever. bind() reconciles and deletes; capped small - // because chip-absent settles (other tabs, replays) also land here and - // never get consumed. - var _preBindSettles = new Map(); // msgId -> folded + // would skip it forever. bind() reconciles and deletes. Expiry is + // TTL-based, NOT a size cap: when a window closes with many deferred + // sends, the drain settles them FIFO in one burst — this tab's own + // raced settle parks FIRST and a count cap would evict exactly it as + // the foreign settles (other tabs' messages, which never get consumed + // here) pile in behind. 30s dominates every path that can still + // consume an entry: bind() runs off the send POST, client-aborted at + // 15s in both panes. Growth is rate-bounded (settle frequency × TTL); + // stale foreign entries expire on the next insert's purge. + var PRE_BIND_SETTLE_TTL_MS = 30000; + var _preBindSettles = new Map(); // msgId -> {folded, at} // Upper bound on the dequeue DELETE so a wedged proxied node (the exact // case this flow targets) can't leave a card stuck "dismissing" forever. var DELETE_TIMEOUT_MS = 15000; @@ -366,9 +373,9 @@ export function createQueueController(opts) { // fold-in settle → clear the flag, the chip is a normal queued // interjection from here. if (el.dataset.deferred && _preBindSettles.has(msgId)) { - var racedFolded = _preBindSettles.get(msgId); + var raced = _preBindSettles.get(msgId); _preBindSettles.delete(msgId); - if (racedFolded) { + if (raced.folded) { delete el.dataset.deferred; } else { _promote(el); @@ -456,11 +463,13 @@ export function createQueueController(opts) { }); if (!target) { // No bound chip yet: either another tab's message (never consumed — - // hence the cap) or this tab's bind() is still in the POST + // hence the TTL expiry) or this tab's bind() is still in the POST // round-trip; park the settle for bind() to reconcile. - _preBindSettles.set(msgId, !!folded); - if (_preBindSettles.size > 8) - _preBindSettles.delete(_preBindSettles.keys().next().value); + var now = Date.now(); + _preBindSettles.forEach(function (v, k) { + if (now - v.at > PRE_BIND_SETTLE_TTL_MS) _preBindSettles.delete(k); + }); + _preBindSettles.set(msgId, { folded: !!folded, at: now }); return; } if (target.hasAttribute("aria-busy")) return; // dismiss in flight — let it settle @@ -481,6 +490,120 @@ export function createQueueController(opts) { }; } +// --- Send-response settle (shared by both panes) ----------------------------- + +// Strip the "!!!" priority prefix for the optimistic bubble — the server +// re-parses it authoritatively; this is display-only. Exported here because +// priority is this module's vocabulary (it feeds addQueuedMessage's badge) +// and both panes need the identical parse pre-POST. +export function parsePriority(text) { + if (text.startsWith("!!!")) { + return { displayText: text.slice(3).trimStart(), priority: "important" }; + } + return { displayText: text, priority: "notice" }; +} + +// Settle a parsed /send response against the pane's optimistic state — the +// ONE implementation of the status dispatch both panes share (the +// applyCompactionEvent hooks pattern: everything pane-specific arrives via +// ctx). The fetch-stage concerns (409 pre-parse, network .catch) stay +// per-pane; this owns everything after a parsed 2xx/handled body. +// +// ctx: +// queuedEl: the pre-POST queued chip (busy pane) or null +// optimisticEl: the pre-POST sent bubble (idle pane) or null +// isBusy: the pane's busy flag AT SEND TIME +// displayText/priority: parsePriority() of the sent text +// setBusy(b): pane busy setter +// 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) +// renderError(msg): pane error row +// consumeAttachments(attached_ids, dropped_ids): composer chip sync +// +// Status arms: +// queued — bind the chip (retro-converting the idle pane's optimistic +// bubble into a REAL queued chip when deferred: a parked, +// still-retractable, restart-droppable message must not render as +// 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. +// 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 +// under the same guard (no worker and no drain may exist to ever emit +// a state event; leaving busy strands the composer in Stop mode). +// busy / attachments_busy / cross_user_interjection / unknown-ok — +// the panes' historical shapes, verbatim. +export function settleSendResponse(queue, data, ctx) { + var status = data && data.status; + if (status === "queued" && data.msg_id) { + var queuedEl = ctx.queuedEl; + if (!queuedEl && data.deferred) { + if (ctx.optimisticEl && ctx.optimisticEl.isConnected) + ctx.optimisticEl.remove(); + queuedEl = queue.addQueuedMessage(ctx.displayText, ctx.priority); + } + if (queuedEl) { + queue.bind(queuedEl, data.msg_id, { + deferred: !!data.deferred, + attachedCount: (data.attached_ids || []).length, + }); + if (data.deferred && ctx.busyIsOptimistic()) ctx.setBusy(false); + } 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. + ctx.setBusy(true); + } + ctx.consumeAttachments(data.attached_ids, data.dropped_attachment_ids); + return; + } + if (status === "busy") { + if (ctx.queuedEl) queue.remove(ctx.queuedEl); + ctx.renderError("Server is busy. Please wait."); + if (!ctx.isBusy) ctx.setBusy(false); + return; + } + if (status === "queue_full") { + if (ctx.queuedEl) { + queue.remove(ctx.queuedEl); + } else { + if (ctx.optimisticEl && ctx.optimisticEl.isConnected) + ctx.optimisticEl.remove(); + if (ctx.busyIsOptimistic()) ctx.setBusy(false); + } + ctx.renderError("Message queue full. Please wait."); + return; + } + if (status === "attachments_busy") { + if (ctx.queuedEl) queue.remove(ctx.queuedEl); + ctx.renderError( + "Attachments can't be sent while the assistant is working. " + + "Send a text-only message now, or wait and resend with attachments.", + ); + return; + } + if (status === "cross_user_interjection") { + if (ctx.queuedEl) queue.remove(ctx.queuedEl); + ctx.renderError( + data.error || + "Another participant's turn is in progress. Wait for it to " + + "finish, then send your message.", + ); + if (!ctx.isBusy) ctx.setBusy(false); + return; + } + // Unknown / "ok" status (e.g. the stale-busy race: the client + // optimistically queued but the server ran the send on a fresh worker). + // Settle the optimistic chip as a normal sent message so a pre-bind × + // can't strand it; promote() notifies "already sent" if it was dismissed. + 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. diff --git a/turnstone/shared_static/interactive.js b/turnstone/shared_static/interactive.js index 760ff23e..b903624d 100644 --- a/turnstone/shared_static/interactive.js +++ b/turnstone/shared_static/interactive.js @@ -47,7 +47,11 @@ import { createAttachmentController, kindIcon, } from "./composer_attachments.js"; -import { createQueueController } from "./composer_queue.js"; +import { + createQueueController, + parsePriority, + settleSendResponse, +} from "./composer_queue.js"; import { StatusBar } from "./status_bar.js"; import { streamingRender, streamingRenderFinalize } from "./renderer.js"; import { setMarkdown, operatorSourceLabel } from "./utils.js"; @@ -218,6 +222,9 @@ class Pane { this.currentReasoningEl = null; this.contentBuffer = ""; this.busy = false; + // Provenance of the current busy=true (see setBusy): "server" | + // "optimistic" | null when idle. + this.busySource = null; this.isThinking = false; // Acting user (turn initiator) of the in-flight turn, from state_change // events; drives the shared-workstream cross-user send gate. Carries the @@ -385,8 +392,15 @@ class Pane { // reset). queue.onIdleEdge runs only on the actual edge — it carries // the heavier work (querySelectorAll-driven promote sweep + cancel- // timer cleanup wired via the queue's onIdle hook). - setBusy(b) { + setBusy(b, source) { const next = !!b; + // Who asserted busy: "server" (default — state events, thinking_start, + // every existing/future writer) or "optimistic" (ONLY the send flow's + // pre-POST flip). The deferred/queue_full settle arms may clear busy + // solely when it is still this send's own optimistic flip — a server- + // stamped busy is a real turn and must never be clobbered. Centralized + // HERE so an unstamped future writer fails safe as "server". + this.busySource = next ? source || "server" : null; this.composer.setBusy(next); this.messagesEl.dataset.busy = next ? "true" : "false"; const edge = next !== this.busy; @@ -2041,6 +2055,14 @@ class Pane { this._pendingEditSend = null; this.setBusy(true); this.addUserMessage(editText); + // Known settle gap (deliberately deferred, pre-branch path): + // this POST consumes only the .catch — a queued/deferred/ + // queue_full body is silently dropped, so a /compact window + // opened from another tab in exactly this instant leaves the + // resent message parked with no chip and busy stranded + // "server". Narrow (rewind just ran; the slot was ours) and + // original-strata; route through settleSendResponse when + // this flow is next touched. authFetch( this._base + "/v1/api/workstreams/" + @@ -3728,13 +3750,20 @@ class Pane { body.error || "Session is busy — try again shortly.", ); } else if (body.status === "running") { - // The command outlived the endpoint's 25s completion backstop - // (kept under the console proxy's 30s bound precisely so this - // answer can traverse a proxied pane). The worker is still - // going; its output and pane refreshes arrive over SSE. + // The command outlived the endpoint's completion backstop + // (kept under the console proxy's client bound precisely so + // this answer can traverse a proxied pane). The worker is + // still going; its output and pane refreshes arrive over SSE. this.addInfoMessage( "Command is still running — results will appear here when it finishes.", ); + } else if (body.status === "error") { + // 503: the command worker could not be started (thread + // exhaustion) — the command did NOT run. Loud, like the busy + // arm: silence here reads as success. + this.addErrorMessage( + body.error || "Command failed to start — retry shortly.", + ); } }) .catch(() => {}); @@ -3750,22 +3779,19 @@ class Pane { let optimisticEl = null; const snap = this.attachments.snapshot(); - // Server re-parses the !!! prefix to set queue priority — the - // optimistic bubble strips it for display. Parsed outside the busy - // branch: the queued+deferred response arm needs it too (an idle - // pane's send can defer behind a command window / pending list). - let displayText = text; - let priority = "notice"; - if (text.startsWith("!!!")) { - displayText = text.slice(3).trimStart(); - priority = "important"; - } + // Display-only strip of the !!! prefix (the server re-parses it + // authoritatively); shared parse so the settle helper's retro-convert + // renders the same chip either pane would have built pre-POST. + const { displayText, priority } = parsePriority(text); if (isBusy) { this.removeEmptyState(); queuedEl = this.queue.addQueuedMessage(displayText, priority); } else { - this.setBusy(true); + // "optimistic": no server state event asserted this — the settle + // arms may undo it if the send turns out deferred/refused (see + // setBusy's busySource contract). + this.setBusy(true, "optimistic"); optimisticEl = this.addUserMessage(text, snap.attachments); } this.composer.clear(); @@ -3840,72 +3866,22 @@ class Pane { return r.json(); }) .then((data) => { - if (data.status === "queued" && data.msg_id) { - // queuedEl-present path: bind() handles the three known races - // (pre-bind dismiss, promote sweep raced ahead, normal accept). - // queuedEl-absent + deferred: the pane thought it was idle but - // the send parked on the server's deferred list (command window - // / order barrier) — a plain sent bubble would present a parked, - // still-retractable, restart-droppable message as delivered, so - // replace the optimistic bubble with a real queued chip carrying - // the dismiss affordance. queuedEl-absent + undeferred: plain - // interjection into a live turn the client hadn't seen yet; - // keep the historical small-UX-gap behavior (busy flip only). - if (!queuedEl && data.deferred) { - if (optimisticEl && optimisticEl.isConnected) optimisticEl.remove(); - queuedEl = this.queue.addQueuedMessage(displayText, priority); - } - if (queuedEl) { - this.queue.bind(queuedEl, data.msg_id, { - deferred: !!data.deferred, - attachedCount: (data.attached_ids || []).length, - }); - } else this.setBusy(true); - this.attachments.consume( - data.attached_ids, - data.dropped_attachment_ids, - ); - } else if (data.status === "busy") { - if (queuedEl) this.queue.remove(queuedEl); - this.addErrorMessage("Server is busy. Please wait."); - if (!isBusy) this.setBusy(false); - } else if (data.status === "queue_full") { - if (queuedEl) this.queue.remove(queuedEl); - this.addErrorMessage("Message queue full. Please wait."); - } else if (data.status === "attachments_busy") { - // Attachments can't ride a queued user turn — server held the - // chips' reservations long enough to bounce the request and - // released them. Surface to the user; chips stay in the - // composer so they can retry once the assistant finishes. - if (queuedEl) this.queue.remove(queuedEl); - this.addErrorMessage( - "Attachments can't be sent while the assistant is working. " + - "Send a text-only message now, or wait and resend with attachments.", - ); - } else if (data.status === "cross_user_interjection") { - // Another participant's turn is in flight; the server refused the - // interjection so it can't run under their credentials or be - // misattributed. The send gate normally disables the button first; - // this handles the race where the click beat the state_change. - if (queuedEl) this.queue.remove(queuedEl); - this.addErrorMessage( - data.error || - "Another participant's turn is in progress. Wait for it to " + - "finish, then send your message.", - ); - if (!isBusy) this.setBusy(false); - } else { - // Unknown / "ok" status (e.g. the stale-busy race: the client - // optimistically queued but the server ran the send on a fresh - // worker). Settle the optimistic bubble as a normal sent message - // so a pre-bind × can't strand it in the dismissing state; - // promote() notifies "already sent" if it was dismissed. - if (queuedEl) this.queue.promote(queuedEl); - this.attachments.consume( - data.attached_ids, - data.dropped_attachment_ids, - ); - } + // The full status dispatch (queued/retro-convert, busy, + // queue_full, attachments_busy, cross_user, unknown-ok) lives in + // the shared helper — ONE settle matrix for both panes; see + // settleSendResponse's contract for the arm semantics. + settleSendResponse(this.queue, data, { + queuedEl, + optimisticEl, + isBusy, + displayText, + priority, + setBusy: (b) => this.setBusy(b), + busyIsOptimistic: () => this.busy && this.busySource === "optimistic", + renderError: (msg) => this.addErrorMessage(msg), + consumeAttachments: (attached, droppedIds) => + this.attachments.consume(attached, droppedIds), + }); }) .catch((err) => { if (queuedEl) this.queue.remove(queuedEl);