fix(compaction): review round 7 — drain crash/order/settle rows, protocol-default fallback, bool event-id guard

Completes the defer-and-drain seam against the matrix rows round 6 never
enumerated (the defer contract itself took no hits):

- Crash row: a claimed entry survives a dispatch crash — the
  per-iteration handler re-inserts it at head (claim-flagged so a
  claim-section failure can't duplicate it), backs off ~1s, retries.
  The last-resort handler spawns no successor (Thread.start fails under
  the exact exhaustion that reaches it): the route's ensure-drain stays
  the single spawn site, so single-flight is structural and a dead
  drain revives on the next defer.
- Order row: the pending list is the order authority. The /send route
  pre-checks pending/drain-alive under the same lock acquisition that
  appends (one _defer_send helper serves the barrier and command-window
  triggers); the coordinator adapter refuses via its return value; the
  queued-nudge wake gate yields to pending sends and is re-armed by the
  drain's clean exit — which covers lists emptied by pure retraction —
  as well as every deferred turn's exit; retry-after-rewind is a
  documented accepted overtake; init/create is fresh-ws-by-construction.
- Client settle row: queued responses carry "deferred": true
  (SendResponse + regenerated openapi-server.json, status enumeration
  completed); bind(el, msgId, {deferred, attachedCount}) replaces the
  _deferredAttachments expando; the idle sweep skips deferred and
  unbound chips; the shared dispatch attempt emits pane-tier
  message_dispatched (folded: true for interjection fold-ins — the chip
  clears only its deferred flag and keeps a live x while DELETE still
  genuinely retracts); settles that beat bind() park in a bounded
  buffer; an idle-thinking pane retro-converts its optimistic bubble
  into a real queued chip instead of presenting a parked message as
  sent. Rejection polling waits on the slot flags — one dispatch
  attempt per slot-state change, not 4 Hz.
- SessionUI.on_compaction's protocol stub became a real default body
  (the classic on_info rendering): explicit subclasses inherit protocol
  members as real methods, which defeated _compaction_event's getattr
  fallback for exactly the pre-1.8 embedders it serves.
- _coerce_event_id() rejects bools (isinstance(True, int) is True) at
  all three duck-typed event-id coercions: the compaction marker stamp,
  on_system_turn's persisted return, and _ui_event_id.
- Quick-command backstop 60s -> 25s, under the console proxy's 30s so
  the degraded "running" answer can traverse a proxied pane (which now
  surfaces it); /resume docs drop the fictional history SSE event
  (clear_ui + REST re-fetch is the contract); /send response docs match
  the wire.

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

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"}`.
 

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": {

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

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

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"

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

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

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
 # ---------------------------------------------------------------------------

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,

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,

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

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

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
 

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

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,

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,
   };
 }

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.
   }
 }
 

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,