diff --git a/CHANGELOG.md b/CHANGELOG.md index 33f12c86..46540321 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -214,7 +214,18 @@ Earlier stable lines (`stable/1.6`, `stable/1.5`) are frozen. 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). + documents the at-most-once durability contract). Deferred responses + carry `"deferred": true`, the pending list is the **order authority** + (a fresh send — or a coordinator dispatch, or a queued-nudge wake — + lines up behind acknowledged entries instead of overtaking them, and + the wake gate re-arms at the drain's exit even when everything pending + was retracted), a dispatch crash re-queues the entry instead of eating + an acknowledged message, and each dispatch emits a pane-tier + `message_dispatched` event (`folded: true` for interjection fold-ins) + so queued-bubble UI keeps its retract affordance exactly until the + message truly leaves — including when the send was accepted by a pane + that believed the workstream idle, which now renders a real queued + chip instead of a sent-looking bubble. 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, @@ -223,10 +234,13 @@ Earlier stable lines (`stable/1.6`, `stable/1.5`) are frozen. (without parking an executor thread per request); the post-command pane refreshes (`clear_ui` after `/clear`/`/new`/`/resume`, the workstream-name sync) ride the worker itself, so a command that - outlives the endpoint's 60s response backstop still refreshes every - pane on completion (the `/command` response contract — `ok` / `running`, - with busy refusals answering a loud HTTP 409 rather than a silent 200 — - is now documented in the API reference and the OpenAPI spec). Manual compaction + outlives the endpoint's 25s response backstop still refreshes every + pane on completion (the backstop sits under the console proxy's 30s + client timeout so the degraded `running` answer can actually traverse + a proxied pane, which now surfaces it instead of silence; the + `/command` response contract — `ok` / `running`, with busy refusals + answering a loud HTTP 409 rather than a silent 200 — is now documented + in the API reference and the OpenAPI spec). Manual compaction success also refreshes the status line/context pill immediately (parity with auto-compaction), compaction failures keep feeding the typed `error` event and the node error counter (while a CLI Ctrl-C reports as diff --git a/docs/api-reference.md b/docs/api-reference.md index 139b884f..45f33959 100644 --- a/docs/api-reference.md +++ b/docs/api-reference.md @@ -788,32 +788,37 @@ Sends a user message to a workstream. Spawns a daemon worker thread that calls **Request body:** ```json -{"message": "Explain how the server works"} +{"message": "Explain how the server works", "attachment_ids": ["a1"]} ``` -| Field | Type | Required | Description | -|-----------|--------|----------|-------------------------| -| `message` | string | yes | The user's message text | +| Field | Type | Required | Description | +|------------------|------------|----------|------------------------------------------------------| +| `message` | string | yes | The user's message text | +| `attachment_ids` | string[] | no | Staged uploads to attach (omit = auto-consume; `[]` = none) | -**Response (success):** +**Response.** Every 200 body carries `attached_ids` and +`dropped_attachment_ids` (empty lists when no attachments are involved): -```json -{"status": "ok"} -``` - -**Response (busy):** Returned if the workstream's worker thread is still alive -from a previous request. Also pushes a `busy_error` event to the SSE stream. - -```json -{"status": "busy"} -``` +- `{"status": "ok", ...}` — a fresh turn was dispatched. +- `{"status": "queued", "priority", "msg_id", ...}` — folded into the live + turn's interjection queue; delivered at the next tool-result seam. + `DELETE .../send` with the `msg_id` retracts it before delivery. +- `{"status": "queued", "deferred": true, ...}` — parked on the deferred-send + list (a command window holds the slot, or earlier deferred sends are + pending) and dispatched as its own full-fidelity send afterwards; see the + defer contract under `POST /v1/api/command`. +- `{"status": "queue_full", ...}` — the live worker's queue is at capacity; + retry shortly. +- `{"status": "attachments_busy", ...}` — attachments can't ride a queued + turn; the staged uploads survive for a retry once the worker idles. **Error responses:** -| Status | Body | Condition | -|--------|------------------------------------|------------------------| -| 400 | `{"error": "Empty message"}` | Message is empty | -| 404 | `{"error": "Unknown workstream"}` | `ws_id` not found | +| Status | Body | Condition | +|--------|-------------------------------------------------|----------------------------------------| +| 400 | `{"error": "message is required"}` | Message is empty | +| 404 | `{"error": "Unknown workstream"}` | `ws_id` not found (or closed mid-send) | +| 409 | `{"status": "cross_user_interjection", ...}` | Another participant's turn is in flight | --- @@ -863,9 +868,11 @@ synchronous: - **Quick commands** (everything except `/compact`): the endpoint waits for completion, so `{"status": "ok"}` means the command ran. A command still - running after 60 s answers `{"status": "running"}` — the worker keeps + running after 25 s answers `{"status": "running"}` — the worker keeps going, its output reaches the pane via SSE, and the post-command pane - refreshes below still fire when it completes. + refreshes below still fire when it completes. (The bound sits under + common 30 s client/proxy timeouts — the console proxy's included — so + the degraded answer actually reaches bounded callers.) - **`/compact`**: dispatched fire-and-forget — `{"status": "ok"}` means the compaction *started*. A large context can legitimately compact for many minutes; progress streams as `compaction` SSE events (see the event @@ -878,18 +885,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 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 +While a command holds the slot — and afterwards, while earlier deferred +sends are still waiting (the pending list is the order authority: a fresh +send never overtakes a message already acknowledged) — `POST .../send` +requests are **deferred**: the server answers `{"status": "queued", +"deferred": true, "msg_id": ...}` immediately and dispatches the message +as an ordinary full-fidelity send (attachments and sender identity +included) in arrival order once the slot frees — 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. +discards any attachments it carried; re-attach to send them again. When a +deferred send dispatches, panes receive a `message_dispatched` event +(`msg_id`, plus `folded: true` when it folded into a live turn's +interjection queue rather than spawning its own turn) so queued-message +UI can settle the right way. Durability: deferred sends are **node-local and in-memory** (the same lifetime as the interjection queue). `"queued"` is at-most-once intake, not @@ -909,10 +923,10 @@ SSE stream / in `/history`). | `command` | string | yes | The slash command (e.g. `/clear`) | | `ws_id` | string | yes | Target workstream ID | -If the command is `/clear` or `/new`, the server pushes a `clear_ui` SSE event -to instruct the client to reset its message display. If the command is -`/resume`, the server pushes `clear_ui` followed by a `history` event -containing the resumed session's messages. These follow-ups are emitted by the +If the command is `/clear`, `/new`, or `/resume`, the server pushes a +`clear_ui` SSE event to instruct the client to reset its message display and +re-fetch the transcript via `GET .../history` (there is no SSE event that +carries the messages themselves). These follow-ups are emitted by the command worker itself, so they fire even when the endpoint already answered `{"status": "running"}`. diff --git a/sdk/typescript/openapi-server.json b/sdk/typescript/openapi-server.json index 321d7eac..185edf28 100644 --- a/sdk/typescript/openapi-server.json +++ b/sdk/typescript/openapi-server.json @@ -2280,16 +2280,23 @@ "SendResponse": { "properties": { "status": { - "description": "'ok', 'busy', 'queued', or 'queue_full'", + "description": "'ok' (fresh turn dispatched), 'queued' (folded into the live turn's interjection queue, or \u2014 when `deferred` is true \u2014 parked for dispatch after the current command window), 'queue_full', 'attachments_busy' (attachments can't ride a queued turn; retry when idle), or 'cross_user_interjection' (another participant's turn is in flight; carried on the 409 body).", "examples": [ "ok", - "busy", "queued", - "queue_full" + "queue_full", + "attachments_busy", + "cross_user_interjection" ], "title": "Status", "type": "string" }, + "deferred": { + "default": false, + "description": "Set on `queued` responses: the message is parked on the workstream's deferred-send list (a slash-command window holds the worker slot, or earlier deferred sends are still pending) and dispatches as an ordinary full-fidelity send afterwards \u2014 it is NOT in a live turn's interjection queue. `DELETE .../send` retracts it until dispatch. Node-local and in-memory: a node restart before dispatch drops it (at-most-once intake).", + "title": "Deferred", + "type": "boolean" + }, "attached_ids": { "description": "Attachment ids actually attached to this turn. Subset of the request's `attachment_ids` (or the auto-consumed pending set). Empty when the send carries no attachments.", "items": { diff --git a/tests/test_cooperative_compaction.py b/tests/test_cooperative_compaction.py index 3b16643d..4d5295d8 100644 --- a/tests/test_cooperative_compaction.py +++ b/tests/test_cooperative_compaction.py @@ -2013,6 +2013,91 @@ class TestPreHookUICompat: ) assert infos == [] + def test_explicit_protocol_subclass_inherits_classic_lines(self, session): + """An explicit ``class MyUI(SessionUI)`` never lands in the getattr + fallback — it INHERITS the protocol member as a real method — so + the protocol default body itself must be the pre-1.8 rendering. + With a bare ``...`` stub, exactly these embedders (the ones the + fallback was built for) silently swallowed every lifecycle event.""" + from turnstone.core.session import SessionUI + + infos: list[str] = [] + + class _ExplicitUI(SessionUI): + def on_info(self, message: str) -> None: + infos.append(message) + + session.ui = _ExplicitUI() + result = session._compaction_event( + 0, + { + "phase": "end", + "ok": True, + "trigger": "auto", + "before_tokens": 900, + "after_tokens": 100, + "summary": "dense", + }, + ) + assert result is None # inherited default returns None, not a stub echo + # Rendered exactly once — via the inherited default, with no second + # pass through the duck-type fallback (emit is not None here). + assert sum("compacted: ~900 -> ~100 tokens" in m for m in infos) == 1 + assert any("dense" in m for m in infos) + + def test_explicit_subclass_override_suppresses_default_lines(self, session): + """A subclass that implements the hook owns the rendering: no + classic info lines from the default body, and its return value + reaches the marker stamp.""" + from turnstone.core.session import SessionUI + + infos: list[str] = [] + seen: list[dict] = [] + + class _HookedUI(SessionUI): + def on_info(self, message: str) -> None: + infos.append(message) + + def on_compaction(self, payload: dict) -> int | None: + seen.append(payload) + return 42 + + session.ui = _HookedUI() + result = session._compaction_event( + 0, {"phase": "end", "ok": True, "trigger": "auto", "summary": "s"} + ) + assert result == 42 + assert len(seen) == 1 + assert infos == [] + + def test_duck_hook_bool_return_never_reaches_marker_stamp(self, session): + """A duck-typed on_compaction returning ``True`` (bool ⊂ int) must + coerce to ``None`` — a boolean stamped into the persisted marker's + event_id fails the PG INSERT after the history swap committed and + mis-keys the SQLite dedupe (same guard as parse_checkpoint_watermark, + via _coerce_event_id).""" + session.ui = SimpleNamespace( + on_thinking_start=lambda: None, + on_thinking_stop=lambda: None, + on_error=lambda _m: None, + on_compaction=lambda _payload: True, + ) + result = session._compaction_event( + 0, {"phase": "end", "ok": True, "trigger": "auto", "summary": "s"} + ) + assert result is None + + def test_coerce_event_id_rejects_bools(self): + """The shared coercion helper: ints pass, bools and non-ints don't.""" + from turnstone.core.session import _coerce_event_id + + assert _coerce_event_id(46) == 46 + assert _coerce_event_id(0) == 0 + assert _coerce_event_id(True) is None + assert _coerce_event_id(False) is None + assert _coerce_event_id(None) is None + assert _coerce_event_id("46") is None + class TestCompactionNoticeStamp: """_compaction_event is the single display-policy site: failed ends diff --git a/tests/test_idle_nudge_watcher.py b/tests/test_idle_nudge_watcher.py index 2c0b4a68..c853b632 100644 --- a/tests/test_idle_nudge_watcher.py +++ b/tests/test_idle_nudge_watcher.py @@ -39,6 +39,10 @@ class _FakeWorkstream: self._worker_running = False self._closed = False self.worker_thread: Any = None + # Deferred /send entries — the wake gate yields while any are + # pending (order barrier); empty is the default every other test + # assumes. + self._pending_sends: list[Any] = [] class _FakeManager: @@ -194,6 +198,25 @@ class TestWakeWorkstreamIfPending: assert wake_workstream_if_pending(ws) is False assert mock_send.call_count == 0 + def test_yields_to_pending_deferred_sends(self, fake_mgr_and_ws): + """Deferred /send entries hold the order barrier: a wake worker + claiming the slot would push messages already acknowledged + "queued" behind its whole turn, so the gate yields. Re-armed + structurally — every deferred turn's exit re-runs the gate, and + the drain's clean exit (trigger="drain-exit") covers a list that + emptied by pure retraction and never ran a turn.""" + _mgr, ws = fake_mgr_and_ws + ws.session._nudge_queue.enqueue("watch_triggered", "output", "any") + ws._pending_sends.append(object()) + with patch("turnstone.core.session_worker.send", return_value=True) as mock_send: + assert wake_workstream_if_pending(ws, trigger="worker-exit") is False + assert mock_send.call_count == 0 + # Barrier cleared (the drain retired) — the same call dispatches. + ws._pending_sends.clear() + with patch("turnstone.core.session_worker.send", return_value=True) as mock_send: + assert wake_workstream_if_pending(ws, trigger="drain-exit") is True + assert mock_send.call_count == 1 + def test_skips_closed_ws(self, fake_mgr_and_ws): """A workstream mid-``close()`` must not get a wake spawned on its torn-down session, even while its ``state`` field still diff --git a/tests/test_interactive_pane_js.py b/tests/test_interactive_pane_js.py index cce94bbc..849a561f 100644 --- a/tests/test_interactive_pane_js.py +++ b/tests/test_interactive_pane_js.py @@ -669,8 +669,60 @@ def test_send_post_abort_machinery_is_gone() -> None: # 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 + # The deferred-attachment count rides bind()'s documented options seam + # (controller dataset) — the per-pane element expando is dead. + for name, src in ( + ("interactive.js", interactive), + ("coordinator.js", coordinator), + ("composer_queue.js", composer_queue), + ): + assert "_deferredAttachments" not in src, ( + f"{name}: deferred state must ride bind(el, msgId, opts), not an expando" + ) + + +def test_deferred_send_settle_protocol_pins() -> None: + """The deferred-chip settle protocol (round 7, C4): a deferred send's + queued chip keeps its retraction affordance exactly until the message + truly leaves the parked list. Pins the controller's contract and both + panes' wiring — losing any of these silently re-promotes parked + messages to "sent" while the server still honors DELETE (loss + disguised as delivery on a node restart).""" + interactive = _INTERACTIVE.read_text(encoding="utf-8") + coordinator = (_ROOT / "turnstone/console/static/coordinator/coordinator.js").read_text( + encoding="utf-8" + ) + composer_queue = (_ROOT / "turnstone/shared_static/composer_queue.js").read_text( + encoding="utf-8" + ) + # Controller: bind() stores the options on its own dataset state... + assert "function bind(el, msgId, opts)" in composer_queue + assert 'el.dataset.deferred = "1"' in composer_queue + assert "el.dataset.attachedCount = String(opts.attachedCount)" in composer_queue + # ...the idle sweep skips deferred AND unbound chips (the "idle ⇒ + # drained" invariant is untrue for both)... + assert "if (el.dataset.deferred) return;" in composer_queue + assert "if (!el.dataset.msgId) return;" in composer_queue + # ...and settleDeferred branches on the fold-in arm: clear the flag + # only (DELETE still genuinely removes a folded message until the seam + # drains), promote only on the fresh-spawn arm. + assert "function settleDeferred(msgId, folded)" in composer_queue + assert "delete target.dataset.deferred;" in composer_queue + assert "settleDeferred: settleDeferred" in composer_queue + # A barrier-deferred entry can dispatch within milliseconds of its ack, + # so the SSE settle can beat the POST response's bind(): the controller + # parks chip-absent settles and bind() reconciles them — without this a + # raced chip stays flagged deferred and the idle sweep skips it forever. + assert "_preBindSettles" in composer_queue + assert "_preBindSettles.has(msgId)" in composer_queue + # Both panes pass the response through the options seam and consume the + # pane-tier settle event. + for name, src in (("interactive.js", interactive), ("coordinator.js", coordinator)): + assert "deferred: !!data.deferred" in src, f"{name}: bind must carry the deferred flag" + assert "attachedCount: (data.attached_ids || []).length" in src, name + assert 'case "message_dispatched"' in src, f"{name}: settle event not consumed" + assert "settleDeferred(" in src, name + # queuedEl-absent + deferred: the idle-thinking pane must render a + # real queued chip (retro-convert), not leave a sent-looking bubble + # for a parked, still-retractable message. + assert "!queuedEl && data.deferred" in src, f"{name}: idle-pane defer must chip" diff --git a/tests/test_server_attachments_endpoints.py b/tests/test_server_attachments_endpoints.py index d09d53f0..56657a0f 100644 --- a/tests/test_server_attachments_endpoints.py +++ b/tests/test_server_attachments_endpoints.py @@ -504,6 +504,11 @@ class TestSendMessageAttachments: ws.worker_thread = None ws._worker_running = False ws._closed = False # a bare Mock attr is truthy → send() would refuse + # Same truthy-Mock trap as _closed: the /send order barrier reads + # both — a bare Mock attr would defer every send behind a phantom + # pending list. + ws._pending_sends = [] + ws._pending_drain = None ws._lock = threading.RLock() mgr.get.return_value = ws return captured, session @@ -703,6 +708,10 @@ class TestQueuedSendWithAttachments: ws.worker_thread = worker ws._worker_running = True ws._closed = False # a bare Mock attr is truthy → send() would refuse + # Same truthy-Mock trap: the /send order barrier reads both — a + # bare Mock attr would defer every send behind a phantom list. + ws._pending_sends = [] + ws._pending_drain = None ws._lock = threading.RLock() mgr.get.return_value = ws return captured @@ -767,6 +776,10 @@ class TestBusyWorkerAttachments: ws.session = session ws.worker_thread = worker ws._closed = False # a bare Mock attr is truthy → send() would refuse + # Same truthy-Mock trap: the /send order barrier reads both — a + # bare Mock attr would defer every send behind a phantom list. + ws._pending_sends = [] + ws._pending_drain = None ws._lock = threading.RLock() mgr.get.return_value = ws return ws, session diff --git a/tests/test_server_authz.py b/tests/test_server_authz.py index 1bdda0c0..975e58e4 100644 --- a/tests/test_server_authz.py +++ b/tests/test_server_authz.py @@ -19,6 +19,8 @@ from unittest.mock import MagicMock import pytest from starlette.testclient import TestClient +from tests._helpers import wait_until + _TEST_JWT_SECRET = "test-jwt-secret-minimum-32-chars!" @@ -1305,16 +1307,11 @@ class TestCompactCommandDispatch: # The worker wrapped the run in busy/idle state transitions. assert ws.ui.states == ["thinking", "idle"] - 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 + # Drain outcomes are polled with tests._helpers.wait_until (flake- + # hardened final re-check; raises instead of returning False) at + # timeout=8.0 — the drain thread runs at a 0.25s cadence, so dispatch + # is asynchronous with respect to the released command worker and the + # helper's 5s default is too tight for CI descheduling stalls. def _drain_idle(self, ws) -> bool: """True once the pending-send drain retired itself (list empty, @@ -1355,6 +1352,9 @@ class TestCompactCommandDispatch: assert r.status_code == 200 body = r.json() assert body["status"] == "queued" + # The SDK-facing discriminator between "interjection-queued into a + # live turn" and "parked on the deferred list" (SendResponse). + assert body["deferred"] is True assert body["msg_id"] assert body["priority"] == "notice" # Registered, not dispatched: the window is still open. @@ -1363,7 +1363,7 @@ class TestCompactCommandDispatch: with ws._lock: assert len(ws._pending_sends) == 1 gate.set() # compaction finishes; the drain dispatches - assert self._wait_for(lambda: ws.session.sends) + wait_until(lambda: ws.session.sends, timeout=8.0) # 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). @@ -1372,7 +1372,7 @@ class TestCompactCommandDispatch: # 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)) + wait_until(lambda: self._drain_idle(ws), timeout=8.0) def test_cancelled_compact_flushes_queue_without_answering(self, app_client): """A user-stopped compaction must not auto-run a turn they may no @@ -1486,13 +1486,13 @@ class TestCompactCommandDispatch: assert body["attached_ids"] == ["a1"] assert ws.session.sends == [] # still deferred gate.set() - assert self._wait_for(lambda: ws.session.sends) + wait_until(lambda: ws.session.sends, timeout=8.0) 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)) + wait_until(lambda: self._drain_idle(ws), timeout=8.0) def test_send_during_quick_command_window_defers(self, app_client): """The defer applies to EVERY command window, not just /compact — @@ -1536,10 +1536,10 @@ class TestCompactCommandDispatch: gate.set() runner.join(timeout=10) assert cmd_result["body"] == {"status": "ok"} - assert self._wait_for(lambda: ws.session.sends) + wait_until(lambda: ws.session.sends, timeout=8.0) 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)) + wait_until(lambda: self._drain_idle(ws), timeout=8.0) 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 — @@ -1719,7 +1719,7 @@ class TestCompactCommandDispatch: 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)) + wait_until(lambda: self._drain_idle(ws), timeout=8.0) assert ws.session.sends == [] assert ws.session.queue_calls == [] # Unknown ids still answer not_found after checking both holders. @@ -1759,9 +1759,9 @@ class TestCompactCommandDispatch: 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) + wait_until(lambda: len(ws.session.sends) == 2, timeout=8.0) assert [s[0] for s in ws.session.sends] == ["first", "second"] - assert self._wait_for(lambda: self._drain_idle(ws)) + wait_until(lambda: self._drain_idle(ws), timeout=8.0) def test_force_cancel_of_wedged_command_releases_drain(self, app_client): """Force-cancelling a wedged command clears the slot flags — the @@ -1794,12 +1794,12 @@ class TestCompactCommandDispatch: ) assert resp.status_code == 200 # Dispatch happens while the zombie is STILL wedged on the gate. - assert self._wait_for(lambda: ws.session.sends) + wait_until(lambda: ws.session.sends, timeout=8.0) 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)) + wait_until(lambda: self._drain_idle(ws), timeout=8.0) def test_ws_close_mid_window_drops_pending_and_drain_exits(self, app_client): """A workstream closed with deferred sends outstanding drops them @@ -1825,7 +1825,7 @@ class TestCompactCommandDispatch: with ws._lock: ws._closed = True # the SessionManager.close tombstone shape gate.set() - assert self._wait_for(lambda: self._drain_idle(ws)) + wait_until(lambda: self._drain_idle(ws), timeout=8.0) assert ws.session.sends == [] assert ws.session.queue_calls == [] @@ -1855,10 +1855,10 @@ class TestCompactCommandDispatch: 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) + wait_until(lambda: new_session.sends, timeout=8.0) 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)) + wait_until(lambda: self._drain_idle(ws), timeout=8.0) def test_rejected_deferred_entry_waits_for_slot_then_fresh_spawns(self, app_client): """The drain's rejection arm: an entry the interjection fallback @@ -1893,15 +1893,274 @@ class TestCompactCommandDispatch: # 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"]) + wait_until(lambda: ws.session.queue_calls == ["second"], timeout=8.0) 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) + wait_until(lambda: len(ws.session.sends) == 2, timeout=8.0) # 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)) + wait_until(lambda: self._drain_idle(ws), timeout=8.0) + + def test_drain_crash_requeues_entry_and_retries(self, app_client): + """A crash inside a claimed entry's dispatch attempt (the real-world + shape: Thread.start raising under thread exhaustion) must not eat + the acknowledged message — the per-iteration handler re-inserts it + at head and retries after a backoff, keeping the docstring's + no-silent-drop contract.""" + 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": "survive the crash"}, + headers=_auth("user-1"), + ) + assert r.json()["status"] == "queued" + with ws._lock: + entry = ws._pending_sends[0] + real_attempt = entry.attempt + calls = {"n": 0} + + def crash_once(session): + calls["n"] += 1 + if calls["n"] == 1: + raise RuntimeError("simulated Thread.start failure") + return real_attempt(session) + + entry.attempt = crash_once + gate.set() + wait_until(lambda: ws.session.sends, timeout=8.0) + assert [s[0] for s in ws.session.sends] == ["survive the crash"] + assert calls["n"] == 2 # crashed once, re-queued, dispatched on retry + wait_until(lambda: self._drain_idle(ws), timeout=8.0) + + def test_persistently_crashing_entry_is_never_dropped_and_retract_frees_drain(self, app_client): + """A persistently crashing attempt keeps the entry alive (retry + loop, never a silent drop) — and the user's DELETE still works + mid-crash-loop: the re-inserted entry is marked retracted, the + loop-top purge drops it, and the drain retires cleanly.""" + 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": "doomed"}, + headers=_auth("user-1"), + ) + msg_id = r.json()["msg_id"] + with ws._lock: + entry = ws._pending_sends[0] + calls = {"n": 0} + + def always_crash(_session): + calls["n"] += 1 + raise RuntimeError("persistent dispatch failure") + + entry.attempt = always_crash + gate.set() + # Two attempts across the ~1s backoff prove the loop survived the + # first crash with the entry intact (a dropped entry can't be + # re-attempted) and nothing was dispatched behind the user's back. + wait_until(lambda: calls["n"] >= 2, timeout=8.0) + assert ws.session.sends == [] + 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"} + wait_until(lambda: self._drain_idle(ws), timeout=8.0) + assert ws.session.sends == [] + + def test_fresh_send_defers_behind_claimed_entry(self, app_client): + """The order barrier's drain-alive term: a send arriving while the + drain has CLAIMED an entry (off the list, dispatch in flight) must + defer behind it, not overtake — the pre-fix route dispatched + immediately (the list looked empty) and inverted answer order + against the acknowledged send.""" + 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"), + ) + r1 = client.post( + f"/v1/api/workstreams/{ws_id}/send", + json={"message": "first"}, + headers=_auth("user-1"), + ) + assert r1.json()["deferred"] is True + with ws._lock: + entry = ws._pending_sends[0] + real_attempt = entry.attempt + claimed = threading.Event() + release = threading.Event() + + def gated_attempt(session): + claimed.set() + assert release.wait(timeout=10) + return real_attempt(session) + + entry.attempt = gated_attempt + gate.set() # window closes; the drain claims "first" and blocks + wait_until(claimed.is_set, timeout=8.0) + with ws._lock: + assert ws._pending_sends == [] # claimed — the list alone says "free" + big = "y" * 5000 # full-fidelity: can't fold into any live turn + r2 = client.post( + f"/v1/api/workstreams/{ws_id}/send", + json={"message": big}, + headers=_auth("user-1"), + ) + # Deferred via the barrier (drain alive), never dispatched directly. + assert r2.json()["status"] == "queued" + assert r2.json()["deferred"] is True + assert ws.session.sends == [] + release.set() + wait_until(lambda: len(ws.session.sends) == 2, timeout=8.0) + assert [s[0] for s in ws.session.sends] == ["first", big] + wait_until(lambda: self._drain_idle(ws), timeout=8.0) + + def test_drain_exit_re_arms_wake_gate_after_pure_retraction(self, app_client, monkeypatch): + """A pending list that empties by RETRACTION never runs a deferred + turn, so no worker-exit backstop would re-run the wake gate that + yielded to the barrier — the drain's clean exit must re-arm it + itself (trigger="drain-exit"), or a nudge parked during the window + strands until some unrelated future dispatch.""" + from turnstone.core import idle_nudge_watcher + + wake_calls: list[str] = [] + monkeypatch.setattr( + idle_nudge_watcher, + "wake_workstream_if_pending", + lambda ws, *, trigger="unspecified": (wake_calls.append(trigger), False)[1], + ) + client, mgr = app_client + ws_id = self._create_ws(client) + ws = mgr.get(ws_id) + assert ws is not None + gate = threading.Event() + ws.session.compact_gate = gate + client.post( + "/v1/api/command", + json={"command": "/compact", "ws_id": ws_id}, + headers=_auth("user-1"), + ) + r = client.post( + f"/v1/api/workstreams/{ws_id}/send", + json={"message": "will be retracted"}, + headers=_auth("user-1"), + ) + d = client.request( + "DELETE", + f"/v1/api/workstreams/{ws_id}/send", + json={"msg_id": r.json()["msg_id"]}, + headers=_auth("user-1"), + ) + assert d.json() == {"status": "removed"} + gate.set() + wait_until(lambda: self._drain_idle(ws), timeout=8.0) + wait_until(lambda: "drain-exit" in wake_calls, timeout=8.0) + assert ws.session.sends == [] + + def test_deferred_dispatch_emits_message_dispatched(self, app_client): + """Fresh-spawn arm of the settle protocol: dispatching a deferred + entry emits pane-tier ``message_dispatched {msg_id}`` (no + ``folded`` key) so the queued chip promotes exactly when the + retraction window truly closes.""" + 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": "settle me"}, + headers=_auth("user-1"), + ) + msg_id = r.json()["msg_id"] + queued_events = [e for e in ws.ui._enqueued if e.get("type") == "message_queued"] + assert [e["msg_id"] for e in queued_events] == [msg_id] + gate.set() + wait_until( + lambda: any(e.get("type") == "message_dispatched" for e in ws.ui._enqueued), + timeout=8.0, + ) + dispatched = [e for e in ws.ui._enqueued if e.get("type") == "message_dispatched"] + assert dispatched == [{"type": "message_dispatched", "msg_id": msg_id}] + wait_until(lambda: self._drain_idle(ws), timeout=8.0) + + def test_folded_deferred_dispatch_emits_folded_flag(self, app_client): + """Interjection fold-in arm: a deferred entry that dispatches INTO a + live turn's queue emits ``folded: true`` — the client must clear + only its deferred flag (DELETE still genuinely removes the message + from the interjection queue until the seam drains), not promote.""" + 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 + 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() + gate.set() + # "first" spawns fresh and wedges in send() — a live turn holds the + # slot, so "second" (small, no attachments) folds into its + # interjection queue. + wait_until(lambda: ws.session.queue_calls == ["second"], timeout=8.0) + send_gate.set() + wait_until(lambda: self._drain_idle(ws), timeout=8.0) + dispatched = [e for e in ws.ui._enqueued if e.get("type") == "message_dispatched"] + assert dispatched == [ + {"type": "message_dispatched", "msg_id": first["msg_id"]}, + {"type": "message_dispatched", "msg_id": second["msg_id"], "folded": True}, + ] + assert [s[0] for s in ws.session.sends] == ["first"] def test_exit_command_emits_ended_info_and_never_answers(self, app_client): """should_exit commands shut the session down — the worker emits diff --git a/tests/test_sse_cursor_resume.py b/tests/test_sse_cursor_resume.py index 72552a9b..5a9db258 100644 --- a/tests/test_sse_cursor_resume.py +++ b/tests/test_sse_cursor_resume.py @@ -236,6 +236,27 @@ def test_append_system_turn_stamps_row_with_its_sse_event_id( assert ui._event_buffer[-1][1]["type"] == "system_turn" +def test_system_turn_bool_hook_return_falls_back_to_counter( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A duck-typed ``on_system_turn`` returning ``True`` (bool ⊂ int) must + never stamp a boolean into the persisted row: PostgreSQL fails the + INSERT after the turn already appended, SQLite stores ``1`` and + mis-keys the /history-vs-replay dedupe. ``_coerce_event_id`` rejects + bools at the shared chokepoint and the stamp falls back to the + ring-buffer counter (same guard class as ``parse_checkpoint_watermark``).""" + session = make_session() + captured: dict[str, Any] = {} + monkeypatch.setattr( + "turnstone.core.session.save_message", + lambda *a, **k: captured.update(event_id=k.get("event_id")), + ) + session.ui.on_system_turn = lambda *_a, **_k: True + session._append_system_turn("start", "ground yourself") + assert not isinstance(captured["event_id"], bool) + assert captured["event_id"] == session._ui_event_id() + + # --------------------------------------------------------------------------- # Storage: event_id round-trip, get_max_event_id, _event_id reseed # --------------------------------------------------------------------------- diff --git a/turnstone/api/server_schemas.py b/turnstone/api/server_schemas.py index 64ad9eb4..9bf411c8 100644 --- a/turnstone/api/server_schemas.py +++ b/turnstone/api/server_schemas.py @@ -39,8 +39,28 @@ class DequeueRequest(BaseModel): class SendResponse(BaseModel): status: str = Field( - description="'ok', 'busy', 'queued', or 'queue_full'", - examples=["ok", "busy", "queued", "queue_full"], + description=( + "'ok' (fresh turn dispatched), 'queued' (folded into the live " + "turn's interjection queue, or — when `deferred` is true — " + "parked for dispatch after the current command window), " + "'queue_full', 'attachments_busy' (attachments can't ride a " + "queued turn; retry when idle), or 'cross_user_interjection' " + "(another participant's turn is in flight; carried on the 409 " + "body)." + ), + examples=["ok", "queued", "queue_full", "attachments_busy", "cross_user_interjection"], + ) + deferred: bool = Field( + default=False, + description=( + "Set on `queued` responses: the message is parked on the " + "workstream's deferred-send list (a slash-command window holds " + "the worker slot, or earlier deferred sends are still pending) " + "and dispatches as an ordinary full-fidelity send afterwards — " + "it is NOT in a live turn's interjection queue. `DELETE .../send` " + "retracts it until dispatch. Node-local and in-memory: a node " + "restart before dispatch drops it (at-most-once intake)." + ), ) attached_ids: list[str] = Field( default_factory=list, diff --git a/turnstone/console/coordinator_adapter.py b/turnstone/console/coordinator_adapter.py index 2b144907..e3bcd5ad 100644 --- a/turnstone/console/coordinator_adapter.py +++ b/turnstone/console/coordinator_adapter.py @@ -391,6 +391,25 @@ class CoordinatorAdapter: interjector_user_id=acting_user_id, ) + # Order-barrier yield, mirroring the /send route's pre-check: once + # deferred sends are pending (or a claimed entry's dispatch is in + # flight — the drain-alive term), a spawn here would overtake + # messages already acknowledged "queued". Refuse via the return + # value — the adapter's documented backpressure surface, which the + # sole call site reports as queue_full/undelivered — NEVER by + # raising queue.Full from the body: only enqueue closures may (the + # dispatcher catches it there); a body-raise would escape into the + # create handler as a crash. Unreachable today (that caller + # dispatches on a freshly created workstream, which cannot have + # pending sends); the guard exists for future dispatch callers. + drain_t = ws._pending_drain + if ws._pending_sends or (drain_t is not None and drain_t.is_alive()): + log.warning( + "coord_adapter.send_refused_pending_sends ws=%s count=%d", + ws.id[:8], + len(ws._pending_sends), + ) + return False return session_worker.send( ws, enqueue=_enqueue, diff --git a/turnstone/console/static/coordinator/coordinator.js b/turnstone/console/static/coordinator/coordinator.js index 2fbe4923..1911f25a 100644 --- a/turnstone/console/static/coordinator/coordinator.js +++ b/turnstone/console/static/coordinator/coordinator.js @@ -1956,15 +1956,18 @@ function createCoordinatorPane(root, wsId, opts) { const snap = attachments.snapshot(); let queuedEl = null; + let optimisticEl = null; + // Server re-parses the !!! prefix to set queue priority — the + // optimistic bubble strips it for display. Parsed outside the busy + // branch: the queued+deferred response arm needs it too (an idle + // pane's send can defer behind a command window / pending list). + let displayText = trimmed; + let priority = "notice"; + if (trimmed.startsWith("!!!")) { + displayText = trimmed.slice(3).trimStart(); + priority = "important"; + } if (busy) { - // Server re-parses the !!! prefix to set queue priority — the - // optimistic bubble strips it for display. - let displayText = trimmed; - let priority = "notice"; - if (trimmed.startsWith("!!!")) { - displayText = trimmed.slice(3).trimStart(); - priority = "important"; - } queuedEl = queue.addQueuedMessage(displayText, priority); } else { setBusy(true); @@ -1972,9 +1975,13 @@ function createCoordinatorPane(root, wsId, opts) { // for every stable chip the composer holds; pass it through so // the optimistic user bubble shows the same pill cluster the // history-replay path renders below. - appendUserMessageWithAttachments(trimmed, snap.attachments, { - label: "you", - }); + optimisticEl = appendUserMessageWithAttachments( + trimmed, + snap.attachments, + { + label: "you", + }, + ); } composer.clear(); @@ -2045,21 +2052,24 @@ function createCoordinatorPane(root, wsId, opts) { }) .then((data) => { if (data && data.status === "queued" && data.msg_id) { - // Race: server returned queued but the client thought it was - // idle (SSE state_change hadn't arrived yet on initial load / - // reconnect). The optimistic user bubble is already in the - // log; we can't bind msg_id to a queued bubble retroactively - // without flipping the visual state mid-stream. Flip the busy - // 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. + // queuedEl-present: bind() handles the known races. queuedEl- + // absent + deferred: the pane thought it was idle but the send + // parked on the server's deferred list (command window / order + // barrier) — a plain sent bubble would present a parked, + // still-retractable, restart-droppable message as delivered, so + // replace the optimistic bubble with a real queued chip. + // queuedEl-absent + undeferred: plain interjection into a live + // turn the client hadn't seen yet (SSE state_change race); keep + // the historical small-UX-gap behavior (busy flip only). + if (!queuedEl && data.deferred) { + if (optimisticEl && optimisticEl.isConnected) optimisticEl.remove(); + queuedEl = queue.addQueuedMessage(displayText, priority); + } if (queuedEl) { - // 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); + queue.bind(queuedEl, data.msg_id, { + deferred: !!data.deferred, + attachedCount: (data.attached_ids || []).length, + }); } else setBusy(true); attachments.consume(data.attached_ids, data.dropped_attachment_ids); } else if (data && data.status === "busy") { @@ -3041,6 +3051,14 @@ function createCoordinatorPane(root, wsId, opts) { // surfaced an extra info row, which doubled up with the // queued bubble once the composer started rendering one.) break; + case "message_dispatched": + // A deferred send left the parked list: fresh spawn (promote the + // chip — the ×'s window is over) or interjection fold-in + // (folded: true — only the deferred flag clears; the chip resumes + // the normal queued lifecycle). No-op when this tab holds no + // matching chip. + queue.settleDeferred(ev.msg_id, !!ev.folded); + break; case "busy_error": // Worker is still alive after a cancel attempt; re-arm the // Stop button so the user can try again (or escalate to diff --git a/turnstone/core/idle_nudge_watcher.py b/turnstone/core/idle_nudge_watcher.py index 6eccae2b..d0a95bad 100644 --- a/turnstone/core/idle_nudge_watcher.py +++ b/turnstone/core/idle_nudge_watcher.py @@ -75,6 +75,10 @@ def wake_workstream_if_pending(ws: Workstream, *, trigger: str = "unspecified") queue at its own seams (``ATTENTION``/``THINKING``/``RUNNING`` all imply a live worker), and ``ERROR`` stays parked for the operator rather than burning inference unattended. + * ``ws._pending_sends`` non-empty — deferred sends hold the order + barrier; the wake yields and is re-armed by the deferred turns' + exit backstops (or the drain's clean exit when everything was + retracted). See the inline comment for the staleness argument. * nothing gate-eligible under ``WAKE_PENDING`` — tool-only/quiet entries belong to the next tool-result seam, not a synthetic empty user turn (``deliver_wake_nudge_from_queue`` would no-op on them). @@ -101,6 +105,21 @@ def wake_workstream_if_pending(ws: Workstream, *, trigger: str = "unspecified") session = ws.session if session is None or ws._closed or ws.state is not WorkstreamState.IDLE: return False + if ws._pending_sends: + # Order-barrier yield: deferred sends (acknowledged "queued" + # during a command window — see _PendingSend) are older than any + # nudge, and a wake worker claiming the slot would push them + # behind its whole turn. Lockless peek, benign both ways: a + # stale non-empty skips once more (the next exit backstop + # converges), a stale empty means the concurrent defer holds no + # order contract against this wake anyway. Convergence is + # structural — every path that clears the barrier re-runs this + # gate: each deferred turn's exit via ``_retry_pending_wake``, + # and the drain's own clean exit (trigger="drain-exit"), which + # covers a list that empties by pure retraction and so never + # runs a turn. + log.info("nudge_wake.yielded_to_pending_sends ws=%s trigger=%s", ws.id[:8], trigger) + return False nudge_queue = getattr(session, "_nudge_queue", None) # Gate on WAKE_PENDING, not USER_DRAIN: ``"quiet"`` entries (external # events demoted by a user cancel) deliver at the next legitimate seam diff --git a/turnstone/core/session.py b/turnstone/core/session.py index e2e4120c..4fb53bbc 100644 --- a/turnstone/core/session.py +++ b/turnstone/core/session.py @@ -1150,6 +1150,21 @@ def _is_ctx_overflow(exc: BaseException) -> bool: ) +def _coerce_event_id(value: Any) -> int | None: + """Narrow a duck-typed hook return / attribute to a usable event id. + + ``bool`` is an ``int`` subclass, so a bare ``isinstance(value, int)`` + lets a hook returning ``True`` stamp a boolean into a persisted + ``event_id`` — PostgreSQL then fails the INSERT after the state it + annotates already committed, and SQLite stores ``1`` and mis-keys + the /history-vs-replay dedupe. Same guard as + ``parse_checkpoint_watermark`` (storage/_utils.py); every consumer + of a duck-typed event-id source must route through here rather than + re-deriving the isinstance chain per site. + """ + return value if isinstance(value, int) and not isinstance(value, bool) else None + + class SessionUI(Protocol): def on_turn_start(self) -> None: ... def on_turn_committed(self) -> None: ... @@ -1187,8 +1202,22 @@ class SessionUI(Protocol): assigns one (see :meth:`on_system_turn`) so the persisted compaction marker row can be stamped with the matching resume cursor; ``None`` for UIs without an event stream. + + The default body IS the pre-1.8 compat rendering — the classic + ``on_info`` lines via the shared renderer. It must not be a bare + ``...`` stub: a Protocol member's body is inherited as a REAL + method by explicit subclasses, so a pre-1.8 ``class MyUI(SessionUI)`` + that never implemented this hook would satisfy the getattr probe + in ``_compaction_event`` with a silent no-op and defeat the very + fallback built for it — every lifecycle event swallowed, history + swapping with zero announcement. Event-stream UIs override and + return the assigned id; a subclass implementing neither hook + no-ops through ``on_info``'s own stub (the never-crash floor). """ - ... + from turnstone.core.compaction_render import render_compaction_event_as_info + + render_compaction_event_as_info(payload, self.on_info) + return None def on_state_change(self, state: str) -> None: ... def on_rename(self, name: str) -> None: ... @@ -3178,7 +3207,7 @@ class ChatSession: available" (the synthetic-snapshot floor). """ eid = getattr(self.ui, "_event_id", None) - return eid if isinstance(eid, int) else None + return _coerce_event_id(eid) def _tool_def_chars(self) -> int: """Total serialized char size of the active tool definitions (resent on @@ -5701,7 +5730,9 @@ class ChatSession: # event was delivered to double anyway). emitted_event_id: int | None = None try: - emitted_event_id = self.ui.on_system_turn(content, source, meta or None) + emitted_event_id = _coerce_event_id( + self.ui.on_system_turn(content, source, meta or None) + ) except Exception: log.warning("ui.on_system_turn failed; system turn still appended", exc_info=True) save_message( @@ -5709,7 +5740,7 @@ class ChatSession: "system", content, source=source, - event_id=emitted_event_id if isinstance(emitted_event_id, int) else self._ui_event_id(), + event_id=emitted_event_id if emitted_event_id is not None else self._ui_event_id(), meta=meta_json, ) @@ -7822,7 +7853,13 @@ class ChatSession: # getattr-guarded like on_generation_claimed/on_aux_usage: a # duck-typed SessionUI predating the hook must not hit an # AttributeError that wedges every long session at its first - # auto-compaction. + # auto-compaction. This probe only catches DUCK-typed UIs — an + # explicit ``class MyUI(SessionUI)`` inherits the protocol + # member as a real method and never lands in the None arm, which + # is why the protocol default body renders the same classic + # lines itself (see SessionUI.on_compaction): both compat routes + # converge on render_compaction_event_as_info, policy + # single-sited. emit = getattr(self.ui, "on_compaction", None) if emit is None: # Pre-hook UIs get the classic info lines back (an @@ -7846,8 +7883,9 @@ class ChatSession: return None result = emit(event) # Duck-typed hooks aren't bound to the protocol's return type; the - # marker-stamp consumer needs int-or-None, nothing else. - return result if isinstance(result, int) else None + # marker-stamp consumer needs int-or-None, nothing else (and a + # hook returning True must not stamp a bool — see _coerce_event_id). + return _coerce_event_id(result) def _compaction_bailed( self, @@ -9086,15 +9124,6 @@ class ChatSession: """ return self._flush_queued_messages() - @staticmethod - def _combine_queued_items(items: list[tuple[str, str]]) -> str: - """Render drained queue items to one text block ([IMPORTANT] framing).""" - from turnstone.core.tool_advisory import PRIORITY_IMPORTANT - - return "\n\n".join( - f"[IMPORTANT] {msg}" if pri == PRIORITY_IMPORTANT else msg for msg, pri in items - ) - def compact_now(self) -> bool: """Manual compaction with send()'s full generation discipline. @@ -9165,10 +9194,14 @@ class ChatSession: Returns ``True`` when any user row was appended (prefix or items), ``False`` when both were empty. """ + from turnstone.core.tool_advisory import PRIORITY_IMPORTANT + with self._queued_lock: items = list(self._queued_messages.values()) self._queued_messages.clear() - queued_text = self._combine_queued_items(items) + queued_text = "\n\n".join( + f"[IMPORTANT] {msg}" if pri == PRIORITY_IMPORTANT else msg for msg, pri in items + ) if not queued_text and not prefix: return False diff --git a/turnstone/core/session_routes.py b/turnstone/core/session_routes.py index fbc527a5..30b340b4 100644 --- a/turnstone/core/session_routes.py +++ b/turnstone/core/session_routes.py @@ -3978,10 +3978,22 @@ class _PendingSend: and in-memory, the interjection queue's lifetime — entries die with the workstream or the process, so "queued" is at-most-once intake, not durable acceptance. + + Invariant both the route's order barrier and the drain depend on: + **drain not alive ⇒ nothing claimed.** The drain pops an entry only + while it lives and re-inserts it at head on ANY non-dispatch outcome + (rejection or crash), so a dead/absent drain means every accepted + entry is on this list — the route's barrier term pair + (``_pending_sends`` non-empty OR drain alive) therefore covers the + claimed-entry window with no third state. + + (No ``priority`` field: dispatch is strictly FIFO — arrival order is + the contract — and the queued response/event use the route's parsed + local. A deferred entry's ``!!!`` prefix still reaches the model: + the full text dispatches as an ordinary send.) """ msg_id: str - priority: str attempt: Callable[[ChatSession], tuple[bool, dict[str, Any]]] retracted: bool = False @@ -4152,6 +4164,25 @@ def _make_dispatch_attempt( ws.id[:8] if ws.id else "", exc_info=True, ) + if ok and defer_fidelity and "rejected" not in queue_outcome and cfg.emit_message_queued: + # A deferred entry actually dispatched — the settle signal the + # panes' queued chips wait on (the busy→idle sweep skips + # deferred chips: "idle ⇒ drained" is untrue for them). Lives + # HERE, not in the drain, which is endpoint-agnostic by design: + # this arm has ``cfg``/``ui`` in scope and fires for BOTH + # dispatch shapes. ``folded`` marks the interjection fold-in + # (non-empty outcome): the message moved to the live turn's + # queue where DELETE still genuinely removes it, so the client + # clears only its deferred flag and lets the chip resume the + # normal interjection lifecycle — a flat promote there would + # strip the ✕ while retraction is still honored. Pane-tier + # like ``message_queued`` (not SDK-typed); best-effort via + # _emit_send_ui — an emission failure must not look like a + # dispatch failure (the drain would re-insert and DOUBLE-send). + event: dict[str, Any] = {"type": "message_dispatched", "msg_id": send_id} + if queue_outcome: + event["folded"] = True + _emit_send_ui(ws, ui, "_enqueue", event) return ok, queue_outcome return attempt @@ -4182,14 +4213,28 @@ def _drain_pending_sends(ws: Workstream) -> None: 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. + before its dispatch attempt and re-inserted at head on ANY + non-dispatch outcome — rejection or a crash inside the attempt — so + the DELETE fall-through (which marks only in-list entries) can never + "remove" a message whose dispatch already left the station, and the + :class:`_PendingSend` invariant (drain not alive ⇒ nothing claimed) + holds on every exit path. A claimed entry answers ``not_found`` + ("already sent"), which its eventual dispatch makes true. + + Clean-exit wake backstop: the drain's retirement is the moment the + /send order barrier clears, and a list that empties by RETRACTION + never runs a deferred turn — so no worker exit would ever re-run the + wake gate that yielded to us (see the pending-sends yield in + :func:`~turnstone.core.idle_nudge_watcher.wake_workstream_if_pending`). + Re-running the gate here, outside ``ws._lock`` (session_worker's + exit-backstop discipline), closes that strand; when entries DID + dispatch, it's a cheap no-op re-check after the last turn's own exit + backstop. """ import time + from turnstone.core.idle_nudge_watcher import wake_workstream_if_pending + try: while True: with ws._lock: @@ -4208,7 +4253,7 @@ def _drain_pending_sends(ws: Workstream) -> None: return if not pending: ws._pending_drain = None - return + break # clean exit — wake backstop below, outside the lock entry = pending[0] if ws._worker_running and ws.worker_kind == "command": # The park, relocated server-side: the poll cadence @@ -4221,14 +4266,34 @@ def _drain_pending_sends(ws: Workstream) -> None: # 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) + claimed = False + try: + 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 + claimed = True + ok, outcome = entry.attempt(session_now) + except Exception: + # An entry acknowledged "queued" must never be eaten by a + # crash (Thread.start under thread exhaustion, MemoryError + # in the dispatch path): restore the claim, back off, and + # retry — the docstring's no-give-up contract. ``claimed`` + # gates the re-insert so a claim-section failure can't + # duplicate the head entry. + log.exception( + "ws.send.pending_dispatch_crashed ws=%s msg_id=%s — entry retained", + ws.id[:8] if ws.id else "", + entry.msg_id, + ) + if claimed: + with ws._lock: + ws._pending_sends.insert(0, entry) + time.sleep(1.0) + continue if not ok or outcome.get("rejected") in ( "command_window", "defer_full_fidelity", @@ -4239,16 +4304,45 @@ def _drain_pending_sends(ws: Workstream) -> None: # 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. + # closed (resolved at the loop top). Unclaim, pace, wait. with ws._lock: ws._pending_sends.insert(0, entry) time.sleep(0.25) + if outcome.get("rejected") != "command_window": + # Every non-window rejection is stable for the CURRENT + # worker (cross-user / attachments / full-fidelity are + # per-turn structural; queue.Full clears only at the + # turn's drain seams and the entry is already acked, so + # turn-bounded delay is contract-legal) — wait on the + # cheap flags instead of re-running the full dispatch + # machinery against ``ws._lock`` at 4 Hz for the length + # of a turn. Lockless reads: DELETE only MARKS + # ``retracted`` (the loop-top purge under the lock is + # authoritative), a ``worker_kind`` flip to "command" + # exits into the window arm above, and force-cancel's + # flag-clear releases this exactly as it released the + # old park. One dispatch attempt per slot-state change. + while ( + ws._worker_running + and ws.worker_kind != "command" + and not entry.retracted + and not ws._closed + ): + time.sleep(0.25) continue # Dispatched: fresh spawn (empty outcome) or interjection - # fallback (msg_id preserved) — this entry is done. + # fallback (msg_id preserved) — this entry is done. The + # settle event (message_dispatched) fired inside the attempt. + wake_workstream_if_pending(ws, trigger="drain-exit") except Exception: # Never die holding the single-flight slot — a wedged drain would - # strand every future deferred send for this workstream. + # strand every future deferred send for this workstream. With the + # per-iteration handler above, reaching here means the loop + # machinery itself failed; entries stay on the list and the + # route's barrier arm re-ensures a drain on the next /send + # (deliberately NO successor spawn here: Thread.start fails under + # the same exhaustion that gets you here, and the route staying + # the single spawn site is what makes single-flight structural). log.exception("ws.send.pending_drain_failed ws=%s", ws.id[:8] if ws.id else "") with ws._lock: ws._pending_drain = None @@ -4290,12 +4384,15 @@ def make_send_handler(cfg: SessionEndpointConfig) -> Handler: 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), 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. + injection at the next tool-result seam), OR — with ``"deferred": + true`` — parked on ``ws._pending_sends`` (a slash-command window + holds the slot, or earlier deferred sends hold the order barrier) + and dispatched full-fidelity by :func:`_drain_pending_sends` when + the slot frees — see :class:`_PendingSend` for the durability + contract. ``DELETE {prefix}/{ws_id}/send`` with the ``msg_id`` + retracts either kind before dispatch; a deferred dispatch also + emits the pane-tier ``message_dispatched`` settle event (see + :func:`_make_dispatch_attempt`). - 200 ``{"status": "queue_full", "attached_ids", "dropped_attachment_ids"}`` — live worker's queue at capacity; reservations released. Caller should retry. The @@ -4402,8 +4499,6 @@ def make_send_handler(cfg: SessionEndpointConfig) -> Handler: await asyncio.sleep(0.1) if not ws._worker_running: break - if ws.session is None: - return JSONResponse({"error": "No session"}, status_code=500) # Defer-and-drain. While a slash-command worker holds the slot (a # manual /compact can hold it for MINUTES), a send must not take @@ -4416,26 +4511,25 @@ def make_send_handler(cfg: SessionEndpointConfig) -> Handler: # 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 per-workstream drain thread 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": + # + # Two triggers share ``_defer_send`` below: the command-window + # rejection (the attempt's enqueue closure reports it), and the + # ORDER BARRIER — once entries are pending (or a claimed entry's + # dispatch is in flight: the drain-alive term, backed by the + # _PendingSend invariant), the pending list is the order + # authority, and a fresh send lines up behind it instead of + # overtaking messages already acknowledged "queued". The barrier + # check and the append happen under ONE ``ws._lock`` acquisition — + # ws._lock is not reentrant and session_worker.send takes it, so + # the lock is always released before any dispatch attempt; the + # command-window trigger re-acquires for its append, which is safe + # because that rejection was reported under the lock the attempt + # itself held. + def _defer_send(*, require_barrier: bool) -> Response | None: # ``send_id`` is minted only when attachments are enabled — # the deferred entry needs a truthy id regardless: it is the # client's dismiss/bind handle and the drain's @@ -4444,7 +4538,6 @@ def make_send_handler(cfg: SessionEndpointConfig) -> Handler: cleaned_display, pending_priority = parse_priority(message) entry = _PendingSend( msg_id=pending_msg_id, - priority=pending_priority, attempt=_make_dispatch_attempt( ws, cfg, @@ -4459,6 +4552,10 @@ def make_send_handler(cfg: SessionEndpointConfig) -> Handler: ), ) with ws._lock: + if require_barrier: + drain_t = ws._pending_drain + if not ws._pending_sends and not (drain_t is not None and drain_t.is_alive()): + return None # no barrier — caller dispatches directly if ws._closed: # Mirror the dispatch-refusal 404 below: a "queued" # answer for a workstream whose next resolution 404s @@ -4467,6 +4564,8 @@ def make_send_handler(cfg: SessionEndpointConfig) -> Handler: ws._pending_sends.append(entry) drain = ws._pending_drain if drain is None or not drain.is_alive(): + # Single drain-spawn site — also the recovery path for + # a drain that died in its last-resort handler. t = threading.Thread( target=_drain_pending_sends, args=(ws,), @@ -4487,6 +4586,11 @@ def make_send_handler(cfg: SessionEndpointConfig) -> Handler: return JSONResponse( { "status": "queued", + # Parked on ws._pending_sends, NOT in a live turn's + # interjection queue: the panes keep the chip's ✕ past + # the busy→idle edge until message_dispatched settles + # it. See SendResponse for the SDK-facing contract. + "deferred": True, "priority": pending_priority, "msg_id": pending_msg_id, "attached_ids": list(ordered_taken), @@ -4495,6 +4599,31 @@ def make_send_handler(cfg: SessionEndpointConfig) -> Handler: ], } ) + + barrier_resp = _defer_send(require_barrier=True) + if barrier_resp is not None: + return barrier_resp + + 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": + window_resp = _defer_send(require_barrier=False) + if window_resp is None: # unreachable: only the barrier probe returns None + return JSONResponse({"error": "defer failed"}, status_code=500) + return window_resp if not ok: if ws._closed: # ``send`` refused because the workstream closed between our diff --git a/turnstone/server.py b/turnstone/server.py index 3a2d7ee3..837c1535 100644 --- a/turnstone/server.py +++ b/turnstone/server.py @@ -698,6 +698,14 @@ def _interactive_dispatch_retry(ws: Workstream, user_msg: str) -> None: the pre-lift inline behaviour). The shared dispatcher owns the ``_worker_running`` lifecycle, so the ``run`` closure needs no ``finally`` flag-clear of its own. + + Deliberately NOT gated on the /send order barrier + (``ws._pending_sends``): a retry is an explicit user action that + rewinds a COMPLETED turn — dispatching it ahead of deferred sends is + an accepted overtake (the user just asked for exactly that turn to + run again), not the silent send-vs-send inversion the barrier exists + to prevent. Deferred entries dispatch after it, order among + themselves preserved. """ from turnstone.core import session_worker @@ -1787,7 +1795,7 @@ async def command(request: Request) -> JSONResponse: # Fire-and-forget: the response returns as soon as the worker is # dispatched and NO completion bound applies — a large context # can legitimately compact for many minutes; progress streams - # over SSE and Stop cancels it. (The 60s wait below is for the + # over SSE and Stop cancels it. (The 25s wait below is for the # quick commands only.) def _run_compact() -> None: @@ -1852,9 +1860,9 @@ async def command(request: Request) -> JSONResponse: try: should_exit = session.handle_command(cmd) # Post-command follow-ups run HERE, on the worker, not - # after the endpoint's done-wait: past the 60s backstop the + # after the endpoint's done-wait: past the 25s backstop the # endpoint has already answered {"status": "running"}, and - # follow-ups parked there were silently skipped — a >60s + # follow-ups parked there were silently skipped — a slow # /resume left every pane rendering the pre-resume # transcript against a session whose history had changed, # and the workstream list kept the stale name. @@ -1907,9 +1915,15 @@ async def command(request: Request) -> JSONResponse: # the worker itself, so a late completion still refreshes the # panes. Loop-native wait (call_soon_threadsafe from the worker's # finally): a thread parked in Event.wait via to_thread would hold - # a shared default-executor slot for up to 60s per wedged command. + # a shared default-executor slot for the whole wait per wedged + # command. 25s: strictly under the console proxy's 30s client + # timeout (console/server.py proxy_client) so the degraded + # ``running`` answer can actually traverse a proxied pane — at 60s + # the proxy aborted first, the pane's dispatch swallowed the 5xx, + # and the user saw nothing at all (the same bounded-caller + # reasoning that redesigned /send's command-window path). try: - async with asyncio.timeout(60): + async with asyncio.timeout(25): await done.wait() except TimeoutError: return JSONResponse({"status": "running"}) @@ -2537,6 +2551,9 @@ async def _interactive_create_post_install( # creation) unless a caller-supplied ws_id is raced — hence # ``_enqueue_init`` preserves the message and leaves the staged # attachments recoverable instead of assuming the branch is dead. + # No /send order-barrier pre-check for the same by-construction + # reason: a freshly created workstream cannot have deferred sends + # pending (``ws._pending_sends`` only ever grows via that route). init_ok = session_worker.send( ws, enqueue=_enqueue_init, diff --git a/turnstone/shared_static/composer_queue.js b/turnstone/shared_static/composer_queue.js index 85418926..0abff579 100644 --- a/turnstone/shared_static/composer_queue.js +++ b/turnstone/shared_static/composer_queue.js @@ -57,20 +57,40 @@ * Returned controller surface: * addQueuedMessage(text, priority) -> el * priority: "important" | anything-else (treated as "notice") - * bind(el, msgId) + * bind(el, msgId, opts) * Server returned status:queued + msg_id. Stamps msgId so the × can * dequeue; if the user already clicked × (pre-bind) runs the * confirming delete now; if the idle sweep already promoted the * bubble (already delivered), leaves it untouched. + * opts (optional): { deferred, attachedCount } from the send + * response — deferred chips are skipped by the idle sweep (they + * settle via settleDeferred instead), and attachedCount feeds the + * discarded-attachments notice on a confirmed retract. This is the + * ONLY channel for both facts; consumers must not stash expandos + * on the element. * promote(el) * Settle an optimistic bubble as a normal sent message — used by the * consumer when the send response wasn't "queued" (e.g. an "ok" * stale-busy race) so a pre-bind × can't strand the card. * remove(el) * Drop the bubble (busy / queue_full / connection-error path). + * settleDeferred(msgId, folded) + * Consume a `message_dispatched` pane event: the deferred send left + * the parked list. folded=false → fresh turn spawned; promote (the + * ×'s window is over — a late DELETE resolves via not_found). + * folded=true → interjection fold-in; clear ONLY the deferred flag + * so the chip resumes the normal queued lifecycle (DELETE still + * genuinely removes it until the seam drains; the idle sweep + * promotes it at the turn edge). No-op when no live chip carries + * msgId (other tabs, replays) or a dismiss is in flight. * onIdleEdge() * Caller invokes once per busy → idle transition. Promotes every * not-in-flight queued bubble and then fires the onIdle hook. + * Skips deferred chips ("idle ⇒ drained" is untrue for them — they + * dispatch when the server-side drain runs, possibly minutes later) + * and unbound chips (POST round-trip still in flight — a quick + * command window can close inside it; the response arms settle + * every unbound chip, so the sweep never needs to). */ export function createQueueController(opts) { if (!opts || !opts.messagesEl) @@ -98,6 +118,14 @@ export function createQueueController(opts) { // Live queued bubbles — the idle sweep iterates this instead of querying // the whole messages container (see onIdleEdge). var _liveQueued = new Set(); + // Settles that arrived before bind() could stamp the chip: the order + // barrier lets a deferred entry dispatch within milliseconds of its ack, + // so the SSE message_dispatched can beat the POST response's .then — + // without this, the chip would stay flagged deferred and the idle sweep + // would skip it forever. bind() reconciles and deletes; capped small + // because chip-absent settles (other tabs, replays) also land here and + // never get consumed. + var _preBindSettles = new Map(); // msgId -> folded // Upper bound on the dequeue DELETE so a wedged proxied node (the exact // case this flow targets) can't leave a card stuck "dismissing" forever. var DELETE_TIMEOUT_MS = 15000; @@ -268,11 +296,13 @@ export function createQueueController(opts) { // 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) + // Say so instead of letting the files silently expire. The count + // rides bind()'s opts (dataset), never an element expando. + var nAtt = parseInt(el.dataset.attachedCount || "0", 10); + if (nAtt > 0 && onNotice) onNotice( "Message removed. Its " + - el._deferredAttachments + + nAtt + " attachment(s) were discarded — re-attach them to send again.", ); // `removed` is the only verdict that mutated server-side queue @@ -317,16 +347,34 @@ export function createQueueController(opts) { } // Server returned status:queued + msg_id. - function bind(el, msgId) { + function bind(el, msgId, opts) { if (!el || !msgId) return; // Idle sweep already promoted the bubble (worker drained mid-POST): the - // message is on its way, so leave it as a sent bubble. We deliberately do - // NOT delete here — idle can be emitted before the worker actually drains, - // so a delete could cancel a still-queued message; and a dismissed card - // never reaches this branch (onIdleEdge skips aria-busy cards), so doing - // nothing is the safe action. + // message is on its way, so leave it as a sent bubble. Dead-but-harmless + // since the sweep skips unbound chips; kept as the safe action for any + // promote path we didn't enumerate. We deliberately do NOT delete here — + // a delete could cancel a still-queued message; and a dismissed card + // never reaches this branch (onIdleEdge skips aria-busy cards). if (!el.classList.contains("msg-queued")) return; el.dataset.msgId = msgId; + if (opts && opts.deferred) el.dataset.deferred = "1"; + if (opts && opts.attachedCount > 0) + el.dataset.attachedCount = String(opts.attachedCount); + // A settle raced ahead of this bind (see _preBindSettles): apply it + // now. Fresh-spawn settle → the dispatch already happened, promote + // (fires "already sent" if the × was clicked — the dispatch won); + // fold-in settle → clear the flag, the chip is a normal queued + // interjection from here. + if (el.dataset.deferred && _preBindSettles.has(msgId)) { + var racedFolded = _preBindSettles.get(msgId); + _preBindSettles.delete(msgId); + if (racedFolded) { + delete el.dataset.deferred; + } else { + _promote(el); + return; + } + } // User clicked × before the id arrived → confirm the delete now // (removes only on a confirmed `removed`; promotes on not_found). if (el.dataset.dismissAttempted) _confirmDequeue(el, msgId); @@ -348,6 +396,8 @@ export function createQueueController(opts) { var attempted = el.dataset.dismissAttempted; el.classList.remove("msg-queued", "msg-queued-important"); delete el.dataset.msgId; + delete el.dataset.deferred; + delete el.dataset.attachedCount; delete el.dataset.dismissAttempted; el.removeAttribute("role"); el.removeAttribute("aria-label"); @@ -378,16 +428,55 @@ export function createQueueController(opts) { return; } if (el.hasAttribute("aria-busy")) return; // mid-dequeue — let it settle + // Deferred sends outlive the busy→idle edge: they dispatch when the + // server-side drain runs (message_dispatched → settleDeferred), and + // promoting here would strip the × while DELETE still genuinely + // retracts — presenting a parked message as sent, which on a node + // restart before dispatch becomes loss disguised as delivery. + if (el.dataset.deferred) return; + // Unbound chips (no msg_id — the POST round-trip is still in + // flight): a quick command window can close inside the round-trip, + // firing this edge before bind() could stamp the deferred flag. + // Every response arm (bind/promote/remove/catch) settles unbound + // chips, so the sweep never needs them. + if (!el.dataset.msgId) return; _promote(el); }); if (onIdle) onIdle(); } + // Consume a `message_dispatched {msg_id, folded}` pane event — see the + // header contract. Promote on a fresh spawn; on a fold-in clear only the + // deferred flag so the chip re-enters the normal interjection lifecycle. + function settleDeferred(msgId, folded) { + if (!msgId) return; + var target = null; + _liveQueued.forEach(function (el) { + if (el.isConnected && el.dataset.msgId === msgId) target = el; + }); + if (!target) { + // No bound chip yet: either another tab's message (never consumed — + // hence the cap) or this tab's bind() is still in the POST + // round-trip; park the settle for bind() to reconcile. + _preBindSettles.set(msgId, !!folded); + if (_preBindSettles.size > 8) + _preBindSettles.delete(_preBindSettles.keys().next().value); + return; + } + if (target.hasAttribute("aria-busy")) return; // dismiss in flight — let it settle + if (folded) { + delete target.dataset.deferred; + return; + } + _promote(target); + } + return { addQueuedMessage: addQueuedMessage, bind: bind, promote: _promote, remove: remove, + settleDeferred: settleDeferred, onIdleEdge: onIdleEdge, }; } diff --git a/turnstone/shared_static/conversation.js b/turnstone/shared_static/conversation.js index d50ddc32..cf291bd2 100644 --- a/turnstone/shared_static/conversation.js +++ b/turnstone/shared_static/conversation.js @@ -265,6 +265,7 @@ export function applyCompactionEvent(holder, evt, hooks) { ), ); if (eid) hooks.renderedIds.add(eid); + hooks.scroll(true); } else if (owns && evt.notice) { // cancelled / not_enough_messages / irreducible / empty_summary — // informational, not an error state. Whether the message is shown @@ -281,8 +282,12 @@ export function applyCompactionEvent(holder, evt, hooks) { // (A superseded OK end above still renders its result card — the // history swap really happened.) hooks.onNotice(evt.message || "Compaction skipped."); + hooks.scroll(true); } - hooks.scroll(true); + // No follow-scroll on the remaining paths: a non-owning failed end + // renders nothing (scrolling would yank the view with zero visual + // change — the round-7 finding), and an owning teardown only REMOVES + // the progress card, which needs no scroll chase. } } diff --git a/turnstone/shared_static/interactive.js b/turnstone/shared_static/interactive.js index bfbb8fe6..760ff23e 100644 --- a/turnstone/shared_static/interactive.js +++ b/turnstone/shared_static/interactive.js @@ -640,6 +640,9 @@ class Pane { this._addUserMsgActions(el, text); this.messagesEl.appendChild(el); this.scrollToBottom(true); + // Returned so the send flow can retro-convert the optimistic bubble + // into a queued chip when the server answers queued+deferred. + return el; } // --- Approval-cycle bookkeeping ----------------------------------------- @@ -1944,6 +1947,15 @@ class Pane { // The UI already showed the message optimistically in addQueuedMessage. break; + case "message_dispatched": + // A deferred send left the parked list: fresh spawn (promote the + // chip — the ×'s window is over) or interjection fold-in + // (folded: true — only the deferred flag clears; the chip resumes + // the normal queued lifecycle). No-op when this tab holds no + // matching chip. + this.queue.settleDeferred(evt.msg_id, !!evt.folded); + break; + case "busy_error": // Server is still busy — don't transition to send mode. // Re-enable the stop button so the user can try cancelling. @@ -3715,6 +3727,14 @@ class Pane { this.addInfoMessage( body.error || "Session is busy — try again shortly.", ); + } else if (body.status === "running") { + // The command outlived the endpoint's 25s completion backstop + // (kept under the console proxy's 30s bound precisely so this + // answer can traverse a proxied pane). The worker is still + // going; its output and pane refreshes arrive over SSE. + this.addInfoMessage( + "Command is still running — results will appear here when it finishes.", + ); } }) .catch(() => {}); @@ -3727,22 +3747,26 @@ class Pane { const isBusy = this.busy; let queuedEl = null; + let optimisticEl = null; const snap = this.attachments.snapshot(); + // Server re-parses the !!! prefix to set queue priority — the + // optimistic bubble strips it for display. Parsed outside the busy + // branch: the queued+deferred response arm needs it too (an idle + // pane's send can defer behind a command window / pending list). + let displayText = text; + let priority = "notice"; + if (text.startsWith("!!!")) { + displayText = text.slice(3).trimStart(); + priority = "important"; + } + if (isBusy) { - // Server re-parses the !!! prefix to set queue priority — the - // optimistic bubble strips it for display. - let displayText = text; - let priority = "notice"; - if (text.startsWith("!!!")) { - displayText = text.slice(3).trimStart(); - priority = "important"; - } this.removeEmptyState(); queuedEl = this.queue.addQueuedMessage(displayText, priority); } else { this.setBusy(true); - this.addUserMessage(text, snap.attachments); + optimisticEl = this.addUserMessage(text, snap.attachments); } this.composer.clear(); @@ -3819,20 +3843,23 @@ class Pane { if (data.status === "queued" && data.msg_id) { // queuedEl-present path: bind() handles the three known races // (pre-bind dismiss, promote sweep raced ahead, normal accept). - // queuedEl-absent path: client thought it was idle but the - // server saw a live worker (SSE state_change hadn't arrived - // yet). Flip busy so subsequent sends queue correctly; the - // 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). + // queuedEl-absent + deferred: the pane thought it was idle but + // the send parked on the server's deferred list (command window + // / order barrier) — a plain sent bubble would present a parked, + // still-retractable, restart-droppable message as delivered, so + // replace the optimistic bubble with a real queued chip carrying + // the dismiss affordance. queuedEl-absent + undeferred: plain + // interjection into a live turn the client hadn't seen yet; + // keep the historical small-UX-gap behavior (busy flip only). + if (!queuedEl && data.deferred) { + if (optimisticEl && optimisticEl.isConnected) optimisticEl.remove(); + queuedEl = this.queue.addQueuedMessage(displayText, priority); + } if (queuedEl) { - // 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); + this.queue.bind(queuedEl, data.msg_id, { + deferred: !!data.deferred, + attachedCount: (data.attached_ids || []).length, + }); } else this.setBusy(true); this.attachments.consume( data.attached_ids,