diff --git a/CHANGELOG.md b/CHANGELOG.md index 107e5d29..33f12c86 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -199,14 +199,22 @@ Earlier stable lines (`stable/1.6`, `stable/1.5`) are frozen. summary HTTP call itself (the compaction lane registers its stream in the same abort seam the main loop uses), so cancelling a compaction is immediate instead of waiting out a model call. Messages sent while any - slash command holds the worker slot **park** and then run as ordinary - full-fidelity sends when the command finishes — they are never routed - through the mid-turn interjection queue, whose semantics are - turn-shaped: previously a send during a manual `/compact` was silently - truncated to 2,000 characters, a second participant in a shared - workstream was locked out with a misleading "another participant's - turn" 409 for the whole compaction, and a message queued across a - `/resume`/`/new` could be answered into the post-swap workstream. + slash command holds the worker slot are **deferred**: answered + `{"status": "queued", "msg_id"}` immediately and dispatched as ordinary + full-fidelity sends (attachments and sender identity included) when the + command finishes — never routed through the mid-turn interjection + queue, whose semantics are turn-shaped: previously a send during a + manual `/compact` was silently truncated to 2,000 characters, a second + participant in a shared workstream was locked out with a misleading + "another participant's turn" 409 for the whole compaction, and a + message queued across a `/resume`/`/new` could be answered into the + post-swap workstream. Because the response is immediate, + timeout-bounded callers — the coordinator's `send_message`, the console + proxy, SDKs, anything behind a stock reverse proxy — can no longer lose + a message to a multi-minute command window; the deferred send is + 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). 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, @@ -230,17 +238,31 @@ Earlier stable lines (`stable/1.6`, `stable/1.5`) are frozen. compaction (a new turn's generation claim breaks a stale latch). Every retry backoff on the session (stream retries, task agents, notify delivery, compaction) now aborts immediately on Stop via one shared - cancel-aware helper instead of sleeping out its exponential delay. The - web composer bounds a parked send at 10 minutes while a compaction card - is visible (15 s wedged-node default otherwise, shared by both panes), - and dismissing a queued bubble aborts the in-flight POST so a dismissed - message can't dispatch minutes later. A compaction failure reports + cancel-aware helper instead of sleeping out its exponential delay. + Dismissing a queued bubble — interjection or deferred — is a + server-confirmed `DELETE`, and retracting a deferred send that carried + attachments tells the user they were discarded instead of silently + expiring them. A compaction failure reports exactly once (auto-compaction errors defer to the turn's fatal handler - instead of doubling the red row and the error metric), and a manual - `/compact` failure no longer crashes the CLI REPL. Post-command pane - refreshes and error notices are owner-guarded, so a force-cancelled - wedged command that unwedges late can't wipe panes or inject stray - notices into a successor turn. + instead of doubling the red row and the error metric), failed-end + notice suppression is computed once by the emitter (a `notice` bool on + the end event — in the SDKs — replaces hand-synced client policy), and + a manual `/compact` failure no longer crashes the CLI REPL. `/compact` + on a workstream showing the `error` badge restores the badge on exit + instead of stamping `idle` over it (the compaction neither retried nor + resolved the failed turn). A force-cancelled initial send that + completes late still delivers its scheduled-run completion + notification (the only completion signal unattended workstreams have); + the other post-command pane refreshes and error notices remain + owner-guarded, so a force-cancelled wedged command that unwedges late + can't wipe panes or inject stray notices into a successor turn. + Embedders driving `ChatSession` with a pre-1.8 duck-typed `SessionUI` + (no `on_compaction` hook) get the classic `on_info` compaction lines + back — threshold notice, `part k/N`, retry waits, token delta + + summary box — instead of silent history swaps. **Breaking (1.8):** + compaction feedback moved from `info` events to the typed `compaction` + SSE event; pre-1.8 SSE/SDK clients that ignore unknown event types no + longer see compaction lines (they are deliberately not dual-emitted). - **Static MCP servers: a pushed catalog change no longer wedges the shared session (#839).** The static-path `*/list_changed` handler awaited its diff --git a/docs/api-reference.md b/docs/api-reference.md index 5e080650..139b884f 100644 --- a/docs/api-reference.md +++ b/docs/api-reference.md @@ -482,13 +482,16 @@ carries `before_tokens`/`after_tokens` and the produced `summary`; `"cancelled"` / `"error"`) and a human-readable `message` — for `reason: "error"` the same message is also emitted as a paired typed `error` event (that is the renderable error surface; the end event is -card-teardown). Every end (ok or failed) carries `trigger`, and every -event carries `compaction_id` — an opaque integer correlating the -start/progress/end of one compaction run (a client that force-stopped one -compaction can use it to ignore stragglers from the abandoned run). End -events also carry `superseded`: `true` marks a force-abandoned compaction -retiring after a successor generation took over — skip failure notices for -those (an OK end's result card still stands: the history swap happened). +card-teardown). Failed ends also carry `notice`: the emitter-computed +display verdict — show `message` only when it is `true` (the server +suppresses error-reason, superseded, and cancelled-auto notices once, +centrally, so clients don't re-derive that policy). Every end (ok or +failed) carries `trigger`, and every event carries `compaction_id` — an +opaque integer correlating the start/progress/end of one compaction run (a +client that force-stopped one compaction can use it to ignore stragglers +from the abandoned run). End events also carry `superseded`: `true` marks +a force-abandoned compaction retiring after a successor generation took +over (an OK end's result card still stands: the history swap happened). Superseded start/progress events are never emitted. Exactly one `start` and one `end` are emitted per attempt, so clients can key an in-progress affordance (progress bar) on the pair. A @@ -875,19 +878,25 @@ synchronous: endpoint executed commands unconditionally mid-turn; the 409 makes the refusal loud for callers that only check the HTTP status.) -While a command holds the slot, `POST .../send` requests **park** server-side -and dispatch as ordinary full-fidelity sends when the command's window closes -— they are never routed through the mid-turn interjection queue (no length -cap, no cross-user rejection). A client that aborts a parked send before the -window closes abandons it: the message is not dispatched. Clients must -therefore bound parked sends generously: the bundled web composer uses a -10-minute abort while a compaction progress card is visible (its usual bound -is ~15 s, sized for wedged-node detection), and dismissing a queued bubble -aborts the in-flight POST so a dismissed message can't dispatch later. -Deployment note: a reverse proxy's read timeout bounds the effective park — -behind a stock 60 s proxy, sends parked longer than that fail at the proxy -(the park sees the disconnect and abandons; nothing is dispatched — resend -after the command completes, or raise the proxy read timeout). +While a command holds the slot, `POST .../send` requests are **deferred**: +the server answers `{"status": "queued", "msg_id": ...}` immediately and +dispatches the message as an ordinary full-fidelity send (attachments and +sender identity included) when the command's window closes — it is never +routed through the mid-turn interjection queue (no length cap, no cross-user +rejection). The response arrives within normal round-trip time, so +timeout-bounded clients (SDKs, proxies, the coordinator) need no special +handling. To retract a deferred send before it dispatches, issue the same +`DELETE .../send` with its `msg_id` used for queued interjections — +`{"status": "removed"}` confirms it will not dispatch; `"not_found"` means +it already dispatched (or is dispatching). Retracting a deferred send +discards any attachments it carried; re-attach to send them again. + +Durability: deferred sends are **node-local and in-memory** (the same +lifetime as the interjection queue). `"queued"` is at-most-once intake, not +durable acceptance — if the workstream is closed or the node restarts before +the window ends, the message is dropped. Anything that must survive a +restart should be re-sent after confirming dispatch (the turn appears on the +SSE stream / in `/history`). **Request body:** diff --git a/sdk/typescript/src/events.ts b/sdk/typescript/src/events.ts index c9715a55..8f8e7967 100644 --- a/sdk/typescript/src/events.ts +++ b/sdk/typescript/src/events.ts @@ -176,6 +176,12 @@ export interface CompactionEvent { * those (an OK end's result still stands; the history swap happened). */ superseded?: boolean; + /** + * Failed ends only: the emitter-computed display verdict — show + * `message` only when true, instead of re-deriving suppression from + * reason/trigger/superseded client-side. + */ + notice?: boolean; /** Present on start and on every end (ok or failed). */ trigger?: "manual" | "auto"; where?: string; diff --git a/tests/test_cooperative_compaction.py b/tests/test_cooperative_compaction.py index 54aeac66..3b16643d 100644 --- a/tests/test_cooperative_compaction.py +++ b/tests/test_cooperative_compaction.py @@ -1891,7 +1891,9 @@ class TestPreHookUICompat: """A duck-typed SessionUI predating the compaction hook gets no lifecycle events but must not crash — an unguarded emit would AttributeError inside send()'s auto-compaction and permanently - wedge every long session on such a UI.""" + wedge every long session on such a UI. (This one has no on_info + either — the fallback below is getattr-guarded the same way, so + the never-crash floor holds even for a hookless minimal UI.)""" _seed_two_messages(session) session.ui = SimpleNamespace( on_thinking_start=lambda: None, @@ -1902,6 +1904,193 @@ class TestPreHookUICompat: with patch.object(session, "_utility_completion", return_value=summary): assert session._compact_messages() is True + def _duck_ui_with_info(self): + infos: list[str] = [] + ui = SimpleNamespace( + on_thinking_start=lambda: None, + on_thinking_stop=lambda: None, + on_error=lambda _m: None, + on_info=infos.append, + ) + return ui, infos + + def test_pre_hook_ui_gets_classic_info_lines(self, session): + """A duck-typed UI WITH on_info but WITHOUT on_compaction gets the + pre-1.8 lines back through the shared renderer — an auto-compaction + must never swap history with zero announcement for embedders that + predate the hook (the old lines reached them unconditionally).""" + session.ui, infos = self._duck_ui_with_info() + session._compaction_event( + 0, {"phase": "start", "trigger": "auto", "where": "mid-turn", "pct": 80} + ) + session._compaction_event(0, {"phase": "progress", "part": 1, "total": 2, "depth": 0}) + session._compaction_event( + 0, + { + "phase": "end", + "ok": True, + "trigger": "auto", + "before_tokens": 900, + "after_tokens": 100, + "summary": "dense", + }, + ) + assert any("Auto-compacting mid-turn" in m and "80%" in m for m in infos) + assert any("compacting part 1/2" in m for m in infos) + assert any("compacted: ~900 -> ~100 tokens" in m for m in infos) + assert any("dense" in m for m in infos) + + def test_pre_hook_fallback_consumes_notice_not_policy(self, session): + """Failed-end display rides the emitter's ``notice`` stamp: a + manual cancelled end prints its message; an error-reason end (the + red row already came through on_error) and a cancelled AUTO end + stay silent — the fallback never re-derives the policy.""" + session.ui, infos = self._duck_ui_with_info() + session._compaction_event( + 0, + { + "phase": "end", + "ok": False, + "reason": "cancelled", + "message": "Compaction cancelled.", + "trigger": "manual", + }, + ) + assert infos == ["Compaction cancelled."] + infos.clear() + session._compaction_event( + 0, + { + "phase": "end", + "ok": False, + "reason": "error", + "message": "boom", + "trigger": "manual", + }, + ) + session._compaction_event( + 0, + { + "phase": "end", + "ok": False, + "reason": "cancelled", + "message": "Compaction cancelled.", + "trigger": "auto", + }, + ) + assert infos == [] + + def test_superseded_ok_end_still_renders_via_fallback(self, session): + """A superseded OK end announces a history swap that REALLY + committed — suppressing it re-creates the silent-swap failure for + duck-typed embedders (the web reducer renders the result card for + superseded OK ends for the same reason). A superseded FAILED end + stays silent via the notice stamp.""" + session.ui, infos = self._duck_ui_with_info() + session._generation = 7 # the emitting generation 3 is stale + session._compaction_event( + 3, + { + "phase": "end", + "ok": True, + "trigger": "manual", + "before_tokens": 500, + "after_tokens": 50, + "summary": "kept", + }, + ) + assert any("compacted: ~500 -> ~50 tokens" in m for m in infos) + infos.clear() + session._compaction_event( + 3, + { + "phase": "end", + "ok": False, + "reason": "cancelled", + "message": "Compaction cancelled.", + "trigger": "manual", + }, + ) + assert infos == [] + + +class TestCompactionNoticeStamp: + """_compaction_event is the single display-policy site: failed ends + carry ``notice`` — renderers show the message iff it is true, instead + of each re-deriving suppression from reason/trigger/superseded (the + cross-runtime drift trap the old hand-synced cli.py/conversation.js + clauses documented).""" + + def _emit(self, session, payload, my_generation=0): + seen: dict = {} + with patch.object(session.ui, "on_compaction", side_effect=lambda p: seen.update(p)): + session._compaction_event(my_generation, payload) + return seen + + def test_manual_cancelled_end_notice_true(self, session): + seen = self._emit( + session, + { + "phase": "end", + "ok": False, + "reason": "cancelled", + "message": "x", + "trigger": "manual", + }, + ) + assert seen["notice"] is True + + def test_error_reason_end_notice_false(self, session): + """reason=error already fired the one red on_error row — the end + event is card-teardown, never a second line.""" + seen = self._emit( + session, + { + "phase": "end", + "ok": False, + "reason": "error", + "message": "boom", + "trigger": "manual", + }, + ) + assert seen["notice"] is False + + def test_cancelled_auto_end_notice_false(self, session): + """A cancelled AUTO compaction is part of cancelling the turn — + the send loop prints its own '[Generation cancelled]'.""" + seen = self._emit( + session, + {"phase": "end", "ok": False, "reason": "cancelled", "message": "x", "trigger": "auto"}, + ) + assert seen["notice"] is False + + def test_stale_generation_end_notice_false(self, session): + """A superseded end is one nobody is waiting on — its notice + mid-turn reads as the LIVE work being cancelled.""" + session._generation = 9 + seen = self._emit( + session, + { + "phase": "end", + "ok": False, + "reason": "cancelled", + "message": "x", + "trigger": "manual", + }, + my_generation=4, + ) + assert seen["superseded"] is True + assert seen["notice"] is False + + def test_ok_end_and_start_carry_no_notice(self, session): + seen = self._emit( + session, + {"phase": "end", "ok": True, "trigger": "manual", "summary": "s"}, + ) + assert "notice" not in seen + seen = self._emit(session, {"phase": "start", "trigger": "manual"}) + assert "notice" not in seen + class TestPreSwapQueueFlush: def test_new_flushes_stranded_queue_into_old_workstream(self, session): diff --git a/tests/test_interactive_pane_js.py b/tests/test_interactive_pane_js.py index f6eaab12..cce94bbc 100644 --- a/tests/test_interactive_pane_js.py +++ b/tests/test_interactive_pane_js.py @@ -638,33 +638,39 @@ def test_connectsse_defers_open_when_tab_hidden() -> None: assert head.index('addEventListener("visibilitychange"') < head.index("if (document.hidden) {") -def test_send_abort_bound_is_compaction_aware() -> None: - """The send POST's abort bound must be selected via the shared - ``sendAbortMs`` helper in BOTH panes: sends during a slash-command - window park server-side (a manual /compact legitimately runs for - minutes), so a hard-coded ~15s abort silently dropped any send made - >15s into a long compaction. The compaction card is the long-bound - signal; the wedged-node default stays otherwise.""" +def test_send_post_abort_machinery_is_gone() -> None: + """The parked-POST era's client abort machinery must stay deleted in + BOTH panes: sends during a command window are answered "queued" + immediately (server-side defer-and-drain), so there is no long-lived + POST for a compaction-aware bound (``sendAbortMs``) to protect, and + dismissal is bind() → server-confirmed DELETE — never a POST abort + (``_sendAbort``), which fired on the interjection path too and + dispatched "dismissed" messages anyway. Reintroducing either hook + means re-parking the POST; that design deterministically dropped + messages from every timeout-bounded caller (coordinator client and + console proxy at 30s, SDKs, stock proxies).""" interactive = _INTERACTIVE.read_text(encoding="utf-8") - assert "sendAbortMs(this._compaction)" in interactive, ( - "interactive sendMessage must select its abort bound off the compaction holder" - ) coordinator = (_ROOT / "turnstone/console/static/coordinator/coordinator.js").read_text( encoding="utf-8" ) - assert "sendAbortMs(compactionHolder)" in coordinator, ( - "coordinator coordSend must share the same abort-bound policy" - ) conversation = (_ROOT / "turnstone/shared_static/conversation.js").read_text(encoding="utf-8") - assert "export function sendAbortMs" in conversation - assert "600000 : 15000" in conversation.replace("\n", " "), ( - "long bound while a compaction card is live; 15s wedged-node default otherwise" - ) - # The queued bubble's pre-response x must abort the in-flight (possibly - # parked) POST — otherwise a dismissed message dispatches anyway when - # the command window closes. - assert "queuedEl._sendAbort = () => sendCtrl.abort()" in interactive composer_queue = (_ROOT / "turnstone/shared_static/composer_queue.js").read_text( encoding="utf-8" ) - assert "el._sendAbort()" in composer_queue + for name, src in ( + ("interactive.js", interactive), + ("coordinator.js", coordinator), + ("conversation.js", conversation), + ("composer_queue.js", composer_queue), + ): + assert "sendAbortMs" not in src, f"{name}: the compaction-aware abort bound is dead" + assert "_sendAbort" not in src, f"{name}: dismiss must be bind() → DELETE, not a POST abort" + # The flat wedged-node bound stands in both panes: every /send answers + # within RTT now (dispatched / queued / deferred-with-msg_id). + assert "sendCtrl.abort(), 15000" in interactive + assert "sendCtrl.abort(), 15000" in coordinator + # Retracting a deferred send discards its attachments — both panes + # stash the count and composer_queue surfaces the consequence. + assert "_deferredAttachments" in interactive + assert "_deferredAttachments" in coordinator + assert "_deferredAttachments" in composer_queue diff --git a/tests/test_server_authz.py b/tests/test_server_authz.py index 62227a5e..1bdda0c0 100644 --- a/tests/test_server_authz.py +++ b/tests/test_server_authz.py @@ -240,12 +240,25 @@ class _FakeSession: self.compact_gate: threading.Event | None = None self.command_gate: threading.Event | None = None self.command_raises: BaseException | None = None - # Every queue_message call — the park invariant is that command + # Gate for send() — lets a test hold a TURN in flight so a + # deferred entry provably lands in (or is refused by) the + # interjection fallback at drain time. + self.send_gate: threading.Event | None = None + # Every queue_message call — the defer invariant is that command # windows NEVER reach the interjection queue. self.queue_calls: list[str] = [] + # When set, queue_message records the attempt and then raises it + # (e.g. CrossUserInterjectionError for the drain's re-park arm). + self.queue_raises: BaseException | None = None + # DELETE /send fall-through: ids the route asked this session to + # dequeue (the fake never holds interjections, so it returns + # False and the route falls through to the deferred-send list). + self.dequeues: list[str] = [] def send(self, text: str, *, attachments: Any = None, send_id: Any = None) -> None: self.sends.append((text, attachments, send_id)) + if self.send_gate is not None: + self.send_gate.wait(timeout=10) if self.send_raises is not None: raise self.send_raises @@ -257,9 +270,15 @@ class _FakeSession: interjector_user_id: str = "", ) -> tuple[str, str, str]: self.queue_calls.append(text) + if self.queue_raises is not None: + raise self.queue_raises cleaned = text[:2000] + "..." if len(text) > 2000 else text return cleaned, "notice", queue_msg_id or "m1" + def dequeue_message(self, msg_id: str) -> bool: + self.dequeues.append(msg_id) + return False + def set_watch_runner(self, *_a: Any, **_kw: Any) -> None: pass @@ -1275,23 +1294,43 @@ class TestCompactCommandDispatch: assert ws.session.compacts == 1 assert ws._worker_running is False # The slot was classified as a command window (what the /send - # route's park keys on; stale after exit is harmless — every + # route's defer keys on; stale after exit is harmless — every # reader conjoins _worker_running). assert ws.worker_kind == "command" # Exit seam: stranded-text backstop flush only — no drain, no - # answering send (sends during the window park in the /send route - # and dispatch as their own workers afterwards). + # answering send (sends during the window defer in the /send + # route and dispatch as their own workers afterwards). assert ws.session.queued_flushes == 1 assert ws.session.sends == [] # The worker wrapped the run in busy/idle state transitions. assert ws.ui.states == ["thinking", "idle"] - def test_send_during_compact_window_parks_then_dispatches_fully(self, app_client): - """A /send during a manual /compact PARKS and then runs as an - ordinary full-fidelity send — it must never enter the interjection - queue, whose 2000-char cap silently truncated pasted logs/code and - whose cross-user guard locked second participants out for the - whole compaction.""" + def _wait_for(self, predicate, timeout: float = 8.0) -> bool: + """Poll until ``predicate()`` — the deferred-send drain runs on the + app's event loop at a 0.25s cadence, so dispatch is asynchronous + with respect to the released command worker.""" + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + if predicate(): + return True + time.sleep(0.02) + return False + + def _drain_idle(self, ws) -> bool: + """True once the pending-send drain retired itself (list empty, + single-flight slot cleared) — the leaked-task guard for these + tests.""" + with ws._lock: + return not ws._pending_sends and ws._pending_drain is None + + def test_send_during_compact_window_defers_then_dispatches_fully(self, app_client): + """A /send during a manual /compact is answered "queued" + IMMEDIATELY (no parked POST — a 30s-bounded caller like the + coordinator client or console proxy must never lose a message to + a multi-minute window) and then runs as an ordinary full-fidelity + send — never the interjection queue, whose 2000-char cap silently + truncated pasted logs/code and whose cross-user guard locked + second participants out for the whole compaction.""" client, mgr = app_client ws_id = self._create_ws(client) ws = mgr.get(ws_id) @@ -1305,35 +1344,35 @@ class TestCompactCommandDispatch: ) assert resp.json() == {"status": "ok"} big = "x" * 5000 # over the interjection cap — must survive intact - send_result: dict = {} - - def _send() -> None: - r = client.post( - f"/v1/api/workstreams/{ws_id}/send", - json={"message": big}, - headers=_auth("user-1"), - ) - send_result["status"] = r.status_code - send_result["body"] = r.json() - - sender = threading.Thread(target=_send, daemon=True) - sender.start() - # The send is parked: give it time to have taken the park path, - # then prove nothing was dispatched or queued yet. - for _ in range(50): - if ws.session.queue_calls or ws.session.sends: - break - time.sleep(0.02) + # The response is immediate even though the window is wedged open: + # this request runs on the same synchronous test client, so a + # parked POST would deadlock the test rather than pass it. + r = client.post( + f"/v1/api/workstreams/{ws_id}/send", + json={"message": big}, + headers=_auth("user-1"), + ) + assert r.status_code == 200 + body = r.json() + assert body["status"] == "queued" + assert body["msg_id"] + assert body["priority"] == "notice" + # Registered, not dispatched: the window is still open. assert ws.session.sends == [] assert ws.session.queue_calls == [] # the queue is unreachable - gate.set() # compaction finishes; the park releases - sender.join(timeout=10) - assert not sender.is_alive() - assert send_result["status"] == 200 - assert send_result["body"]["status"] == "ok" - # Full fidelity: the exact 5000-char text, via a normal send. + with ws._lock: + assert len(ws._pending_sends) == 1 + gate.set() # compaction finishes; the drain dispatches + assert self._wait_for(lambda: ws.session.sends) + # Full fidelity: the exact 5000-char text, via a normal send (an + # oversized entry must take the fresh-spawn arm — the interjection + # fallback would truncate it). assert [s[0] for s in ws.session.sends] == [big] assert ws.session.queue_calls == [] + # The dispatched send carries the deferred msg_id as its send_id, + # so the client's queued bubble reconciles against the turn. + assert ws.session.sends[0][2] == body["msg_id"] + assert self._wait_for(lambda: self._drain_idle(ws)) def test_cancelled_compact_flushes_queue_without_answering(self, app_client): """A user-stopped compaction must not auto-run a turn they may no @@ -1413,10 +1452,10 @@ class TestCompactCommandDispatch: assert ws.session.sends == [] assert ws.ui.states == [] - def test_parked_send_delivers_attachments_after_release(self, app_client, monkeypatch): - """Attachments are peek-resolved BEFORE the park (the bytes ride the - request closure), so a send parked through a long compaction still - delivers them on dispatch — the pre-park refusal + def test_deferred_send_delivers_attachments_after_release(self, app_client, monkeypatch): + """Attachments are peek-resolved BEFORE the defer (the bytes ride + the pending entry), so a send deferred through a long compaction + still delivers them on dispatch — the queue-path refusal (attachments_busy) must never apply to a command window.""" client, mgr = app_client ws_id = self._create_ws(client) @@ -1435,34 +1474,31 @@ class TestCompactCommandDispatch: json={"command": "/compact", "ws_id": ws_id}, headers=_auth("user-1"), ) - send_result: dict = {} - - def _send() -> None: - r = client.post( - f"/v1/api/workstreams/{ws_id}/send", - json={"message": "with attachment", "attachment_ids": ["a1"]}, - headers=_auth("user-1"), - ) - send_result["body"] = r.json() - - sender = threading.Thread(target=_send, daemon=True) - sender.start() - time.sleep(0.3) - assert ws.session.sends == [] # still parked + r = client.post( + f"/v1/api/workstreams/{ws_id}/send", + json={"message": "with attachment", "attachment_ids": ["a1"]}, + headers=_auth("user-1"), + ) + body = r.json() + assert body["status"] == "queued" + # The attachments are TAKEN by the deferred entry (the client + # consumes its chips off this list — a retract discards them). + assert body["attached_ids"] == ["a1"] + assert ws.session.sends == [] # still deferred gate.set() - sender.join(timeout=10) - assert not sender.is_alive() - assert send_result["body"]["status"] == "ok" - assert send_result["body"]["attached_ids"] == ["a1"] - text, attachments, _sid = ws.session.sends[0] + assert self._wait_for(lambda: ws.session.sends) + text, attachments, sid = ws.session.sends[0] assert text == "with attachment" assert attachments == ["fake-attachment-bytes"] + assert sid == body["msg_id"] assert ws.session.queue_calls == [] + assert self._wait_for(lambda: self._drain_idle(ws)) - def test_send_during_quick_command_window_parks(self, app_client): - """The park applies to EVERY command window, not just /compact — a - send racing a quick command dispatches after the window with full - fidelity instead of entering the interjection queue.""" + def test_send_during_quick_command_window_defers(self, app_client): + """The defer applies to EVERY command window, not just /compact — + a send racing a quick command is answered "queued" and dispatches + after the window with full fidelity instead of entering the + interjection queue.""" client, mgr = app_client ws_id = self._create_ws(client) ws = mgr.get(ws_id) @@ -1488,29 +1524,22 @@ class TestCompactCommandDispatch: time.sleep(0.02) assert ws._worker_running assert ws.worker_kind == "command" - send_result: dict = {} - - def _send() -> None: - r = client.post( - f"/v1/api/workstreams/{ws_id}/send", - json={"message": "mid-command send"}, - headers=_auth("user-1"), - ) - send_result["body"] = r.json() - - sender = threading.Thread(target=_send, daemon=True) - sender.start() - time.sleep(0.3) # long enough for a queue-path regression to show + r = client.post( + f"/v1/api/workstreams/{ws_id}/send", + json={"message": "mid-command send"}, + headers=_auth("user-1"), + ) + send_body = r.json() + assert send_body["status"] == "queued" assert ws.session.queue_calls == [] assert ws.session.sends == [] gate.set() runner.join(timeout=10) - sender.join(timeout=10) - assert not sender.is_alive() assert cmd_result["body"] == {"status": "ok"} - assert send_result["body"]["status"] == "ok" + assert self._wait_for(lambda: ws.session.sends) assert [s[0] for s in ws.session.sends] == ["mid-command send"] assert ws.session.queue_calls == [] + assert self._wait_for(lambda: self._drain_idle(ws)) def test_clear_ui_rides_the_worker_not_the_endpoint(self, app_client): """The clear_ui follow-up runs on the worker after handle_command — @@ -1590,10 +1619,294 @@ class TestCompactCommandDispatch: ) assert ws.ui.errors == [] + def test_abandoned_init_worker_still_fires_notify(self, app_client, monkeypatch): + """The completion notify is WORKSTREAM-scoped, not slot-scoped: + _fire_notify_targets has exactly one call site (the init worker), + successor turns never notify, so an owner gate here has no + duplicate to prevent — it only converts force-cancel into + permanent notification loss for scheduled/unattended workstreams. + This pins the un-gated behavior against the tempting symmetry + 'fix' (the finally's siblings ARE owner-gated, correctly — they + mutate live slot/UI state).""" + client, mgr = app_client + fired: list = [] + monkeypatch.setattr( + "turnstone.server._fire_notify_targets", + lambda ws, content: fired.append(ws.id), + ) + gate = threading.Event() + started = threading.Event() + orig_send = _FakeSession.send + + def wedged_send(self, text, **kwargs): + started.set() + assert gate.wait(timeout=10) + orig_send(self, text, **kwargs) + + monkeypatch.setattr(_FakeSession, "send", wedged_send) + resp = client.post( + "/v1/api/workstreams/new", + json={"name": "sched", "initial_message": "do the thing"}, + headers=_auth("user-1"), + ) + assert resp.status_code == 200 + ws_id = resp.json()["ws_id"] + ws = mgr.get(ws_id) + assert ws is not None + assert started.wait(timeout=10) + worker = ws.worker_thread + with ws._lock: # the force-cancel abandon shape + ws.worker_thread = None + ws._worker_running = False + gate.set() # the abandoned init completes LATE + worker.join(timeout=10) + assert not worker.is_alive() + assert fired == [ws_id] # exactly once, despite the abandonment + + def test_compact_on_error_workstream_restores_error_badge(self, app_client): + """/compact on an ERROR workstream must exit back to 'error', not + stamp 'idle' over the operator's investigatable badge — the + compaction neither retried nor resolved the failed turn (mirrors + the orphan reaper's ERROR carve-out). The existing happy-path + test pins ['thinking', 'idle'] for an IDLE workstream.""" + from turnstone.core.workstream import WorkstreamState + + client, mgr = app_client + ws_id = self._create_ws(client) + ws = mgr.get(ws_id) + assert ws is not None + ws.state = WorkstreamState.ERROR # a prior fatal turn's badge + resp = client.post( + "/v1/api/command", + json={"command": "/compact", "ws_id": ws_id}, + headers=_auth("user-1"), + ) + assert resp.json() == {"status": "ok"} + ws.worker_thread.join(timeout=5) + assert not ws.worker_thread.is_alive() + # THINKING during the window is honest (work is happening); the + # exit restores the badge instead of clearing it. + assert ws.ui.states == ["thinking", "error"] + + def test_retracted_deferred_send_never_dispatches(self, app_client): + """DELETE /send with a deferred msg_id retracts the entry before + dispatch — the drain must skip it entirely. The route falls + through the session's interjection dequeue (which misses) to the + workstream's pending list.""" + 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"), + ) + r = client.post( + f"/v1/api/workstreams/{ws_id}/send", + json={"message": "changed my mind"}, + headers=_auth("user-1"), + ) + msg_id = r.json()["msg_id"] + d = client.request( + "DELETE", + f"/v1/api/workstreams/{ws_id}/send", + json={"msg_id": msg_id}, + headers=_auth("user-1"), + ) + assert d.json() == {"status": "removed"} + assert ws.session.dequeues == [msg_id] # fall-through was exercised + gate.set() + assert self._wait_for(lambda: self._drain_idle(ws)) + assert ws.session.sends == [] + assert ws.session.queue_calls == [] + # Unknown ids still answer not_found after checking both holders. + d2 = client.request( + "DELETE", + f"/v1/api/workstreams/{ws_id}/send", + json={"msg_id": "nope"}, + headers=_auth("user-1"), + ) + assert d2.json() == {"status": "not_found"} + + def test_deferred_sends_dispatch_in_arrival_order(self, app_client): + """Two sends deferred behind one window dispatch in arrival order, + each as its own full-fidelity turn (the fake's sends complete + instantly, so the second never needs the interjection fallback).""" + 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"), + ) + first = client.post( + f"/v1/api/workstreams/{ws_id}/send", + json={"message": "first"}, + headers=_auth("user-1"), + ).json() + second = client.post( + f"/v1/api/workstreams/{ws_id}/send", + json={"message": "second"}, + headers=_auth("user-1"), + ).json() + assert first["status"] == second["status"] == "queued" + assert first["msg_id"] != second["msg_id"] + gate.set() + assert self._wait_for(lambda: len(ws.session.sends) == 2) + assert [s[0] for s in ws.session.sends] == ["first", "second"] + assert self._wait_for(lambda: self._drain_idle(ws)) + + def test_force_cancel_of_wedged_command_releases_drain(self, app_client): + """Force-cancelling a wedged command clears the slot flags — the + drain polls the same (_worker_running, worker_kind) pair, so a + deferred message dispatches WITHOUT waiting for the abandoned + thread to unwedge (the operator's escape hatch keeps working + under defer exactly as it did under park).""" + 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"), + ) + zombie = ws.worker_thread + r = client.post( + f"/v1/api/workstreams/{ws_id}/send", + json={"message": "deferred behind wedge"}, + headers=_auth("user-1"), + ) + assert r.json()["status"] == "queued" + resp = client.post( + f"/v1/api/workstreams/{ws_id}/cancel", + json={"force": True}, + headers=_auth("user-1"), + ) + assert resp.status_code == 200 + # Dispatch happens while the zombie is STILL wedged on the gate. + assert self._wait_for(lambda: ws.session.sends) + assert [s[0] for s in ws.session.sends] == ["deferred behind wedge"] + gate.set() # let the abandoned thread finish and be joined + zombie.join(timeout=5) + assert not zombie.is_alive() + assert self._wait_for(lambda: self._drain_idle(ws)) + + def test_ws_close_mid_window_drops_pending_and_drain_exits(self, app_client): + """A workstream closed with deferred sends outstanding drops them + (documented at-most-once contract) and the drain task retires + itself — no leaked task, no dispatch into a closed workstream.""" + 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"), + ) + r = client.post( + f"/v1/api/workstreams/{ws_id}/send", + json={"message": "never delivered"}, + headers=_auth("user-1"), + ) + assert r.json()["status"] == "queued" + with ws._lock: + ws._closed = True # the SessionManager.close tombstone shape + gate.set() + assert self._wait_for(lambda: self._drain_idle(ws)) + assert ws.session.sends == [] + assert ws.session.queue_calls == [] + + def test_deferred_send_dispatches_into_post_swap_session(self, app_client): + """The drain re-captures ws.session per attempt: a /resume-style + identity swap during the window routes the deferred message into + the POST-swap session — the same guarantee the park's + per-iteration re-capture provided.""" + client, mgr = app_client + ws_id = self._create_ws(client) + ws = mgr.get(ws_id) + assert ws is not None + old_session = ws.session + gate = threading.Event() + old_session.compact_gate = gate + client.post( + "/v1/api/command", + json={"command": "/compact", "ws_id": ws_id}, + headers=_auth("user-1"), + ) + r = client.post( + f"/v1/api/workstreams/{ws_id}/send", + json={"message": "post-swap please"}, + headers=_auth("user-1"), + ) + assert r.json()["status"] == "queued" + new_session = _FakeSession(ws_id=ws_id, user_id="user-1") + ws.session = new_session # the in-place identity swap + gate.set() + assert self._wait_for(lambda: new_session.sends) + assert [s[0] for s in new_session.sends] == ["post-swap please"] + assert old_session.sends == [] + assert self._wait_for(lambda: self._drain_idle(ws)) + + def test_rejected_deferred_entry_waits_for_slot_then_fresh_spawns(self, app_client): + """The drain's rejection arm: an entry the interjection fallback + refuses (cross-user here; attachments/oversized take the same + path) is NOT dropped — it waits for the slot to free and then + dispatches as its own fresh turn. The queued ack must never + become a silent drop while the workstream lives.""" + from turnstone.core.session import CrossUserInterjectionError + + client, mgr = app_client + ws_id = self._create_ws(client) + ws = mgr.get(ws_id) + assert ws is not None + gate = threading.Event() + send_gate = threading.Event() + ws.session.compact_gate = gate + ws.session.send_gate = send_gate + ws.session.queue_raises = CrossUserInterjectionError("another participant's turn") + client.post( + "/v1/api/command", + json={"command": "/compact", "ws_id": ws_id}, + headers=_auth("user-1"), + ) + for msg in ("first", "second"): + r = client.post( + f"/v1/api/workstreams/{ws_id}/send", + json={"message": msg}, + headers=_auth("user-1"), + ) + assert r.json()["status"] == "queued" + gate.set() + # Entry 1 spawns fresh and WEDGES in send() on send_gate — a live + # turn now holds the slot, so entry 2 takes the interjection + # fallback and is refused (the fake raises cross-user). + assert self._wait_for(lambda: ws.session.queue_calls == ["second"]) + assert [s[0] for s in ws.session.sends] == ["first"] + send_gate.set() # the turn ends; the slot frees + assert self._wait_for(lambda: len(ws.session.sends) == 2) + # Entry 2 dispatched as its own fresh turn — exactly one refused + # queue attempt, then the fresh-spawn arm. + assert [s[0] for s in ws.session.sends] == ["first", "second"] + assert ws.session.queue_calls == ["second"] + assert self._wait_for(lambda: self._drain_idle(ws)) + 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 - during the window park in the /send route and belong to whatever + during the window defer in the /send route and belong to whatever follows the shutdown, not to this worker).""" client, mgr = app_client ws_id = self._create_ws(client) diff --git a/turnstone/cli.py b/turnstone/cli.py index 682a9a78..800f5a80 100644 --- a/turnstone/cli.py +++ b/turnstone/cli.py @@ -17,6 +17,7 @@ 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 @@ -333,61 +334,11 @@ class TerminalUI(SessionUI): 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). """ - phase = payload.get("phase") - if phase == "start": - # The threshold notice prints only when a threshold actually - # fired (pct present — _do_auto_compact). The overflow-retry - # path compacts with auto=True but prints its own "[Context - # overflow — auto-compacting and retrying]" line; claiming a - # percentage there would fabricate the trigger. Manual starts - # print nothing — the thinking spinner covers it. - pct = payload.get("pct") - if payload.get("trigger") == "auto" and pct is not None: - where = payload.get("where") or "" - qualifier = f" {where}" if where else "" - self.on_info( - f"\n[Auto-compacting{qualifier}: prompt exceeds {pct}% of context window]" - ) - elif phase == "progress": - if payload.get("warning") == "summary_truncated": - self.on_info("[Warning: compaction summary was truncated]") - elif payload.get("retry_in") is not None: - self.on_info( - f"[Compact retrying in {payload['retry_in']:.0f}s: {payload.get('error', '')}]" - ) - else: - self.on_info(f"[compacting part {payload.get('part')}/{payload.get('total')}…]") - elif phase == "end": - if payload.get("ok"): - before = payload.get("before_tokens", 0) - after = payload.get("after_tokens", 0) - self.on_info(f"[compacted: ~{before:,} -> ~{after:,} tokens]") - separator = "─" * 60 - lines = [separator] - for line in str(payload.get("summary") or "").splitlines(): - lines.append(f" {line}") - lines.append(separator) - self.on_info("\n".join(lines)) - elif payload.get("reason") != "error": - # reason="error" already printed through on_error (red) — - # the end event is card-teardown for the web panes, not a - # second line here. A cancelled AUTO compaction is also - # silent: the surrounding send prints "[Generation - # cancelled]" itself, and pre-lifecycle-events one Stop - # printed exactly one line. (A cancelled MANUAL /compact - # keeps the message — it's the only line that path prints.) - # A SUPERSEDED end is a force-abandoned compaction retiring - # late — nobody is waiting on it; narrating it mid-turn - # reads as the LIVE work being cancelled. - # HAND-SYNCED SIBLING: conversation.js applyCompactionEvent - # implements this same three-clause display policy (skip - # error-reason / skip superseded / skip cancelled+auto) for - # the web panes — change both or they drift. - if not payload.get("superseded") and not ( - payload.get("reason") == "cancelled" and payload.get("trigger") == "auto" - ): - self.on_info(str(payload.get("message") or "")) + render_compaction_event_as_info(payload, self.on_info) return None def on_state_change(self, state: str) -> None: diff --git a/turnstone/console/coordinator_adapter.py b/turnstone/console/coordinator_adapter.py index 147ddb1e..2b144907 100644 --- a/turnstone/console/coordinator_adapter.py +++ b/turnstone/console/coordinator_adapter.py @@ -378,7 +378,7 @@ class CoordinatorAdapter: # at this workstream from the interactive UI) holds the # raced slot: the interjection queue is turn-shaped and must # stay unreachable during command windows — same rule as the - # /send route's park. Fail the dispatch (returns False, the + # /send route's defer. Fail the dispatch (returns False, the # caller's retryable-backpressure surface) rather than queue # a message that would be capped and could cross a /resume # identity swap. diff --git a/turnstone/console/server.py b/turnstone/console/server.py index 71f69ca4..7fddfbde 100644 --- a/turnstone/console/server.py +++ b/turnstone/console/server.py @@ -3887,7 +3887,7 @@ def _audit_coordinator_create( ) -def _coord_spawn_metrics(_request: Request, ui: Any) -> None: +def _coord_spawn_metrics(_request: Request | None, ui: Any) -> None: """Per-spawn counter writes for coord — mirrors interactive's pattern. Wired onto :attr:`SessionEndpointConfig.spawn_metrics`. Increments diff --git a/turnstone/console/static/coordinator/coordinator.js b/turnstone/console/static/coordinator/coordinator.js index 8d9ad242..2fbe4923 100644 --- a/turnstone/console/static/coordinator/coordinator.js +++ b/turnstone/console/static/coordinator/coordinator.js @@ -31,7 +31,6 @@ import { buildCompactionCard, applyCompactionEvent, resetCompactionHolder, - sendAbortMs, buildSystemNudgeMarker, maxSeverityItem, buildConvBatchShell, @@ -1997,10 +1996,11 @@ function createCoordinatorPane(root, wsId, opts) { let sendTimer = null; if (sendCtrl) { sendInit.signal = sendCtrl.signal; - // Same policy as the interactive composer: long bound while this - // pane's compaction card is live (the send parks server-side for - // the command window), wedged-node default otherwise. - sendTimer = setTimeout(() => sendCtrl.abort(), sendAbortMs(compactionHolder)); + // Same policy as the interactive composer: flat wedged-node bound — + // every /send answers within RTT now (dispatched, queued, or + // deferred-with-msg_id during a command window; the server parks + // nothing against this POST). Dismissal is bind() → DELETE. + sendTimer = setTimeout(() => sendCtrl.abort(), 15000); } let sendReq = authFetch( "/v1/api/workstreams/" + encodeURIComponent(wsId) + "/send", @@ -2053,8 +2053,14 @@ function createCoordinatorPane(root, wsId, opts) { // flag so any subsequent send takes the queue path correctly, // and accept the small UX gap (no in-UI dismiss for THIS // message). The server still delivers it on worker drain. - if (queuedEl) queue.bind(queuedEl, data.msg_id); - else setBusy(true); + if (queuedEl) { + // Deferred sends (command-window defer) can carry attachments — + // stash the count BEFORE bind (a pre-bind ✕ confirms inside + // bind) so the dismiss path can surface the + // discarded-attachments consequence. + queuedEl._deferredAttachments = (data.attached_ids || []).length; + queue.bind(queuedEl, data.msg_id); + } else setBusy(true); attachments.consume(data.attached_ids, data.dropped_attachment_ids); } else if (data && data.status === "busy") { if (queuedEl) queue.remove(queuedEl); diff --git a/turnstone/core/compaction_render.py b/turnstone/core/compaction_render.py new file mode 100644 index 00000000..fe029067 --- /dev/null +++ b/turnstone/core/compaction_render.py @@ -0,0 +1,69 @@ +"""Render compaction lifecycle events as classic info-channel text lines. + +One implementation, two consumers: + +* :class:`turnstone.cli.TerminalUI` — the terminal's native rendering of + ``on_compaction`` payloads (the threshold notice, ``part k/N`` + progress, and the token-delta + boxed-summary result it printed before + these became structured events). +* :meth:`ChatSession._compaction_event`'s duck-typed fallback — a + SessionUI implementation that predates the ``on_compaction`` hook gets + these same lines through its ``on_info`` instead of silence (an + auto-compaction that swaps history with zero announcement). + +Display policy: failed ends consult ``payload["notice"]`` — stamped by +the emitter (:meth:`ChatSession._compaction_event`, the single policy +site) — never re-derive suppression from reason/trigger/superseded here. +Superseded OK ends still render: the history swap really committed, and +suppressing its announcement is exactly the silent-swap failure this +module exists to prevent. +""" + +from collections.abc import Callable +from typing import Any + + +def render_compaction_event_as_info( + payload: dict[str, Any], on_info: Callable[[str], None] +) -> None: + """Print one compaction lifecycle event through ``on_info``.""" + phase = payload.get("phase") + if phase == "start": + # The threshold notice prints only when a threshold actually + # fired (pct present — _do_auto_compact). The overflow-retry + # path compacts with auto=True but prints its own "[Context + # overflow — auto-compacting and retrying]" line; claiming a + # percentage there would fabricate the trigger. Manual starts + # print nothing — the caller's own activity display covers it. + pct = payload.get("pct") + if payload.get("trigger") == "auto" and pct is not None: + where = payload.get("where") or "" + qualifier = f" {where}" if where else "" + on_info(f"\n[Auto-compacting{qualifier}: prompt exceeds {pct}% of context window]") + elif phase == "progress": + if payload.get("warning") == "summary_truncated": + on_info("[Warning: compaction summary was truncated]") + elif payload.get("retry_in") is not None: + on_info(f"[Compact retrying in {payload['retry_in']:.0f}s: {payload.get('error', '')}]") + else: + on_info(f"[compacting part {payload.get('part')}/{payload.get('total')}…]") + elif phase == "end": + if payload.get("ok"): + before = payload.get("before_tokens", 0) + after = payload.get("after_tokens", 0) + on_info(f"[compacted: ~{before:,} -> ~{after:,} tokens]") + separator = "─" * 60 + lines = [separator] + for line in str(payload.get("summary") or "").splitlines(): + lines.append(f" {line}") + lines.append(separator) + on_info("\n".join(lines)) + elif payload.get("notice"): + # The emitter stamps ``notice`` on failed ends (suppressing + # error-reason ends — already printed red through on_error — + # plus superseded and cancelled-auto ends). A payload without + # the field (an event replayed from an older node) stays + # silent: these notices are informational, and re-deriving the + # suppression here is the cross-runtime drift trap the stamp + # exists to kill. + on_info(str(payload.get("message") or "")) diff --git a/turnstone/core/session.py b/turnstone/core/session.py index 0a8eb1e5..e2e4120c 100644 --- a/turnstone/core/session.py +++ b/turnstone/core/session.py @@ -7799,16 +7799,52 @@ class ChatSession: not drive the activity-pill restore or animate a successor's card. ``my_generation`` is 0 on direct test invocations — falsy, so such events are never marked superseded (matching ``_check_cancelled``). + + Failed ends additionally carry ``notice`` — the single display- + policy site: renderers show the failure message only when it is + true, instead of each re-deriving suppression from reason/trigger/ + superseded (the cross-runtime drift trap the old hand-synced + cli.py/conversation.js clauses documented). Error-reason ends are + suppressed because :meth:`_compaction_bailed` already fired the one + red ``on_error`` row; cancelled-auto ends because the surrounding + send prints its own "[Generation cancelled]"; superseded ends + because nobody is waiting on a force-abandoned compaction and its + notice mid-turn reads as the LIVE work being cancelled. """ stale = bool(my_generation and my_generation != self._generation) + event: dict[str, Any] = {"compaction_id": my_generation, "superseded": stale, **payload} + if payload.get("phase") == "end" and not payload.get("ok"): + event["notice"] = ( + not stale + and payload.get("reason") != "error" + and not (payload.get("reason") == "cancelled" and payload.get("trigger") == "auto") + ) # getattr-guarded like on_generation_claimed/on_aux_usage: a - # duck-typed SessionUI predating the hook gets no compaction events - # rather than an AttributeError that wedges every long session at - # its first auto-compaction. + # duck-typed SessionUI predating the hook must not hit an + # AttributeError that wedges every long session at its first + # auto-compaction. 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 + + render_compaction_event_as_info(event, info) return None - result = emit({"compaction_id": my_generation, "superseded": stale, **payload}) + 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. return result if isinstance(result, int) else None @@ -9043,9 +9079,10 @@ class ChatSession: queue from BEFORE their window (a dying send worker's closing race) — the /compact worker's exit seam is the caller. Messages sent DURING a command window never reach this queue: the /send route - parks them while ``worker_kind == "command"`` and dispatches them - as ordinary sends afterwards. Must only be called by the thread - that owns the worker slot — it mutates ``self.messages``. + defers them (``ws._pending_sends``) while ``worker_kind == + "command"`` and the drain task dispatches them as ordinary sends + afterwards. Must only be called by the thread that owns the + worker slot — it mutates ``self.messages``. """ return self._flush_queued_messages() @@ -17347,9 +17384,10 @@ class ChatSession: # Flush any stranded queued text BEFORE the identity swap so it # is persisted into the workstream it was ADDRESSED to. Sends - # during the command window itself park in the /send route and - # never queue; this covers only a message stranded by a dying - # send worker's closing race before this command started. + # during the command window itself defer in the /send route + # (ws._pending_sends) and never queue; this covers only a + # message stranded by a dying send worker's closing race + # before this command started. self._flush_queued_messages() self.messages.clear() self._read_files.clear() diff --git a/turnstone/core/session_routes.py b/turnstone/core/session_routes.py index 08c317f8..fbc527a5 100644 --- a/turnstone/core/session_routes.py +++ b/turnstone/core/session_routes.py @@ -103,8 +103,10 @@ AttachmentOwnerResolver = Callable[ ] # (request, ui) — kind's spawn-time bookkeeping. Interactive bumps # ``_metrics.record_message_sent`` + per-UI message counters; coord -# has no analog and wires ``None``. -SpawnMetricsHook = Callable[["Request", Any], None] +# has no analog and wires ``None``. The request is ``None`` when the +# pending-send drain dispatches a deferred entry (no live request +# exists by then) — both installed impls ignore the argument. +SpawnMetricsHook = Callable[["Request | None", Any], None] class CancelForensics(Protocol): @@ -1369,13 +1371,16 @@ def make_cancel_handler( # # Documented bet — force-cancelling a wedged QUICK command # (worker_kind == "command", e.g. /resume stuck in storage - # I/O): clearing the flag releases any parked /send, whose - # fresh worker can then run while the abandoned command - # thread finishes its in-place mutation — quick commands - # have no generation checkpoints to retire them (compact_now - # does). Same blast radius as force-abandoning a send - # worker mid-tool; accepted because force-cancel is the - # operator escape hatch for an already-wedged session, not a + # I/O): clearing the flag releases the pending-send + # drain's park (_drain_pending_sends polls the same + # (_worker_running, worker_kind) pair the parked /send + # used to), so a deferred message's fresh worker can then + # run while the abandoned command thread finishes its + # in-place mutation — quick commands have no generation + # checkpoints to retire them (compact_now does). Same + # blast radius as force-abandoning a send worker + # mid-tool; accepted because force-cancel is the operator + # escape hatch for an already-wedged session, not a # routine path. Revisit if commands ever gain generation # discipline. with ws._lock: @@ -3942,6 +3947,313 @@ def make_detail_handler(cfg: SessionEndpointConfig) -> Handler: return detail +# --------------------------------------------------------------------------- +# Deferred sends (command windows) +# --------------------------------------------------------------------------- + + +@dataclass +class _PendingSend: + """One send deferred during a command window. + + 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. + + 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. + """ + + msg_id: str + priority: str + attempt: Callable[[ChatSession], tuple[bool, dict[str, Any]]] + retracted: bool = False + + +def _emit_send_ui(ws: Workstream, ui: Any, hook_name: str, *args: Any) -> None: + """Best-effort UI hook dispatch for send-worker closures. + + Each call is wrapped in try/except so a failure in one hook (e.g. + listener-queue full → on_error raises) doesn't suppress the others. + Mirrors the pre-P1.5 coord_adapter.send per-hook defense. + """ + if ui is None: + return + method = getattr(ui, hook_name, None) + if method is None: + return + try: + method(*args) + except Exception: + log.debug( + "ws.send.ui_hook_failed ws=%s hook=%s", + ws.id[:8] if ws.id else "", + hook_name, + exc_info=True, + ) + + +def _make_dispatch_attempt( + ws: Workstream, + cfg: SessionEndpointConfig, + ui: Any, + *, + message: str, + resolved_atts: list[Any], + ordered_taken: list[str], + send_id: str, + acting_uid: str, + request: Request | None, + defer_fidelity: bool = False, +) -> Callable[[ChatSession], tuple[bool, dict[str, Any]]]: + """Build one atomic queue-or-spawn attempt bound to ONE session capture. + + The single dispatch implementation shared by the /send route's + immediate path and :func:`_drain_pending_sends` — session re-capture + across /resume//new identity swaps, the cross-user and attachment + queue guards, ``send_id`` threading (the queue path reuses it as + ``queue_msg_id`` so the client's DELETE targets one id either way), + and the spawn-path metrics all live here, once. Callers re-capture + ``ws.session`` before every attempt and pass it in: closures bound + to a pre-swap capture would send the user's message into the wrong + workstream's transcript. + + ``queue_outcome`` (second element of the return) is written only + when the dispatcher takes the live-worker reuse path; empty after a + fresh-spawn dispatch. + + ``defer_fidelity=True`` marks a deferred entry's attempt: it was + answered "queued" under the full-fidelity defer contract, so the + interjection fallback — which truncates at ``queue_message``'s + 2000-char cap and cannot carry attachments — is refused for + oversized or attachment-bearing entries (the drain waits for the + slot and retries into the fresh-spawn arm instead). The refusal + happens inside the enqueue callback, under the same ``ws._lock`` + acquisition as the queue-vs-spawn decision, so a turn claiming the + slot between the drain's poll and this dispatch can never route the + entry into truncation. + + ``request`` is ``None`` for drain-side attempts; the spawn-metrics + hook tolerates it (both installed impls ignore the argument). + """ + import threading + + from turnstone.core import session_worker + from turnstone.core.session import ( + AttachmentsNotQueueableError, + CrossUserInterjectionError, + GenerationCancelled, + ) + + def attempt(session: ChatSession) -> tuple[bool, dict[str, Any]]: + queue_outcome: dict[str, Any] = {} + + def _enqueue() -> None: + # Runs under ``ws._lock`` (session_worker.send calls it inside + # the same acquisition that reads _worker_running), so this + # worker_kind read cannot race the spawn write. A command + # window must NEVER reach queue_message — its cap and + # cross-user guard are turn semantics — so report it and let + # the route defer (or the drain re-park). + if ws.worker_kind == "command": + queue_outcome["rejected"] = "command_window" + return + if defer_fidelity and (resolved_atts or len(message) > 2000): + queue_outcome["rejected"] = "defer_full_fidelity" + return + try: + cleaned, priority, msg_id = session.queue_message( + message, + attachment_ids=list(ordered_taken), + queue_msg_id=send_id or None, + interjector_user_id=acting_uid, + ) + except AttachmentsNotQueueableError: + queue_outcome["rejected"] = "attachments_busy" + return + except CrossUserInterjectionError: + # A different authenticated participant tried to interject + # into someone else's in-flight turn; folding it in would + # borrow the initiator's credentials and misattribute the + # message. Reject so they resend as a fresh turn once the + # worker idles (the drain instead waits and re-attempts). + queue_outcome["rejected"] = "cross_user_interjection" + return + queue_outcome["cleaned"] = cleaned + queue_outcome["priority"] = priority + queue_outcome["msg_id"] = msg_id + + def _run() -> None: + me = threading.current_thread() + try: + kwargs: dict[str, Any] = {} + if resolved_atts: + kwargs["attachments"] = resolved_atts + if send_id: + kwargs["send_id"] = send_id + # Fresh turn: rebind per-user MCP credentials to the + # authenticated sender. Bound here (not via a send() + # kwarg) so per-kind session stubs with explicit send + # signatures keep working; getattr-guarded for the same + # reason. The queue path above never rebinds. + bind = getattr(session, "bind_acting_user", None) + if acting_uid and callable(bind): + bind(acting_uid) + session.send(message, **kwargs) + except GenerationCancelled: + # Safety net — send() normally handles this internally. + # If this thread was force-abandoned, ws.worker_thread + # was set to None — don't emit spurious events. + if ws.worker_thread is me: + _emit_send_ui(ws, ui, "on_stream_end") + _emit_send_ui(ws, ui, "on_state_change", "idle") + except Exception: + # Undrained staged uploads aren't locked (the buffer is a + # peek, not a reservation) — they expire on the buffer TTL + # — so the only cleanup owed here is the UI streaming + # hook: ``session.send()`` already fired ``on_error`` + # (with sanitized text), persisted ``last_error``, and + # emitted ``state='error'`` via + # :meth:`ChatSession._record_fatal_error` before + # re-raising. + if ws.worker_thread is me: + _emit_send_ui(ws, ui, "on_stream_end") + + ok = session_worker.send( + ws, + enqueue=_enqueue, + run=_run, + thread_name=f"send-worker-{ws.id[:8]}", + ) + if ok and not queue_outcome and cfg.spawn_metrics is not None: + # Fresh spawn — the kind's per-turn metrics fire exactly once, + # from the shared attempt so the drain's dispatches count too. + try: + cfg.spawn_metrics(request, ui) + except Exception: + log.debug( + "ws.send.spawn_metrics_failed ws=%s", + ws.id[:8] if ws.id else "", + exc_info=True, + ) + return ok, queue_outcome + + return attempt + + +def _drain_pending_sends(ws: Workstream) -> None: + """Dispatch a workstream's deferred sends once its command window closes. + + Per-workstream single-flight (``ws._pending_drain``), started by the + /send route when it defers an entry and run on a small daemon thread + (the dispatch machinery is synchronous and thread-shaped like every + other worker here, and a thread's lifetime is independent of any + event loop's — the request loop owes this drain nothing once the + route has answered). It owns the waiting the parked POST used to do + — but server-side, so a client timeout or abort can no longer become + message loss. Entries dispatch in arrival order via their prebuilt + attempt closures, re-capturing ``ws.session`` per attempt (a /resume + or /new that swapped the session mid-window routes the message into + the post-swap session, exactly as the park did). + + Terminal outcomes per entry: dispatched (fresh spawn — or, for a + queue-shaped entry, the interjection fallback into a live turn, + msg_id preserved so the client's DELETE still targets it), + retracted (dismissed before dispatch), or dropped because the + workstream closed. There is deliberately no give-up bound: an entry + acknowledged "queued" is never silently dropped while the workstream + lives — rejections wait for the slot to free and retry into the + fresh-spawn arm. + + Claim discipline: an entry is popped under ``ws._lock`` immediately + before its dispatch attempt and re-inserted on rejection, so the + DELETE fall-through (which marks only in-list entries) can never + "remove" a message whose dispatch already left the station — a + claimed entry answers ``not_found`` ("already sent"), which its + eventual dispatch makes true. + """ + import time + + try: + while True: + with ws._lock: + pending = ws._pending_sends + while pending and pending[0].retracted: + pending.pop(0) + if ws._closed: + if pending: + log.warning( + "ws.send.pending_dropped_on_close ws=%s count=%d", + ws.id[:8], + len(pending), + ) + pending.clear() + ws._pending_drain = None + return + if not pending: + ws._pending_drain = None + return + entry = pending[0] + if ws._worker_running and ws.worker_kind == "command": + # The park, relocated server-side: the poll cadence + # matches the old request-handler loop's. + time.sleep(0.25) + continue + session_now = ws.session + if session_now is None: + # Mid-swap / partial-construction gap; the loop-top close + # check terminates this if it's a close in progress. + time.sleep(0.25) + continue + with ws._lock: + if not ws._pending_sends or ws._pending_sends[0] is not entry: + continue # list reshaped under us — re-evaluate + if entry.retracted: + ws._pending_sends.pop(0) + continue + ws._pending_sends.pop(0) # claim + ok, outcome = entry.attempt(session_now) + if not ok or outcome.get("rejected") in ( + "command_window", + "defer_full_fidelity", + "attachments_busy", + "cross_user_interjection", + ): + # Window re-claimed between the poll and the dispatch, a + # live turn holds the slot against a full-fidelity entry, + # the interjection queue is saturated + # (session_worker.send → False on queue.Full), or the ws + # closed (resolved at the loop top). Unclaim and wait. + with ws._lock: + ws._pending_sends.insert(0, entry) + time.sleep(0.25) + continue + # Dispatched: fresh spawn (empty outcome) or interjection + # fallback (msg_id preserved) — this entry is done. + except Exception: + # Never die holding the single-flight slot — a wedged drain would + # strand every future deferred send for this workstream. + log.exception("ws.send.pending_drain_failed ws=%s", ws.id[:8] if ws.id else "") + with ws._lock: + ws._pending_drain = None + + def make_send_handler(cfg: SessionEndpointConfig) -> Handler: """Lifted body for ``POST {prefix}/{ws_id}/send`` — message dispatch. @@ -3977,8 +4289,13 @@ def make_send_handler(cfg: SessionEndpointConfig) -> Handler: requested attachments that landed (may be a strict subset on reservation race losses). - 200 ``{"status": "queued", "priority", "msg_id", "attached_ids", - "dropped_attachment_ids"}`` — reused live worker; queued for - injection at the next tool-result seam. + "dropped_attachment_ids"}`` — reused live worker (queued for + injection at the next tool-result seam), OR deferred during a + slash-command window (registered on ``ws._pending_sends`` and + dispatched full-fidelity by :func:`_drain_pending_sends` when the + window closes — see :class:`_PendingSend` for the durability + contract). ``DELETE {prefix}/{ws_id}/send`` with the ``msg_id`` + retracts either kind before dispatch. - 200 ``{"status": "queue_full", "attached_ids", "dropped_attachment_ids"}`` — live worker's queue at capacity; reservations released. Caller should retry. The @@ -3991,12 +4308,7 @@ def make_send_handler(cfg: SessionEndpointConfig) -> Handler: import threading import uuid - from turnstone.core import session_worker - from turnstone.core.session import ( - AttachmentsNotQueueableError, - CrossUserInterjectionError, - GenerationCancelled, - ) + from turnstone.core.tool_advisory import parse_priority from turnstone.core.web_helpers import auth_user_id, read_json_or_400 async def send(request: Request) -> Response: @@ -4093,162 +4405,94 @@ def make_send_handler(cfg: SessionEndpointConfig) -> Handler: if ws.session is None: return JSONResponse({"error": "No session"}, status_code=500) - def _dispatch_once(session: ChatSession) -> tuple[bool, dict[str, Any]]: - """One atomic queue-or-spawn attempt bound to ONE session capture. - - The park loop below re-captures ``ws.session`` before every - attempt — a /resume or /new that ran while we parked swaps the - session's workstream identity in place, and closures bound to - the pre-swap capture would send the user's message into the - wrong workstream's transcript. Binding happens here, in one - place, per attempt. ``queue_outcome`` is written only when the - dispatcher takes the live-worker reuse path; empty after a - fresh-spawn dispatch. - """ - queue_outcome: dict[str, Any] = {} - - def _enqueue() -> None: - # Runs under ``ws._lock`` (session_worker.send calls it - # inside the same acquisition that reads _worker_running), - # so this worker_kind read cannot race the spawn write. A - # command window must NEVER reach queue_message — its cap - # and cross-user guard are turn semantics — so refuse and - # let the route loop back to the park. - if ws.worker_kind == "command": - queue_outcome["rejected"] = "command_window" - return - try: - cleaned, priority, msg_id = session.queue_message( - message, - attachment_ids=list(ordered_taken), - queue_msg_id=send_id or None, - interjector_user_id=acting_uid, - ) - except AttachmentsNotQueueableError: - queue_outcome["rejected"] = "attachments_busy" - return - except CrossUserInterjectionError: - # A different authenticated participant tried to interject into - # someone else's in-flight turn; folding it in would borrow the - # initiator's credentials and misattribute the message. Reject - # so they resend as a fresh turn once the worker idles. - queue_outcome["rejected"] = "cross_user_interjection" - return - queue_outcome["cleaned"] = cleaned - queue_outcome["priority"] = priority - queue_outcome["msg_id"] = msg_id - - def _run() -> None: - me = threading.current_thread() - try: - kwargs: dict[str, Any] = {} - if resolved_atts: - kwargs["attachments"] = resolved_atts - if send_id: - kwargs["send_id"] = send_id - # Fresh turn: rebind per-user MCP credentials to the - # authenticated sender. Bound here (not via a send() - # kwarg) so per-kind session stubs with explicit send - # signatures keep working; getattr-guarded for the same - # reason. The queue path above never rebinds. - bind = getattr(session, "bind_acting_user", None) - if acting_uid and callable(bind): - bind(acting_uid) - session.send(message, **kwargs) - except GenerationCancelled: - # Safety net — send() normally handles this internally. - # If this thread was force-abandoned, ws.worker_thread - # was set to None — don't emit spurious events. - if ws.worker_thread is me: - _emit_ui("on_stream_end") - _emit_ui("on_state_change", "idle") - except Exception: - # Undrained staged uploads aren't locked (the buffer is a peek, - # not a reservation) — they expire on the buffer TTL — so the - # only cleanup owed here is the UI streaming hook. - if ws.worker_thread is me: - # ``session.send()`` already fired ``on_error`` - # (with sanitized text), persisted ``last_error``, - # and emitted ``state='error'`` via - # :meth:`ChatSession._record_fatal_error` before - # re-raising. The route handler only needs the - # streaming-cleanup hook the worker contract owes - # the UI listeners. - _emit_ui("on_stream_end") - - ok = session_worker.send( - ws, - enqueue=_enqueue, - run=_run, - thread_name=f"send-worker-{ws.id[:8]}", - ) - return ok, queue_outcome - - def _emit_ui(hook_name: str, *args: Any) -> None: - """Best-effort UI hook dispatch. - - Each call is wrapped in try/except so a failure in one - hook (e.g. listener-queue full → on_error raises) doesn't - suppress the others. Mirrors the pre-P1.5 - coord_adapter.send per-hook defense. Defined after - ``_dispatch_once`` but resolved late (when a worker runs) — - every dispatch happens strictly below this line. - """ - if ui is None: - return - method = getattr(ui, hook_name, None) - if method is None: - return - try: - method(*args) - except Exception: - log.debug( - "ws.send.ui_hook_failed ws=%s hook=%s", - ws.id[:8] if ws.id else "", - hook_name, - exc_info=True, - ) - - # Park-then-dispatch. While a slash-command worker holds the slot - # (a manual /compact can hold it for MINUTES), a send must not take + # Defer-and-drain. While a slash-command worker holds the slot (a + # manual /compact can hold it for MINUTES), a send must not take # the interjection-queue path — its 2000-char cap and cross-user # guard are mid-TURN semantics, and a queued message would cross a - # /resume//new identity swap into the wrong workstream. Parking - # reproduces the pre-worker-dispatch observable behavior (the - # request waited out the then-blocked event loop, then ran as a - # normal full-fidelity send under the sender's own identity) with - # the loop free: SSE keeps streaming the compaction progress card - # while we wait. The inner park is deliberately unbounded — the - # composer bounds its POST at ~15s and aborts, and we poll - # ``is_disconnected`` because starlette does NOT cancel a running - # handler on client disconnect: dispatching after the client gave - # up would surprise the user with a duplicate turn when they - # resend. The outer bound only caps command-window FLAPPING - # (back-to-back commands re-claiming the slot between our park - # exit and dispatch). - ok = False - queue_outcome: dict[str, Any] = {} - for _ in range(20): - while ws._worker_running and ws.worker_kind == "command": - if await request.is_disconnected(): - # Client abandoned the send mid-park; the response is - # discarded — the status is for the access log only. - return JSONResponse({"status": "abandoned"}) - await asyncio.sleep(0.25) - session_now = ws.session - if session_now is None: - return JSONResponse({"error": "No session"}, status_code=500) - ok, queue_outcome = _dispatch_once(session_now) - if queue_outcome.get("rejected") != "command_window": - break - else: - # 20 consecutive command-window collisions — surface the same - # retryable backpressure shape as a saturated queue. + # /resume//new identity swap into the wrong workstream. Instead + # of parking THIS request until the window closes (which encoded + # "client disconnected" as "message retracted" — deterministic + # message loss for every bounded caller: the coordinator client + # and console proxy time out at 30s, and the web composers' long + # abort bound raced the compaction card), the send is answered + # "queued" immediately and registered on ``ws._pending_sends``; + # the per-workstream drain task dispatches it full-fidelity when + # the window closes. Dismissal is the same DELETE /send + # {msg_id} the interjection queue uses — server-confirmed, no + # POST-abort side channel. + attempt = _make_dispatch_attempt( + ws, + cfg, + ui, + message=message, + resolved_atts=resolved_atts, + ordered_taken=ordered_taken, + send_id=send_id, + acting_uid=acting_uid, + request=request, + ) + session_now = ws.session + if session_now is None: + return JSONResponse({"error": "No session"}, status_code=500) + ok, queue_outcome = attempt(session_now) + if queue_outcome.get("rejected") == "command_window": + # ``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, + priority=pending_priority, + 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, + ), + ) + with ws._lock: + 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) + ws._pending_sends.append(entry) + drain = ws._pending_drain + if drain is None or not drain.is_alive(): + t = threading.Thread( + target=_drain_pending_sends, + args=(ws,), + name=f"pending-drain-{ws.id[:8]}", + daemon=True, + ) + ws._pending_drain = t + t.start() + if cfg.emit_message_queued and hasattr(ui, "_enqueue"): + ui._enqueue( + { + "type": "message_queued", + "message": cleaned_display, + "priority": pending_priority, + "msg_id": pending_msg_id, + } + ) return JSONResponse( { - "status": "queue_full", - "attached_ids": [], - "dropped_attachment_ids": list(requested_ids), + "status": "queued", + "priority": pending_priority, + "msg_id": pending_msg_id, + "attached_ids": list(ordered_taken), + "dropped_attachment_ids": [ + aid for aid in requested_ids if aid not in taken_set + ], } ) if not ok: @@ -4324,16 +4568,9 @@ def make_send_handler(cfg: SessionEndpointConfig) -> Handler: } ) - # Spawned a fresh worker — kind's metrics fire once per turn. - if cfg.spawn_metrics is not None: - try: - cfg.spawn_metrics(request, ui) - except Exception: - log.debug( - "ws.send.spawn_metrics_failed ws=%s", - ws_id[:8] if ws_id else "", - exc_info=True, - ) + # Spawned a fresh worker — the kind's per-turn metrics fired inside + # the shared attempt (see _make_dispatch_attempt), where the drain + # task's dispatches fire them too. return JSONResponse( { "status": "ok", @@ -4633,12 +4870,17 @@ def make_attachment_handlers(cfg: SessionEndpointConfig) -> AttachmentHandlers: def make_dequeue_handler(cfg: SessionEndpointConfig) -> Handler: """Lifted body for ``DELETE {prefix}/{ws_id}/send`` — cancel a queued message. - Removes a previously-queued message identified by ``msg_id`` from - the workstream's pending queue. Returns ``status: removed`` when - the queue had the entry and ``status: not_found`` otherwise. - Queued messages don't carry attachments (see - :class:`AttachmentsNotQueueableError`), so there's no reservation - side-effect to undo here. + Removes a previously-queued message identified by ``msg_id`` — + first from the session's interjection queue, then (fall-through) + from the workstream's deferred-send list (``ws._pending_sends``, + sends answered "queued" during a command window). Returns + ``status: removed`` when either held the entry and + ``status: not_found`` otherwise. Interjection-queued messages + don't carry attachments (see :class:`AttachmentsNotQueueableError`) + and a deferred entry's resolved attachments die with it (the staged + bytes were peeked, not reserved — they expire on the buffer TTL), + so there's no reservation side-effect to undo here; the client + surfaces the discarded-attachments consequence to the user. """ from turnstone.core.web_helpers import read_json_or_400 @@ -4677,6 +4919,21 @@ def make_dequeue_handler(cfg: SessionEndpointConfig) -> Handler: if ws.session is None: return JSONResponse({"error": "No session"}, status_code=400) removed = ws.session.dequeue_message(msg_id) + if not removed: + # Fall through to the deferred-send list: a send answered + # "queued" during a command window lives on the workstream + # (see _PendingSend), not in the session's interjection + # queue. Marked under the same lock the drain claims under, + # so a retracted entry can never dispatch; an entry the + # drain already claimed is gone from the list and correctly + # answers not_found ("already sent" — its dispatch is in + # flight). + with ws._lock: + for entry in ws._pending_sends: + if entry.msg_id == msg_id and not entry.retracted: + entry.retracted = True + removed = True + break return JSONResponse({"status": "removed" if removed else "not_found"}) return dequeue diff --git a/turnstone/core/session_worker.py b/turnstone/core/session_worker.py index b8d2e2cc..89bb3ccb 100644 --- a/turnstone/core/session_worker.py +++ b/turnstone/core/session_worker.py @@ -113,24 +113,29 @@ def send( ``ws.worker_kind`` in the same lock acquisition as the ``(worker_thread, _worker_running)`` pair. This is the ONLY site that sets ``_worker_running=True``, so the classification cannot be - bypassed by a new dispatch caller. The /send route parks while a + bypassed by a new dispatch caller. The /send route DEFERS while a command holds the slot instead of taking the interjection-queue - path (whose length cap / cross-user guard are turn semantics); an - ``enqueue`` callback that can fire during a command window (the - coordinator adapter's, the init race's) must refuse rather than - queue — see the command-window refusals at those closures. + path (whose length cap / cross-user guard are turn semantics): its + enqueue closure reports the window and the route registers the send + on ``ws._pending_sends`` for the drain task to dispatch when the + window closes. An ``enqueue`` callback that can fire during a + command window (the coordinator adapter's, the init race's) must + refuse rather than queue — see the command-window refusals at those + closures. - The refusal is DELIBERATELY not centralized here despite the three + The refusal is DELIBERATELY not centralized here despite the hand-written guards: each surface needs a different refusal channel - (the /send route signals "re-park" via its ``queue_outcome`` flag; - the coordinator adapter and the init race raise ``queue.Full`` into - their existing backpressure statuses), and a central refusal inside - this function can only return ``False`` — indistinguishable from - queue-full/closed for the route's re-park decision and mislabeled by - the init path's status derivation. Making it distinguishable means - a tri-state contract change across every dispatch caller, which is - more surface than three four-line guards. If you add a NEW enqueue - closure that can queue turn work, copy the guard. + (the /send route's closure signals "command window" via its + ``queue_outcome`` flag — the defer trigger; the coordinator adapter + and the init race raise ``queue.Full`` into their existing + backpressure statuses), and a central refusal inside this function + can only return ``False`` — indistinguishable from queue-full/closed + exactly where the route must distinguish "defer this" from "drop + this". Making it distinguishable means a tri-state contract change + across every dispatch caller — more surface than the guards it + replaces. (A check inside this function's locked section would be + race-free; the cost is the contract change, not atomicity.) If you + add a NEW enqueue closure that can queue turn work, copy the guard. Returns: ``True`` on successful enqueue (existing worker accepted) or diff --git a/turnstone/core/workstream.py b/turnstone/core/workstream.py index d42134ed..2f55ebf9 100644 --- a/turnstone/core/workstream.py +++ b/turnstone/core/workstream.py @@ -13,7 +13,7 @@ import threading import time import uuid from dataclasses import dataclass, field -from typing import TYPE_CHECKING +from typing import TYPE_CHECKING, Any if TYPE_CHECKING: from turnstone.core.session import ChatSession, SessionUI @@ -146,11 +146,24 @@ class Workstream: # sets ``worker_thread``/``_worker_running``, so readers gating on # the running flag see a coherent triple. A stale value after the # worker exits is harmless — every reader conjoins - # ``_worker_running``. The /send route parks (never queues) while + # ``_worker_running``. The /send route defers (never queues) while # this reads "command": the mid-turn interjection queue is # turn-shaped (length cap, cross-user guard) and must be - # unreachable during command windows. + # 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 + # ``_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) + # 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) # 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 diff --git a/turnstone/sdk/events.py b/turnstone/sdk/events.py index 254b4430..724adf46 100644 --- a/turnstone/sdk/events.py +++ b/turnstone/sdk/events.py @@ -314,13 +314,17 @@ class CompactionEvent(ServerEvent): events additionally carry ``superseded``: True marks a force-abandoned compaction retiring after a successor generation took over — clients should skip failure notices for those (an OK end's result still - stands; the history swap happened). + stands; the history swap happened). Failed ends carry ``notice``: + the emitter-computed display verdict — show ``message`` only when it + is True, instead of re-deriving suppression from + reason/trigger/superseded client-side. """ type: str = "compaction" phase: str = "" compaction_id: int = 0 superseded: bool = False + notice: bool = False trigger: str = "" where: str = "" pct: int | None = None diff --git a/turnstone/server.py b/turnstone/server.py index ad41b5a6..3a2d7ee3 100644 --- a/turnstone/server.py +++ b/turnstone/server.py @@ -1737,11 +1737,13 @@ async def command(request: Request) -> JSONResponse: One envelope for both command branches: the unknown-workstream 404 and the busy refusal differ only in the retry hint. The worker slot is claimed with ``worker_kind="command"`` — the - /send route PARKS (never queues) while that kind holds the + /send route DEFERS (never queues) while that kind holds the slot, so the mid-turn interjection queue and its turn-shaped semantics (length cap, cross-user guard) are unreachable for - the whole command window; messages sent mid-command dispatch - as ordinary full-fidelity sends when the window closes. + the whole command window; messages sent mid-command are + answered ``queued`` immediately and dispatched as ordinary + full-fidelity sends by the pending-send drain when the window + closes. """ dispatched = session_worker.send( ws, @@ -1790,6 +1792,15 @@ async def command(request: Request) -> JSONResponse: def _run_compact() -> None: me = threading.current_thread() + # Snapshot for the exit restore: with the slot free to + # claim, only IDLE or ERROR are reachable here (the live + # states imply a held slot and _dispatch_command refuses + # busy). ERROR is the user-investigatable badge (same + # carve-out the orphan reaper honors) and /compact neither + # retries nor resolves the failed turn — the old inline + # /compact never touched ws.state — so the badge must + # survive the window; everything else exits to idle. + prev_state = ws.state try: cmd_ui.on_state_change("thinking") session.compact_now() @@ -1805,7 +1816,7 @@ async def command(request: Request) -> JSONResponse: if ws.worker_thread is me: try: # Backstop only: sends during the command window - # PARK in the /send route (they cannot reach the + # DEFER in the /send route (they cannot reach the # interjection queue — see _dispatch_command), so # anything found here predates the window (a # message stranded by a dying send worker's @@ -1815,7 +1826,9 @@ async def command(request: Request) -> JSONResponse: log.warning("ws.compact.stranded_queue_flushed ws=%s", ws.id[:8]) except Exception: log.exception("ws.compact.exit_seam_failed ws=%s", ws.id[:8]) - cmd_ui.on_state_change("idle") + cmd_ui.on_state_change( + "error" if prev_state is WorkstreamState.ERROR else "idle" + ) refusal = _dispatch_command( _run_compact, f"compact-worker-{ws.id[:8]}", "run /compact again" @@ -1875,11 +1888,12 @@ async def command(request: Request) -> JSONResponse: # Unblock the endpoint response. Suppress the loop-closed # RuntimeError (process shutdown mid-command): nobody is # waiting anymore. No queue drain here: sends during the - # command window PARK in the /send route (the interjection + # command window DEFER in the /send route (the interjection # queue is unreachable while worker_kind == "command"), so - # they dispatch as ordinary sends when this worker exits — - # the stranded-message backstop lives on the /compact seam, - # the only command window long enough to matter. + # the pending-send drain dispatches them as ordinary sends + # when this worker exits — the stranded-message backstop + # lives on the /compact seam, the only command window long + # enough to matter. with contextlib.suppress(RuntimeError): loop.call_soon_threadsafe(done.set) @@ -2458,15 +2472,23 @@ async def _interactive_create_post_install( ws.ui.on_stream_end() ws.ui.on_state_change("idle") finally: - # Owner only — an abandoned init's late notify would - # duplicate the successor's. (Guard, not early-return: a - # return in a finally silences in-flight exceptions.) - if ws.worker_thread is me: - try: - last_content = _extract_last_assistant_content(session) - _fire_notify_targets(ws, last_content) - except Exception: - log.warning("notify_completion.hook_error", ws_id=ws.id, exc_info=True) + # Deliberately NOT owner-gated, unlike the except arm above + # and the _run_cmd/_run_compact follow-ups: those mutate + # live slot/UI state a successor now owns, while the notify + # is workstream-scoped — an outward signal that this + # workstream's initial turn ran to completion and its + # answer is in the transcript. _fire_notify_targets has + # exactly ONE call site (here); successor turns never + # notify, so there is no successor duplicate for an owner + # gate to prevent — gating it turned force-cancel into + # permanent notification loss for scheduled/unattended + # workstreams (the empty-content fallback covers the + # error/cancel exits, as it always did). + try: + last_content = _extract_last_assistant_content(session) + _fire_notify_targets(ws, last_content) + except Exception: + log.warning("notify_completion.hook_error", ws_id=ws.id, exc_info=True) init_enqueued = False @@ -2492,7 +2514,7 @@ async def _interactive_create_post_install( # raced slot: the interjection queue is turn-shaped (length # cap, and the text would cross a /resume identity swap) and # must stay unreachable during command windows — same rule - # as the /send route's park. queue.Full is the existing + # as the /send route's defer. queue.Full is the existing # backpressure surface: the create reports the first message # as undelivered (``queue_full``) and the client retries. raise queue.Full() @@ -4497,7 +4519,7 @@ def create_app( magic-mocked ``ws.user_id``.""" return _require_ws_access(request, ws_id) - def _interactive_spawn_metrics(_request: Request, ui: Any) -> None: + def _interactive_spawn_metrics(_request: Request | None, ui: Any) -> None: """Per-conversation metrics fired once per send that spawns a fresh worker. Coord wires its own :func:`turnstone.console.server._coord_spawn_metrics` (the diff --git a/turnstone/shared_static/composer_queue.js b/turnstone/shared_static/composer_queue.js index 3093913b..85418926 100644 --- a/turnstone/shared_static/composer_queue.js +++ b/turnstone/shared_static/composer_queue.js @@ -265,6 +265,16 @@ export function createQueueController(opts) { var status = data && data.status; if (status === "removed") { el.remove(); + // A deferred send (command-window defer) can carry attachments; + // retracting it discards them — the chips were consumed when the + // send was accepted and a retract does not re-stage the bytes. + // Say so instead of letting the files silently expire. + if (el._deferredAttachments > 0 && onNotice) + onNotice( + "Message removed. Its " + + el._deferredAttachments + + " attachment(s) were discarded — re-attach them to send again.", + ); // `removed` is the only verdict that mutated server-side queue // state, so it's the only one worth re-syncing composer state for. if (onAfterDequeue) onAfterDequeue(); @@ -297,11 +307,6 @@ export function createQueueController(opts) { if (el.getAttribute("aria-busy") === "true") return; el.dataset.dismissAttempted = "1"; _setDismissing(el, true); - // A send whose POST is still in flight (parked server-side during a - // command window — possibly for minutes) attaches an abort hook: the × - // must kill the request itself, or the dismissed message dispatches - // anyway when the window closes. Post-response the hook is a no-op. - if (typeof el._sendAbort === "function") el._sendAbort(); var msgId = el.dataset.msgId; if (!msgId) { // Pre-bind: bind() confirms once the server returns the msg_id (it diff --git a/turnstone/shared_static/conversation.js b/turnstone/shared_static/conversation.js index 5591022b..d50ddc32 100644 --- a/turnstone/shared_static/conversation.js +++ b/turnstone/shared_static/conversation.js @@ -246,11 +246,7 @@ export function applyCompactionEvent(holder, evt, hooks) { return; } if (evt.phase === "end") { - if (owns && holder.card) { - holder.card.remove(); - holder.card = null; - holder.cid = null; - } + if (owns) resetCompactionHolder(holder); if (evt.ok) { // The persisted marker row is stamped with THIS event's id, so // whichever of /history repaint or live/replayed event renders @@ -269,48 +265,27 @@ export function applyCompactionEvent(holder, evt, hooks) { ), ); if (eid) hooks.renderedIds.add(eid); - } else if ( - owns && - !evt.superseded && - evt.reason !== "error" && - !(evt.reason === "cancelled" && evt.trigger === "auto") - ) { + } else if (owns && evt.notice) { // cancelled / not_enough_messages / irreducible / empty_summary — - // informational, not an error state. Three suppressions: a stale - // end's notice would narrate a dead compaction underneath a live - // one; a superseded end (a force-abandoned compaction retiring - // late, flagged by the backend) is one nobody is waiting on — its - // notice mid-turn reads as the LIVE work being cancelled; and an - // auto-compaction's cancel is just part of cancelling the - // surrounding turn — the send loop emits its own "[Generation - // cancelled]" info line, so a second line here stacked two notices - // for one Stop click. (A superseded OK end above still renders - // its result card — the history swap really happened.) - // HAND-SYNCED SIBLING: cli.py TerminalUI.on_compaction implements - // this same three-clause policy for the terminal — change both or - // they drift (two runtimes, no shared code path). + // informational, not an error state. Whether the message is shown + // is the emitter's call: the backend stamps `notice` on failed ends + // (suppressing error-reason / superseded / cancelled-auto ends — + // see ChatSession._compaction_event, the single policy site), so + // this arm stays mechanical. `owns` is the one pane-local clause + // the emitter cannot compute — card ownership via compaction_id vs + // holder.cid — and it guards a reachable divergence: a /resume + // swaps sessions and restarts generation counters, so an abandoned + // old-session end can arrive superseded=false against the new + // session's live card. An end without the field (replayed from an + // older node) stays silent — these notices are informational. + // (A superseded OK end above still renders its result card — the + // history swap really happened.) hooks.onNotice(evt.message || "Compaction skipped."); } hooks.scroll(true); } } -// Send-POST abort bound for a pane, selected off its compaction holder. -// Sends during a slash-command window PARK server-side and dispatch when -// the window closes — a manual /compact legitimately runs for minutes, so -// while ITS progress card is live the composer must not abort the parked -// POST at the wedged-node default (~15s) and silently drop the message. -// The card is the one cross-tab signal that a long window is in progress -// (SSE-driven via applyCompactionEvent); with no card the short default -// stands — a WEDGED quick command past 15s should fail loudly, not hang -// the composer for 10 minutes. Deployment note: reverse proxies bound -// the effective park at their own read timeout (documented in the API -// reference). Shared by the interactive composer and the coordinator -// viewer so the policy can't drift between panes. -export function sendAbortMs(holder) { - return holder && holder.card ? 600000 : 15000; -} - // Retire a pane's in-progress compaction card (if any) and clear the // holder. The teardown half of the lifecycle the reducer above owns — // exported so the panes' stream_end handlers (force-stop abandons the diff --git a/turnstone/shared_static/interactive.js b/turnstone/shared_static/interactive.js index a3e283f6..bfbb8fe6 100644 --- a/turnstone/shared_static/interactive.js +++ b/turnstone/shared_static/interactive.js @@ -26,7 +26,6 @@ import { buildCompactionCard, applyCompactionEvent, resetCompactionHolder, - sendAbortMs, buildSystemNudgeMarker, buildConvBatchShell, buildConvRow, @@ -269,6 +268,12 @@ class Pane { // (conversation.applyCompactionEvent); `card` is the in-progress card // between start and end, nulled wherever the transcript DOM is wiped. this._compaction = { card: null, cid: null }; + // Rendered-event-id dedup for system turns and compaction markers + // (/history repaint vs live/replayed event — whichever renders first + // wins). replayHistory assigns a FRESH Set on every transcript wipe + // so pre-wipe ids can't suppress re-painted rows; this is the + // first-load init. + this._renderedSystemEventIds = new Set(); this._retryHolderEl = null; this._toolRowIndex = new Map(); this._streamElIndex = new Map(); @@ -560,9 +565,6 @@ class Pane { // repaint or live/replayed event renders first wins. reason="error" // ends render through the paired `error` event (red row), not here. this.removeEmptyState(); - if (!this._renderedSystemEventIds) { - this._renderedSystemEventIds = new Set(); - } applyCompactionEvent(this._compaction, evt, { container: this.messagesEl, renderedIds: this._renderedSystemEventIds, @@ -1918,11 +1920,7 @@ class Pane { // the resume-cursor fix this shouldn't recur, but the guard keeps the // /history+replay seam idempotent for system turns regardless. const sysEid = evt._event_id != null ? String(evt._event_id) : null; - if ( - sysEid && - this._renderedSystemEventIds && - this._renderedSystemEventIds.has(sysEid) - ) { + if (sysEid && this._renderedSystemEventIds.has(sysEid)) { break; } this.addSystemContext( @@ -1930,11 +1928,7 @@ class Pane { evt.source || "", evt.meta || null, ); - if (sysEid) { - if (!this._renderedSystemEventIds) - this._renderedSystemEventIds = new Set(); - this._renderedSystemEventIds.add(sysEid); - } + if (sysEid) this._renderedSystemEventIds.add(sysEid); break; } @@ -3769,17 +3763,12 @@ class Pane { let sendTimer = null; if (sendCtrl) { sendInit.signal = sendCtrl.signal; - // Long bound while a compaction card is live (the send is parked - // server-side for the command window); wedged-node default - // otherwise — see sendAbortMs. - sendTimer = setTimeout( - () => sendCtrl.abort(), - sendAbortMs(this._compaction), - ); - // An × on the queued bubble BEFORE the response arrives (pre-bind) - // must also kill the parked POST — otherwise the dismissed message - // dispatches anyway when the command window closes, minutes later. - if (queuedEl) queuedEl._sendAbort = () => sendCtrl.abort(); + // Flat wedged-node bound: every /send answers within RTT now — + // dispatched, queued, or deferred-with-msg_id during a command + // window (the server parks nothing against this POST), so a + // response slower than this means a wedged node, not a long + // command. Dismissal is bind() → DELETE, never a POST abort. + sendTimer = setTimeout(() => sendCtrl.abort(), 15000); } let sendReq = authFetch( this._base + @@ -3836,8 +3825,15 @@ class Pane { // optimistic user bubble is already in the log and the server // still delivers the message on worker drain — accept the // small UX gap (no in-UI dismiss for THIS message). - if (queuedEl) this.queue.bind(queuedEl, data.msg_id); - else this.setBusy(true); + if (queuedEl) { + // A deferred send (command-window defer) can carry + // attachments — stash the count BEFORE bind (a pre-bind ✕ + // confirms inside bind) so the dismiss path can tell the + // user the attachments were discarded: the chips are + // consumed below and a retract does not re-stage them. + queuedEl._deferredAttachments = (data.attached_ids || []).length; + this.queue.bind(queuedEl, data.msg_id); + } else this.setBusy(true); this.attachments.consume( data.attached_ids, data.dropped_attachment_ids,