fix(compaction): review round 6 — defer-and-drain send windows, workstream-scoped notify, ERROR badge survives /compact

Replace park-and-abandon /send semantics with defer-and-drain: a send
landing in a command window is answered {status: queued, msg_id}
immediately and dispatched full-fidelity by a per-workstream drain
thread when the window closes. Parking encoded client disconnect as
message retraction — true only for the composer's ✕-abort; every
bounded caller (coordinator client and console proxy at timeout=30,
SDKs, stock proxies) timed out and lost its message for the whole
window, and the compensating client machinery was racy (one-shot
sendAbortMs sample) and over-broad (_sendAbort fired on the
interjection path, dispatching dismissed messages while showing a
connection error). Dismissal is now uniformly bind() → DELETE, with a
fall-through that retracts pending entries; retracting an
attachment-bearing deferred send surfaces the discarded-attachments
consequence. The drain claims entries under ws._lock immediately
before dispatch (DELETE can never remove an in-flight message),
refuses the truncating interjection fallback for oversized or
attachment entries atomically inside the enqueue callback, and never
gives up while the workstream lives; durability is documented as
node-local at-most-once. sendAbortMs, _sendAbort, the 600s bound and
the park loop are deleted; route and drain share one dispatch
implementation (spawn metrics included).

Also: the initial-send completion notify is un-gated from slot
ownership (_fire_notify_targets has exactly one call site — successor
turns never notify, so the round-5 guard prevented a duplicate that
cannot exist while converting force-cancel into permanent notification
loss for scheduled workstreams); /compact on an ERROR workstream
restores the badge instead of stamping idle over it; duck-typed
SessionUIs without on_compaction get the classic on_info lines back
via a shared renderer (superseded OK ends included — a committed swap
must never be silent; pre-1.8 SSE clients are deliberately not
dual-emitted, documented as a 1.8 breaking change); failed-end notice
suppression is computed once by the emitter as a notice bool on the
end event (SDK py+ts), replacing the hand-synced cli/JS policy while
the panes keep their pane-local card-ownership clause.

CHANGELOG.md

@@ -199,14 +199,22 @@ Earlier stable lines (`stable/1.6`, `stable/1.5`) are frozen.
   summary HTTP call itself (the compaction lane registers its stream in
   the same abort seam the main loop uses), so cancelling a compaction is
   immediate instead of waiting out a model call. Messages sent while any
-  slash command holds the worker slot **park** and then run as ordinary
-  full-fidelity sends when the command finishes — they are never routed
-  through the mid-turn interjection queue, whose semantics are
-  turn-shaped: previously a send during a manual `/compact` was silently
-  truncated to 2,000 characters, a second participant in a shared
-  workstream was locked out with a misleading "another participant's
-  turn" 409 for the whole compaction, and a message queued across a
-  `/resume`/`/new` could be answered into the post-swap workstream.
+  slash command holds the worker slot are **deferred**: answered
+  `{"status": "queued", "msg_id"}` immediately and dispatched as ordinary
+  full-fidelity sends (attachments and sender identity included) when the
+  command finishes — never routed through the mid-turn interjection
+  queue, whose semantics are turn-shaped: previously a send during a
+  manual `/compact` was silently truncated to 2,000 characters, a second
+  participant in a shared workstream was locked out with a misleading
+  "another participant's turn" 409 for the whole compaction, and a
+  message queued across a `/resume`/`/new` could be answered into the
+  post-swap workstream. Because the response is immediate,
+  timeout-bounded callers — the coordinator's `send_message`, the console
+  proxy, SDKs, anything behind a stock reverse proxy — can no longer lose
+  a message to a multi-minute command window; the deferred send is
+  retractable until dispatch via the same `DELETE .../send` used for
+  queued interjections (node-local, in-memory — the API reference
+  documents the at-most-once durability contract).
   A `/compact` raced against an in-flight turn is refused with an
   explicit busy response. Every other slash command runs through the same
   worker slot too — mutual exclusion against sends, a running compaction,
@@ -230,17 +238,31 @@ Earlier stable lines (`stable/1.6`, `stable/1.5`) are frozen.
   compaction (a new turn's generation claim breaks a stale latch). Every
   retry backoff on the session (stream retries, task agents, notify
   delivery, compaction) now aborts immediately on Stop via one shared
-  cancel-aware helper instead of sleeping out its exponential delay. The
-  web composer bounds a parked send at 10 minutes while a compaction card
-  is visible (15 s wedged-node default otherwise, shared by both panes),
-  and dismissing a queued bubble aborts the in-flight POST so a dismissed
-  message can't dispatch minutes later. A compaction failure reports
+  cancel-aware helper instead of sleeping out its exponential delay.
+  Dismissing a queued bubble — interjection or deferred — is a
+  server-confirmed `DELETE`, and retracting a deferred send that carried
+  attachments tells the user they were discarded instead of silently
+  expiring them. A compaction failure reports
   exactly once (auto-compaction errors defer to the turn's fatal handler
-  instead of doubling the red row and the error metric), and a manual
-  `/compact` failure no longer crashes the CLI REPL. Post-command pane
-  refreshes and error notices are owner-guarded, so a force-cancelled
-  wedged command that unwedges late can't wipe panes or inject stray
-  notices into a successor turn.
+  instead of doubling the red row and the error metric), failed-end
+  notice suppression is computed once by the emitter (a `notice` bool on
+  the end event — in the SDKs — replaces hand-synced client policy), and
+  a manual `/compact` failure no longer crashes the CLI REPL. `/compact`
+  on a workstream showing the `error` badge restores the badge on exit
+  instead of stamping `idle` over it (the compaction neither retried nor
+  resolved the failed turn). A force-cancelled initial send that
+  completes late still delivers its scheduled-run completion
+  notification (the only completion signal unattended workstreams have);
+  the other post-command pane refreshes and error notices remain
+  owner-guarded, so a force-cancelled wedged command that unwedges late
+  can't wipe panes or inject stray notices into a successor turn.
+  Embedders driving `ChatSession` with a pre-1.8 duck-typed `SessionUI`
+  (no `on_compaction` hook) get the classic `on_info` compaction lines
+  back — threshold notice, `part k/N`, retry waits, token delta +
+  summary box — instead of silent history swaps. **Breaking (1.8):**
+  compaction feedback moved from `info` events to the typed `compaction`
+  SSE event; pre-1.8 SSE/SDK clients that ignore unknown event types no
+  longer see compaction lines (they are deliberately not dual-emitted).
 
 - **Static MCP servers: a pushed catalog change no longer wedges the shared
   session (#839).** The static-path `*/list_changed` handler awaited its

docs/api-reference.md

@@ -482,13 +482,16 @@ carries `before_tokens`/`after_tokens` and the produced `summary`;
 `"cancelled"` / `"error"`) and a human-readable `message` — for
 `reason: "error"` the same message is also emitted as a paired typed
 `error` event (that is the renderable error surface; the end event is
-card-teardown). Every end (ok or failed) carries `trigger`, and every
-event carries `compaction_id` — an opaque integer correlating the
-start/progress/end of one compaction run (a client that force-stopped one
-compaction can use it to ignore stragglers from the abandoned run). End
-events also carry `superseded`: `true` marks a force-abandoned compaction
-retiring after a successor generation took over — skip failure notices for
-those (an OK end's result card still stands: the history swap happened).
+card-teardown). Failed ends also carry `notice`: the emitter-computed
+display verdict — show `message` only when it is `true` (the server
+suppresses error-reason, superseded, and cancelled-auto notices once,
+centrally, so clients don't re-derive that policy). Every end (ok or
+failed) carries `trigger`, and every event carries `compaction_id` — an
+opaque integer correlating the start/progress/end of one compaction run (a
+client that force-stopped one compaction can use it to ignore stragglers
+from the abandoned run). End events also carry `superseded`: `true` marks
+a force-abandoned compaction retiring after a successor generation took
+over (an OK end's result card still stands: the history swap happened).
 Superseded start/progress events are never emitted.
 Exactly one `start` and one `end` are emitted per attempt,
 so clients can key an in-progress affordance (progress bar) on the pair. A
@@ -875,19 +878,25 @@ synchronous:
   endpoint executed commands unconditionally mid-turn; the 409 makes the
   refusal loud for callers that only check the HTTP status.)
 
-While a command holds the slot, `POST .../send` requests **park** server-side
-and dispatch as ordinary full-fidelity sends when the command's window closes
-— they are never routed through the mid-turn interjection queue (no length
-cap, no cross-user rejection). A client that aborts a parked send before the
-window closes abandons it: the message is not dispatched. Clients must
-therefore bound parked sends generously: the bundled web composer uses a
-10-minute abort while a compaction progress card is visible (its usual bound
-is ~15 s, sized for wedged-node detection), and dismissing a queued bubble
-aborts the in-flight POST so a dismissed message can't dispatch later.
-Deployment note: a reverse proxy's read timeout bounds the effective park —
-behind a stock 60 s proxy, sends parked longer than that fail at the proxy
-(the park sees the disconnect and abandons; nothing is dispatched — resend
-after the command completes, or raise the proxy read timeout).
+While a command holds the slot, `POST .../send` requests are **deferred**:
+the server answers `{"status": "queued", "msg_id": ...}` immediately and
+dispatches the message as an ordinary full-fidelity send (attachments and
+sender identity included) when the command's window closes — it is never
+routed through the mid-turn interjection queue (no length cap, no cross-user
+rejection). The response arrives within normal round-trip time, so
+timeout-bounded clients (SDKs, proxies, the coordinator) need no special
+handling. To retract a deferred send before it dispatches, issue the same
+`DELETE .../send` with its `msg_id` used for queued interjections —
+`{"status": "removed"}` confirms it will not dispatch; `"not_found"` means
+it already dispatched (or is dispatching). Retracting a deferred send
+discards any attachments it carried; re-attach to send them again.
+
+Durability: deferred sends are **node-local and in-memory** (the same
+lifetime as the interjection queue). `"queued"` is at-most-once intake, not
+durable acceptance — if the workstream is closed or the node restarts before
+the window ends, the message is dropped. Anything that must survive a
+restart should be re-sent after confirming dispatch (the turn appears on the
+SSE stream / in `/history`).
 
 **Request body:**
 

sdk/typescript/src/events.ts

@@ -176,6 +176,12 @@ export interface CompactionEvent {
    * those (an OK end's result still stands; the history swap happened).
    */
   superseded?: boolean;
+  /**
+   * Failed ends only: the emitter-computed display verdict — show
+   * `message` only when true, instead of re-deriving suppression from
+   * reason/trigger/superseded client-side.
+   */
+  notice?: boolean;
   /** Present on start and on every end (ok or failed). */
   trigger?: "manual" | "auto";
   where?: string;

tests/test_cooperative_compaction.py

@@ -1891,7 +1891,9 @@ class TestPreHookUICompat:
         """A duck-typed SessionUI predating the compaction hook gets no
         lifecycle events but must not crash — an unguarded emit would
         AttributeError inside send()'s auto-compaction and permanently
-        wedge every long session on such a UI."""
+        wedge every long session on such a UI.  (This one has no on_info
+        either — the fallback below is getattr-guarded the same way, so
+        the never-crash floor holds even for a hookless minimal UI.)"""
         _seed_two_messages(session)
         session.ui = SimpleNamespace(
             on_thinking_start=lambda: None,
@@ -1902,6 +1904,193 @@ class TestPreHookUICompat:
         with patch.object(session, "_utility_completion", return_value=summary):
             assert session._compact_messages() is True
 
+    def _duck_ui_with_info(self):
+        infos: list[str] = []
+        ui = SimpleNamespace(
+            on_thinking_start=lambda: None,
+            on_thinking_stop=lambda: None,
+            on_error=lambda _m: None,
+            on_info=infos.append,
+        )
+        return ui, infos
+
+    def test_pre_hook_ui_gets_classic_info_lines(self, session):
+        """A duck-typed UI WITH on_info but WITHOUT on_compaction gets the
+        pre-1.8 lines back through the shared renderer — an auto-compaction
+        must never swap history with zero announcement for embedders that
+        predate the hook (the old lines reached them unconditionally)."""
+        session.ui, infos = self._duck_ui_with_info()
+        session._compaction_event(
+            0, {"phase": "start", "trigger": "auto", "where": "mid-turn", "pct": 80}
+        )
+        session._compaction_event(0, {"phase": "progress", "part": 1, "total": 2, "depth": 0})
+        session._compaction_event(
+            0,
+            {
+                "phase": "end",
+                "ok": True,
+                "trigger": "auto",
+                "before_tokens": 900,
+                "after_tokens": 100,
+                "summary": "dense",
+            },
+        )
+        assert any("Auto-compacting mid-turn" in m and "80%" in m for m in infos)
+        assert any("compacting part 1/2" in m for m in infos)
+        assert any("compacted: ~900 -> ~100 tokens" in m for m in infos)
+        assert any("dense" in m for m in infos)
+
+    def test_pre_hook_fallback_consumes_notice_not_policy(self, session):
+        """Failed-end display rides the emitter's ``notice`` stamp: a
+        manual cancelled end prints its message; an error-reason end (the
+        red row already came through on_error) and a cancelled AUTO end
+        stay silent — the fallback never re-derives the policy."""
+        session.ui, infos = self._duck_ui_with_info()
+        session._compaction_event(
+            0,
+            {
+                "phase": "end",
+                "ok": False,
+                "reason": "cancelled",
+                "message": "Compaction cancelled.",
+                "trigger": "manual",
+            },
+        )
+        assert infos == ["Compaction cancelled."]
+        infos.clear()
+        session._compaction_event(
+            0,
+            {
+                "phase": "end",
+                "ok": False,
+                "reason": "error",
+                "message": "boom",
+                "trigger": "manual",
+            },
+        )
+        session._compaction_event(
+            0,
+            {
+                "phase": "end",
+                "ok": False,
+                "reason": "cancelled",
+                "message": "Compaction cancelled.",
+                "trigger": "auto",
+            },
+        )
+        assert infos == []
+
+    def test_superseded_ok_end_still_renders_via_fallback(self, session):
+        """A superseded OK end announces a history swap that REALLY
+        committed — suppressing it re-creates the silent-swap failure for
+        duck-typed embedders (the web reducer renders the result card for
+        superseded OK ends for the same reason).  A superseded FAILED end
+        stays silent via the notice stamp."""
+        session.ui, infos = self._duck_ui_with_info()
+        session._generation = 7  # the emitting generation 3 is stale
+        session._compaction_event(
+            3,
+            {
+                "phase": "end",
+                "ok": True,
+                "trigger": "manual",
+                "before_tokens": 500,
+                "after_tokens": 50,
+                "summary": "kept",
+            },
+        )
+        assert any("compacted: ~500 -> ~50 tokens" in m for m in infos)
+        infos.clear()
+        session._compaction_event(
+            3,
+            {
+                "phase": "end",
+                "ok": False,
+                "reason": "cancelled",
+                "message": "Compaction cancelled.",
+                "trigger": "manual",
+            },
+        )
+        assert infos == []
+
+
+class TestCompactionNoticeStamp:
+    """_compaction_event is the single display-policy site: failed ends
+    carry ``notice`` — renderers show the message iff it is true, instead
+    of each re-deriving suppression from reason/trigger/superseded (the
+    cross-runtime drift trap the old hand-synced cli.py/conversation.js
+    clauses documented)."""
+
+    def _emit(self, session, payload, my_generation=0):
+        seen: dict = {}
+        with patch.object(session.ui, "on_compaction", side_effect=lambda p: seen.update(p)):
+            session._compaction_event(my_generation, payload)
+        return seen
+
+    def test_manual_cancelled_end_notice_true(self, session):
+        seen = self._emit(
+            session,
+            {
+                "phase": "end",
+                "ok": False,
+                "reason": "cancelled",
+                "message": "x",
+                "trigger": "manual",
+            },
+        )
+        assert seen["notice"] is True
+
+    def test_error_reason_end_notice_false(self, session):
+        """reason=error already fired the one red on_error row — the end
+        event is card-teardown, never a second line."""
+        seen = self._emit(
+            session,
+            {
+                "phase": "end",
+                "ok": False,
+                "reason": "error",
+                "message": "boom",
+                "trigger": "manual",
+            },
+        )
+        assert seen["notice"] is False
+
+    def test_cancelled_auto_end_notice_false(self, session):
+        """A cancelled AUTO compaction is part of cancelling the turn —
+        the send loop prints its own '[Generation cancelled]'."""
+        seen = self._emit(
+            session,
+            {"phase": "end", "ok": False, "reason": "cancelled", "message": "x", "trigger": "auto"},
+        )
+        assert seen["notice"] is False
+
+    def test_stale_generation_end_notice_false(self, session):
+        """A superseded end is one nobody is waiting on — its notice
+        mid-turn reads as the LIVE work being cancelled."""
+        session._generation = 9
+        seen = self._emit(
+            session,
+            {
+                "phase": "end",
+                "ok": False,
+                "reason": "cancelled",
+                "message": "x",
+                "trigger": "manual",
+            },
+            my_generation=4,
+        )
+        assert seen["superseded"] is True
+        assert seen["notice"] is False
+
+    def test_ok_end_and_start_carry_no_notice(self, session):
+        seen = self._emit(
+            session,
+            {"phase": "end", "ok": True, "trigger": "manual", "summary": "s"},
+        )
+        assert "notice" not in seen
+        seen = self._emit(session, {"phase": "start", "trigger": "manual"})
+        assert "notice" not in seen
+
 
 class TestPreSwapQueueFlush:
     def test_new_flushes_stranded_queue_into_old_workstream(self, session):

tests/test_interactive_pane_js.py

@@ -638,33 +638,39 @@ def test_connectsse_defers_open_when_tab_hidden() -> None:
     assert head.index('addEventListener("visibilitychange"') < head.index("if (document.hidden) {")
 
 
-def test_send_abort_bound_is_compaction_aware() -> None:
-    """The send POST's abort bound must be selected via the shared
-    ``sendAbortMs`` helper in BOTH panes: sends during a slash-command
-    window park server-side (a manual /compact legitimately runs for
-    minutes), so a hard-coded ~15s abort silently dropped any send made
-    >15s into a long compaction.  The compaction card is the long-bound
-    signal; the wedged-node default stays otherwise."""
+def test_send_post_abort_machinery_is_gone() -> None:
+    """The parked-POST era's client abort machinery must stay deleted in
+    BOTH panes: sends during a command window are answered "queued"
+    immediately (server-side defer-and-drain), so there is no long-lived
+    POST for a compaction-aware bound (``sendAbortMs``) to protect, and
+    dismissal is bind() → server-confirmed DELETE — never a POST abort
+    (``_sendAbort``), which fired on the interjection path too and
+    dispatched "dismissed" messages anyway.  Reintroducing either hook
+    means re-parking the POST; that design deterministically dropped
+    messages from every timeout-bounded caller (coordinator client and
+    console proxy at 30s, SDKs, stock proxies)."""
     interactive = _INTERACTIVE.read_text(encoding="utf-8")
-    assert "sendAbortMs(this._compaction)" in interactive, (
-        "interactive sendMessage must select its abort bound off the compaction holder"
-    )
     coordinator = (_ROOT / "turnstone/console/static/coordinator/coordinator.js").read_text(
         encoding="utf-8"
     )
-    assert "sendAbortMs(compactionHolder)" in coordinator, (
-        "coordinator coordSend must share the same abort-bound policy"
-    )
     conversation = (_ROOT / "turnstone/shared_static/conversation.js").read_text(encoding="utf-8")
-    assert "export function sendAbortMs" in conversation
-    assert "600000 : 15000" in conversation.replace("\n", " "), (
-        "long bound while a compaction card is live; 15s wedged-node default otherwise"
-    )
-    # The queued bubble's pre-response x must abort the in-flight (possibly
-    # parked) POST — otherwise a dismissed message dispatches anyway when
-    # the command window closes.
-    assert "queuedEl._sendAbort = () => sendCtrl.abort()" in interactive
     composer_queue = (_ROOT / "turnstone/shared_static/composer_queue.js").read_text(
         encoding="utf-8"
     )
-    assert "el._sendAbort()" in composer_queue
+    for name, src in (
+        ("interactive.js", interactive),
+        ("coordinator.js", coordinator),
+        ("conversation.js", conversation),
+        ("composer_queue.js", composer_queue),
+    ):
+        assert "sendAbortMs" not in src, f"{name}: the compaction-aware abort bound is dead"
+        assert "_sendAbort" not in src, f"{name}: dismiss must be bind() → DELETE, not a POST abort"
+    # The flat wedged-node bound stands in both panes: every /send answers
+    # within RTT now (dispatched / queued / deferred-with-msg_id).
+    assert "sendCtrl.abort(), 15000" in interactive
+    assert "sendCtrl.abort(), 15000" in coordinator
+    # Retracting a deferred send discards its attachments — both panes
+    # stash the count and composer_queue surfaces the consequence.
+    assert "_deferredAttachments" in interactive
+    assert "_deferredAttachments" in coordinator
+    assert "_deferredAttachments" in composer_queue

tests/test_server_authz.py

@@ -240,12 +240,25 @@ class _FakeSession:
         self.compact_gate: threading.Event | None = None
         self.command_gate: threading.Event | None = None
         self.command_raises: BaseException | None = None
-        # Every queue_message call — the park invariant is that command
+        # Gate for send() — lets a test hold a TURN in flight so a
+        # deferred entry provably lands in (or is refused by) the
+        # interjection fallback at drain time.
+        self.send_gate: threading.Event | None = None
+        # Every queue_message call — the defer invariant is that command
         # windows NEVER reach the interjection queue.
         self.queue_calls: list[str] = []
+        # When set, queue_message records the attempt and then raises it
+        # (e.g. CrossUserInterjectionError for the drain's re-park arm).
+        self.queue_raises: BaseException | None = None
+        # DELETE /send fall-through: ids the route asked this session to
+        # dequeue (the fake never holds interjections, so it returns
+        # False and the route falls through to the deferred-send list).
+        self.dequeues: list[str] = []
 
     def send(self, text: str, *, attachments: Any = None, send_id: Any = None) -> None:
         self.sends.append((text, attachments, send_id))
+        if self.send_gate is not None:
+            self.send_gate.wait(timeout=10)
         if self.send_raises is not None:
             raise self.send_raises
 
@@ -257,9 +270,15 @@ class _FakeSession:
         interjector_user_id: str = "",
     ) -> tuple[str, str, str]:
         self.queue_calls.append(text)
+        if self.queue_raises is not None:
+            raise self.queue_raises
         cleaned = text[:2000] + "..." if len(text) > 2000 else text
         return cleaned, "notice", queue_msg_id or "m1"
 
+    def dequeue_message(self, msg_id: str) -> bool:
+        self.dequeues.append(msg_id)
+        return False
+
     def set_watch_runner(self, *_a: Any, **_kw: Any) -> None:
         pass
 
@@ -1275,23 +1294,43 @@ class TestCompactCommandDispatch:
         assert ws.session.compacts == 1
         assert ws._worker_running is False
         # The slot was classified as a command window (what the /send
-        # route's park keys on; stale after exit is harmless — every
+        # route's defer keys on; stale after exit is harmless — every
         # reader conjoins _worker_running).
         assert ws.worker_kind == "command"
         # Exit seam: stranded-text backstop flush only — no drain, no
-        # answering send (sends during the window park in the /send route
-        # and dispatch as their own workers afterwards).
+        # answering send (sends during the window defer in the /send
+        # route and dispatch as their own workers afterwards).
         assert ws.session.queued_flushes == 1
         assert ws.session.sends == []
         # The worker wrapped the run in busy/idle state transitions.
         assert ws.ui.states == ["thinking", "idle"]
 
-    def test_send_during_compact_window_parks_then_dispatches_fully(self, app_client):
-        """A /send during a manual /compact PARKS and then runs as an
-        ordinary full-fidelity send — it must never enter the interjection
-        queue, whose 2000-char cap silently truncated pasted logs/code and
-        whose cross-user guard locked second participants out for the
-        whole compaction."""
+    def _wait_for(self, predicate, timeout: float = 8.0) -> bool:
+        """Poll until ``predicate()`` — the deferred-send drain runs on the
+        app's event loop at a 0.25s cadence, so dispatch is asynchronous
+        with respect to the released command worker."""
+        deadline = time.monotonic() + timeout
+        while time.monotonic() < deadline:
+            if predicate():
+                return True
+            time.sleep(0.02)
+        return False
+
+    def _drain_idle(self, ws) -> bool:
+        """True once the pending-send drain retired itself (list empty,
+        single-flight slot cleared) — the leaked-task guard for these
+        tests."""
+        with ws._lock:
+            return not ws._pending_sends and ws._pending_drain is None
+
+    def test_send_during_compact_window_defers_then_dispatches_fully(self, app_client):
+        """A /send during a manual /compact is answered "queued"
+        IMMEDIATELY (no parked POST — a 30s-bounded caller like the
+        coordinator client or console proxy must never lose a message to
+        a multi-minute window) and then runs as an ordinary full-fidelity
+        send — never the interjection queue, whose 2000-char cap silently
+        truncated pasted logs/code and whose cross-user guard locked
+        second participants out for the whole compaction."""
         client, mgr = app_client
         ws_id = self._create_ws(client)
         ws = mgr.get(ws_id)
@@ -1305,35 +1344,35 @@ class TestCompactCommandDispatch:
         )
         assert resp.json() == {"status": "ok"}
         big = "x" * 5000  # over the interjection cap — must survive intact
-        send_result: dict = {}
-
-        def _send() -> None:
-            r = client.post(
-                f"/v1/api/workstreams/{ws_id}/send",
-                json={"message": big},
-                headers=_auth("user-1"),
-            )
-            send_result["status"] = r.status_code
-            send_result["body"] = r.json()
-
-        sender = threading.Thread(target=_send, daemon=True)
-        sender.start()
-        # The send is parked: give it time to have taken the park path,
-        # then prove nothing was dispatched or queued yet.
-        for _ in range(50):
-            if ws.session.queue_calls or ws.session.sends:
-                break
-            time.sleep(0.02)
+        # The response is immediate even though the window is wedged open:
+        # this request runs on the same synchronous test client, so a
+        # parked POST would deadlock the test rather than pass it.
+        r = client.post(
+            f"/v1/api/workstreams/{ws_id}/send",
+            json={"message": big},
+            headers=_auth("user-1"),
+        )
+        assert r.status_code == 200
+        body = r.json()
+        assert body["status"] == "queued"
+        assert body["msg_id"]
+        assert body["priority"] == "notice"
+        # Registered, not dispatched: the window is still open.
         assert ws.session.sends == []
         assert ws.session.queue_calls == []  # the queue is unreachable
-        gate.set()  # compaction finishes; the park releases
-        sender.join(timeout=10)
-        assert not sender.is_alive()
-        assert send_result["status"] == 200
-        assert send_result["body"]["status"] == "ok"
-        # Full fidelity: the exact 5000-char text, via a normal send.
+        with ws._lock:
+            assert len(ws._pending_sends) == 1
+        gate.set()  # compaction finishes; the drain dispatches
+        assert self._wait_for(lambda: ws.session.sends)
+        # Full fidelity: the exact 5000-char text, via a normal send (an
+        # oversized entry must take the fresh-spawn arm — the interjection
+        # fallback would truncate it).
         assert [s[0] for s in ws.session.sends] == [big]
         assert ws.session.queue_calls == []
+        # The dispatched send carries the deferred msg_id as its send_id,
+        # so the client's queued bubble reconciles against the turn.
+        assert ws.session.sends[0][2] == body["msg_id"]
+        assert self._wait_for(lambda: self._drain_idle(ws))
 
     def test_cancelled_compact_flushes_queue_without_answering(self, app_client):
         """A user-stopped compaction must not auto-run a turn they may no
@@ -1413,10 +1452,10 @@ class TestCompactCommandDispatch:
         assert ws.session.sends == []
         assert ws.ui.states == []
 
-    def test_parked_send_delivers_attachments_after_release(self, app_client, monkeypatch):
-        """Attachments are peek-resolved BEFORE the park (the bytes ride the
-        request closure), so a send parked through a long compaction still
-        delivers them on dispatch — the pre-park refusal
+    def test_deferred_send_delivers_attachments_after_release(self, app_client, monkeypatch):
+        """Attachments are peek-resolved BEFORE the defer (the bytes ride
+        the pending entry), so a send deferred through a long compaction
+        still delivers them on dispatch — the queue-path refusal
         (attachments_busy) must never apply to a command window."""
         client, mgr = app_client
         ws_id = self._create_ws(client)
@@ -1435,34 +1474,31 @@ class TestCompactCommandDispatch:
             json={"command": "/compact", "ws_id": ws_id},
             headers=_auth("user-1"),
         )
-        send_result: dict = {}
-
-        def _send() -> None:
-            r = client.post(
-                f"/v1/api/workstreams/{ws_id}/send",
-                json={"message": "with attachment", "attachment_ids": ["a1"]},
-                headers=_auth("user-1"),
-            )
-            send_result["body"] = r.json()
-
-        sender = threading.Thread(target=_send, daemon=True)
-        sender.start()
-        time.sleep(0.3)
-        assert ws.session.sends == []  # still parked
+        r = client.post(
+            f"/v1/api/workstreams/{ws_id}/send",
+            json={"message": "with attachment", "attachment_ids": ["a1"]},
+            headers=_auth("user-1"),
+        )
+        body = r.json()
+        assert body["status"] == "queued"
+        # The attachments are TAKEN by the deferred entry (the client
+        # consumes its chips off this list — a retract discards them).
+        assert body["attached_ids"] == ["a1"]
+        assert ws.session.sends == []  # still deferred
         gate.set()
-        sender.join(timeout=10)
-        assert not sender.is_alive()
-        assert send_result["body"]["status"] == "ok"
-        assert send_result["body"]["attached_ids"] == ["a1"]
-        text, attachments, _sid = ws.session.sends[0]
+        assert self._wait_for(lambda: ws.session.sends)
+        text, attachments, sid = ws.session.sends[0]
         assert text == "with attachment"
         assert attachments == ["fake-attachment-bytes"]
+        assert sid == body["msg_id"]
         assert ws.session.queue_calls == []
+        assert self._wait_for(lambda: self._drain_idle(ws))
 
-    def test_send_during_quick_command_window_parks(self, app_client):
-        """The park applies to EVERY command window, not just /compact — a
-        send racing a quick command dispatches after the window with full
-        fidelity instead of entering the interjection queue."""
+    def test_send_during_quick_command_window_defers(self, app_client):
+        """The defer applies to EVERY command window, not just /compact —
+        a send racing a quick command is answered "queued" and dispatches
+        after the window with full fidelity instead of entering the
+        interjection queue."""
         client, mgr = app_client
         ws_id = self._create_ws(client)
         ws = mgr.get(ws_id)
@@ -1488,29 +1524,22 @@ class TestCompactCommandDispatch:
             time.sleep(0.02)
         assert ws._worker_running
         assert ws.worker_kind == "command"
-        send_result: dict = {}
-
-        def _send() -> None:
-            r = client.post(
-                f"/v1/api/workstreams/{ws_id}/send",
-                json={"message": "mid-command send"},
-                headers=_auth("user-1"),
-            )
-            send_result["body"] = r.json()
-
-        sender = threading.Thread(target=_send, daemon=True)
-        sender.start()
-        time.sleep(0.3)  # long enough for a queue-path regression to show
+        r = client.post(
+            f"/v1/api/workstreams/{ws_id}/send",
+            json={"message": "mid-command send"},
+            headers=_auth("user-1"),
+        )
+        send_body = r.json()
+        assert send_body["status"] == "queued"
         assert ws.session.queue_calls == []
         assert ws.session.sends == []
         gate.set()
         runner.join(timeout=10)
-        sender.join(timeout=10)
-        assert not sender.is_alive()
         assert cmd_result["body"] == {"status": "ok"}
-        assert send_result["body"]["status"] == "ok"
+        assert self._wait_for(lambda: ws.session.sends)
         assert [s[0] for s in ws.session.sends] == ["mid-command send"]
         assert ws.session.queue_calls == []
+        assert self._wait_for(lambda: self._drain_idle(ws))
 
     def test_clear_ui_rides_the_worker_not_the_endpoint(self, app_client):
         """The clear_ui follow-up runs on the worker after handle_command —
@@ -1590,10 +1619,294 @@ class TestCompactCommandDispatch:
         )
         assert ws.ui.errors == []
 
+    def test_abandoned_init_worker_still_fires_notify(self, app_client, monkeypatch):
+        """The completion notify is WORKSTREAM-scoped, not slot-scoped:
+        _fire_notify_targets has exactly one call site (the init worker),
+        successor turns never notify, so an owner gate here has no
+        duplicate to prevent — it only converts force-cancel into
+        permanent notification loss for scheduled/unattended workstreams.
+        This pins the un-gated behavior against the tempting symmetry
+        'fix' (the finally's siblings ARE owner-gated, correctly — they
+        mutate live slot/UI state)."""
+        client, mgr = app_client
+        fired: list = []
+        monkeypatch.setattr(
+            "turnstone.server._fire_notify_targets",
+            lambda ws, content: fired.append(ws.id),
+        )
+        gate = threading.Event()
+        started = threading.Event()
+        orig_send = _FakeSession.send
+
+        def wedged_send(self, text, **kwargs):
+            started.set()
+            assert gate.wait(timeout=10)
+            orig_send(self, text, **kwargs)
+
+        monkeypatch.setattr(_FakeSession, "send", wedged_send)
+        resp = client.post(
+            "/v1/api/workstreams/new",
+            json={"name": "sched", "initial_message": "do the thing"},
+            headers=_auth("user-1"),
+        )
+        assert resp.status_code == 200
+        ws_id = resp.json()["ws_id"]
+        ws = mgr.get(ws_id)
+        assert ws is not None
+        assert started.wait(timeout=10)
+        worker = ws.worker_thread
+        with ws._lock:  # the force-cancel abandon shape
+            ws.worker_thread = None
+            ws._worker_running = False
+        gate.set()  # the abandoned init completes LATE
+        worker.join(timeout=10)
+        assert not worker.is_alive()
+        assert fired == [ws_id]  # exactly once, despite the abandonment
+
+    def test_compact_on_error_workstream_restores_error_badge(self, app_client):
+        """/compact on an ERROR workstream must exit back to 'error', not
+        stamp 'idle' over the operator's investigatable badge — the
+        compaction neither retried nor resolved the failed turn (mirrors
+        the orphan reaper's ERROR carve-out).  The existing happy-path
+        test pins ['thinking', 'idle'] for an IDLE workstream."""
+        from turnstone.core.workstream import WorkstreamState
+
+        client, mgr = app_client
+        ws_id = self._create_ws(client)
+        ws = mgr.get(ws_id)
+        assert ws is not None
+        ws.state = WorkstreamState.ERROR  # a prior fatal turn's badge
+        resp = client.post(
+            "/v1/api/command",
+            json={"command": "/compact", "ws_id": ws_id},
+            headers=_auth("user-1"),
+        )
+        assert resp.json() == {"status": "ok"}
+        ws.worker_thread.join(timeout=5)
+        assert not ws.worker_thread.is_alive()
+        # THINKING during the window is honest (work is happening); the
+        # exit restores the badge instead of clearing it.
+        assert ws.ui.states == ["thinking", "error"]
+
+    def test_retracted_deferred_send_never_dispatches(self, app_client):
+        """DELETE /send with a deferred msg_id retracts the entry before
+        dispatch — the drain must skip it entirely.  The route falls
+        through the session's interjection dequeue (which misses) to the
+        workstream's pending list."""
+        client, mgr = app_client
+        ws_id = self._create_ws(client)
+        ws = mgr.get(ws_id)
+        assert ws is not None
+        gate = threading.Event()
+        ws.session.compact_gate = gate
+        client.post(
+            "/v1/api/command",
+            json={"command": "/compact", "ws_id": ws_id},
+            headers=_auth("user-1"),
+        )
+        r = client.post(
+            f"/v1/api/workstreams/{ws_id}/send",
+            json={"message": "changed my mind"},
+            headers=_auth("user-1"),
+        )
+        msg_id = r.json()["msg_id"]
+        d = client.request(
+            "DELETE",
+            f"/v1/api/workstreams/{ws_id}/send",
+            json={"msg_id": msg_id},
+            headers=_auth("user-1"),
+        )
+        assert d.json() == {"status": "removed"}
+        assert ws.session.dequeues == [msg_id]  # fall-through was exercised
+        gate.set()
+        assert self._wait_for(lambda: self._drain_idle(ws))
+        assert ws.session.sends == []
+        assert ws.session.queue_calls == []
+        # Unknown ids still answer not_found after checking both holders.
+        d2 = client.request(
+            "DELETE",
+            f"/v1/api/workstreams/{ws_id}/send",
+            json={"msg_id": "nope"},
+            headers=_auth("user-1"),
+        )
+        assert d2.json() == {"status": "not_found"}
+
+    def test_deferred_sends_dispatch_in_arrival_order(self, app_client):
+        """Two sends deferred behind one window dispatch in arrival order,
+        each as its own full-fidelity turn (the fake's sends complete
+        instantly, so the second never needs the interjection fallback)."""
+        client, mgr = app_client
+        ws_id = self._create_ws(client)
+        ws = mgr.get(ws_id)
+        assert ws is not None
+        gate = threading.Event()
+        ws.session.compact_gate = gate
+        client.post(
+            "/v1/api/command",
+            json={"command": "/compact", "ws_id": ws_id},
+            headers=_auth("user-1"),
+        )
+        first = client.post(
+            f"/v1/api/workstreams/{ws_id}/send",
+            json={"message": "first"},
+            headers=_auth("user-1"),
+        ).json()
+        second = client.post(
+            f"/v1/api/workstreams/{ws_id}/send",
+            json={"message": "second"},
+            headers=_auth("user-1"),
+        ).json()
+        assert first["status"] == second["status"] == "queued"
+        assert first["msg_id"] != second["msg_id"]
+        gate.set()
+        assert self._wait_for(lambda: len(ws.session.sends) == 2)
+        assert [s[0] for s in ws.session.sends] == ["first", "second"]
+        assert self._wait_for(lambda: self._drain_idle(ws))
+
+    def test_force_cancel_of_wedged_command_releases_drain(self, app_client):
+        """Force-cancelling a wedged command clears the slot flags — the
+        drain polls the same (_worker_running, worker_kind) pair, so a
+        deferred message dispatches WITHOUT waiting for the abandoned
+        thread to unwedge (the operator's escape hatch keeps working
+        under defer exactly as it did under park)."""
+        client, mgr = app_client
+        ws_id = self._create_ws(client)
+        ws = mgr.get(ws_id)
+        assert ws is not None
+        gate = threading.Event()
+        ws.session.compact_gate = gate
+        client.post(
+            "/v1/api/command",
+            json={"command": "/compact", "ws_id": ws_id},
+            headers=_auth("user-1"),
+        )
+        zombie = ws.worker_thread
+        r = client.post(
+            f"/v1/api/workstreams/{ws_id}/send",
+            json={"message": "deferred behind wedge"},
+            headers=_auth("user-1"),
+        )
+        assert r.json()["status"] == "queued"
+        resp = client.post(
+            f"/v1/api/workstreams/{ws_id}/cancel",
+            json={"force": True},
+            headers=_auth("user-1"),
+        )
+        assert resp.status_code == 200
+        # Dispatch happens while the zombie is STILL wedged on the gate.
+        assert self._wait_for(lambda: ws.session.sends)
+        assert [s[0] for s in ws.session.sends] == ["deferred behind wedge"]
+        gate.set()  # let the abandoned thread finish and be joined
+        zombie.join(timeout=5)
+        assert not zombie.is_alive()
+        assert self._wait_for(lambda: self._drain_idle(ws))
+
+    def test_ws_close_mid_window_drops_pending_and_drain_exits(self, app_client):
+        """A workstream closed with deferred sends outstanding drops them
+        (documented at-most-once contract) and the drain task retires
+        itself — no leaked task, no dispatch into a closed workstream."""
+        client, mgr = app_client
+        ws_id = self._create_ws(client)
+        ws = mgr.get(ws_id)
+        assert ws is not None
+        gate = threading.Event()
+        ws.session.compact_gate = gate
+        client.post(
+            "/v1/api/command",
+            json={"command": "/compact", "ws_id": ws_id},
+            headers=_auth("user-1"),
+        )
+        r = client.post(
+            f"/v1/api/workstreams/{ws_id}/send",
+            json={"message": "never delivered"},
+            headers=_auth("user-1"),
+        )
+        assert r.json()["status"] == "queued"
+        with ws._lock:
+            ws._closed = True  # the SessionManager.close tombstone shape
+        gate.set()
+        assert self._wait_for(lambda: self._drain_idle(ws))
+        assert ws.session.sends == []
+        assert ws.session.queue_calls == []
+
+    def test_deferred_send_dispatches_into_post_swap_session(self, app_client):
+        """The drain re-captures ws.session per attempt: a /resume-style
+        identity swap during the window routes the deferred message into
+        the POST-swap session — the same guarantee the park's
+        per-iteration re-capture provided."""
+        client, mgr = app_client
+        ws_id = self._create_ws(client)
+        ws = mgr.get(ws_id)
+        assert ws is not None
+        old_session = ws.session
+        gate = threading.Event()
+        old_session.compact_gate = gate
+        client.post(
+            "/v1/api/command",
+            json={"command": "/compact", "ws_id": ws_id},
+            headers=_auth("user-1"),
+        )
+        r = client.post(
+            f"/v1/api/workstreams/{ws_id}/send",
+            json={"message": "post-swap please"},
+            headers=_auth("user-1"),
+        )
+        assert r.json()["status"] == "queued"
+        new_session = _FakeSession(ws_id=ws_id, user_id="user-1")
+        ws.session = new_session  # the in-place identity swap
+        gate.set()
+        assert self._wait_for(lambda: new_session.sends)
+        assert [s[0] for s in new_session.sends] == ["post-swap please"]
+        assert old_session.sends == []
+        assert self._wait_for(lambda: self._drain_idle(ws))
+
+    def test_rejected_deferred_entry_waits_for_slot_then_fresh_spawns(self, app_client):
+        """The drain's rejection arm: an entry the interjection fallback
+        refuses (cross-user here; attachments/oversized take the same
+        path) is NOT dropped — it waits for the slot to free and then
+        dispatches as its own fresh turn.  The queued ack must never
+        become a silent drop while the workstream lives."""
+        from turnstone.core.session import CrossUserInterjectionError
+
+        client, mgr = app_client
+        ws_id = self._create_ws(client)
+        ws = mgr.get(ws_id)
+        assert ws is not None
+        gate = threading.Event()
+        send_gate = threading.Event()
+        ws.session.compact_gate = gate
+        ws.session.send_gate = send_gate
+        ws.session.queue_raises = CrossUserInterjectionError("another participant's turn")
+        client.post(
+            "/v1/api/command",
+            json={"command": "/compact", "ws_id": ws_id},
+            headers=_auth("user-1"),
+        )
+        for msg in ("first", "second"):
+            r = client.post(
+                f"/v1/api/workstreams/{ws_id}/send",
+                json={"message": msg},
+                headers=_auth("user-1"),
+            )
+            assert r.json()["status"] == "queued"
+        gate.set()
+        # Entry 1 spawns fresh and WEDGES in send() on send_gate — a live
+        # turn now holds the slot, so entry 2 takes the interjection
+        # fallback and is refused (the fake raises cross-user).
+        assert self._wait_for(lambda: ws.session.queue_calls == ["second"])
+        assert [s[0] for s in ws.session.sends] == ["first"]
+        send_gate.set()  # the turn ends; the slot frees
+        assert self._wait_for(lambda: len(ws.session.sends) == 2)
+        # Entry 2 dispatched as its own fresh turn — exactly one refused
+        # queue attempt, then the fresh-spawn arm.
+        assert [s[0] for s in ws.session.sends] == ["first", "second"]
+        assert ws.session.queue_calls == ["second"]
+        assert self._wait_for(lambda: self._drain_idle(ws))
+
     def test_exit_command_emits_ended_info_and_never_answers(self, app_client):
         """should_exit commands shut the session down — the worker emits
         the ended notice and never launches an answering turn (sends
-        during the window park in the /send route and belong to whatever
+        during the window defer in the /send route and belong to whatever
         follows the shutdown, not to this worker)."""
         client, mgr = app_client
         ws_id = self._create_ws(client)

turnstone/cli.py

@@ -17,6 +17,7 @@ import threading
 from typing import TYPE_CHECKING, Any
 
 from turnstone.core.adapters.interactive_adapter import InteractiveAdapter
+from turnstone.core.compaction_render import render_compaction_event_as_info
 from turnstone.core.judge import JudgeConfig
 from turnstone.core.session import ChatSession, SessionUI
 from turnstone.core.session_manager import SessionManager
@@ -333,61 +334,11 @@ class TerminalUI(SessionUI):
         lines — the notice, ``part k/N`` progress, and the token-delta +
         boxed-summary result the CLI printed before these became structured
         events (the web UI renders the same payloads as a progress card).
+        Shared with :meth:`ChatSession._compaction_event`'s duck-typed
+        fallback so both render identically; failed-end suppression rides
+        the emitter-stamped ``notice`` field (single policy site).
         """
-        phase = payload.get("phase")
-        if phase == "start":
-            # The threshold notice prints only when a threshold actually
-            # fired (pct present — _do_auto_compact).  The overflow-retry
-            # path compacts with auto=True but prints its own "[Context
-            # overflow — auto-compacting and retrying]" line; claiming a
-            # percentage there would fabricate the trigger.  Manual starts
-            # print nothing — the thinking spinner covers it.
-            pct = payload.get("pct")
-            if payload.get("trigger") == "auto" and pct is not None:
-                where = payload.get("where") or ""
-                qualifier = f" {where}" if where else ""
-                self.on_info(
-                    f"\n[Auto-compacting{qualifier}: prompt exceeds {pct}% of context window]"
-                )
-        elif phase == "progress":
-            if payload.get("warning") == "summary_truncated":
-                self.on_info("[Warning: compaction summary was truncated]")
-            elif payload.get("retry_in") is not None:
-                self.on_info(
-                    f"[Compact retrying in {payload['retry_in']:.0f}s: {payload.get('error', '')}]"
-                )
-            else:
-                self.on_info(f"[compacting part {payload.get('part')}/{payload.get('total')}…]")
-        elif phase == "end":
-            if payload.get("ok"):
-                before = payload.get("before_tokens", 0)
-                after = payload.get("after_tokens", 0)
-                self.on_info(f"[compacted: ~{before:,} -> ~{after:,} tokens]")
-                separator = "─" * 60
-                lines = [separator]
-                for line in str(payload.get("summary") or "").splitlines():
-                    lines.append(f"  {line}")
-                lines.append(separator)
-                self.on_info("\n".join(lines))
-            elif payload.get("reason") != "error":
-                # reason="error" already printed through on_error (red) —
-                # the end event is card-teardown for the web panes, not a
-                # second line here.  A cancelled AUTO compaction is also
-                # silent: the surrounding send prints "[Generation
-                # cancelled]" itself, and pre-lifecycle-events one Stop
-                # printed exactly one line.  (A cancelled MANUAL /compact
-                # keeps the message — it's the only line that path prints.)
-                # A SUPERSEDED end is a force-abandoned compaction retiring
-                # late — nobody is waiting on it; narrating it mid-turn
-                # reads as the LIVE work being cancelled.
-                # HAND-SYNCED SIBLING: conversation.js applyCompactionEvent
-                # implements this same three-clause display policy (skip
-                # error-reason / skip superseded / skip cancelled+auto) for
-                # the web panes — change both or they drift.
-                if not payload.get("superseded") and not (
-                    payload.get("reason") == "cancelled" and payload.get("trigger") == "auto"
-                ):
-                    self.on_info(str(payload.get("message") or ""))
+        render_compaction_event_as_info(payload, self.on_info)
         return None
 
     def on_state_change(self, state: str) -> None:

turnstone/console/coordinator_adapter.py

@@ -378,7 +378,7 @@ class CoordinatorAdapter:
                 # at this workstream from the interactive UI) holds the
                 # raced slot: the interjection queue is turn-shaped and must
                 # stay unreachable during command windows — same rule as the
-                # /send route's park.  Fail the dispatch (returns False, the
+                # /send route's defer.  Fail the dispatch (returns False, the
                 # caller's retryable-backpressure surface) rather than queue
                 # a message that would be capped and could cross a /resume
                 # identity swap.

turnstone/console/server.py

@@ -3887,7 +3887,7 @@ def _audit_coordinator_create(
     )
 
 
-def _coord_spawn_metrics(_request: Request, ui: Any) -> None:
+def _coord_spawn_metrics(_request: Request | None, ui: Any) -> None:
     """Per-spawn counter writes for coord — mirrors interactive's pattern.
 
     Wired onto :attr:`SessionEndpointConfig.spawn_metrics`. Increments

turnstone/console/static/coordinator/coordinator.js

@@ -31,7 +31,6 @@ import {
   buildCompactionCard,
   applyCompactionEvent,
   resetCompactionHolder,
-  sendAbortMs,
   buildSystemNudgeMarker,
   maxSeverityItem,
   buildConvBatchShell,
@@ -1997,10 +1996,11 @@ function createCoordinatorPane(root, wsId, opts) {
     let sendTimer = null;
     if (sendCtrl) {
       sendInit.signal = sendCtrl.signal;
-      // Same policy as the interactive composer: long bound while this
-      // pane's compaction card is live (the send parks server-side for
-      // the command window), wedged-node default otherwise.
-      sendTimer = setTimeout(() => sendCtrl.abort(), sendAbortMs(compactionHolder));
+      // Same policy as the interactive composer: flat wedged-node bound —
+      // every /send answers within RTT now (dispatched, queued, or
+      // deferred-with-msg_id during a command window; the server parks
+      // nothing against this POST).  Dismissal is bind() → DELETE.
+      sendTimer = setTimeout(() => sendCtrl.abort(), 15000);
     }
     let sendReq = authFetch(
       "/v1/api/workstreams/" + encodeURIComponent(wsId) + "/send",
@@ -2053,8 +2053,14 @@ function createCoordinatorPane(root, wsId, opts) {
           // flag so any subsequent send takes the queue path correctly,
           // and accept the small UX gap (no in-UI dismiss for THIS
           // message). The server still delivers it on worker drain.
-          if (queuedEl) queue.bind(queuedEl, data.msg_id);
-          else setBusy(true);
+          if (queuedEl) {
+            // Deferred sends (command-window defer) can carry attachments —
+            // stash the count BEFORE bind (a pre-bind ✕ confirms inside
+            // bind) so the dismiss path can surface the
+            // discarded-attachments consequence.
+            queuedEl._deferredAttachments = (data.attached_ids || []).length;
+            queue.bind(queuedEl, data.msg_id);
+          } else setBusy(true);
           attachments.consume(data.attached_ids, data.dropped_attachment_ids);
         } else if (data && data.status === "busy") {
           if (queuedEl) queue.remove(queuedEl);

turnstone/core/compaction_render.py

@@ -0,0 +1,69 @@
+"""Render compaction lifecycle events as classic info-channel text lines.
+
+One implementation, two consumers:
+
+* :class:`turnstone.cli.TerminalUI` — the terminal's native rendering of
+  ``on_compaction`` payloads (the threshold notice, ``part k/N``
+  progress, and the token-delta + boxed-summary result it printed before
+  these became structured events).
+* :meth:`ChatSession._compaction_event`'s duck-typed fallback — a
+  SessionUI implementation that predates the ``on_compaction`` hook gets
+  these same lines through its ``on_info`` instead of silence (an
+  auto-compaction that swaps history with zero announcement).
+
+Display policy: failed ends consult ``payload["notice"]`` — stamped by
+the emitter (:meth:`ChatSession._compaction_event`, the single policy
+site) — never re-derive suppression from reason/trigger/superseded here.
+Superseded OK ends still render: the history swap really committed, and
+suppressing its announcement is exactly the silent-swap failure this
+module exists to prevent.
+"""
+
+from collections.abc import Callable
+from typing import Any
+
+
+def render_compaction_event_as_info(
+    payload: dict[str, Any], on_info: Callable[[str], None]
+) -> None:
+    """Print one compaction lifecycle event through ``on_info``."""
+    phase = payload.get("phase")
+    if phase == "start":
+        # The threshold notice prints only when a threshold actually
+        # fired (pct present — _do_auto_compact).  The overflow-retry
+        # path compacts with auto=True but prints its own "[Context
+        # overflow — auto-compacting and retrying]" line; claiming a
+        # percentage there would fabricate the trigger.  Manual starts
+        # print nothing — the caller's own activity display covers it.
+        pct = payload.get("pct")
+        if payload.get("trigger") == "auto" and pct is not None:
+            where = payload.get("where") or ""
+            qualifier = f" {where}" if where else ""
+            on_info(f"\n[Auto-compacting{qualifier}: prompt exceeds {pct}% of context window]")
+    elif phase == "progress":
+        if payload.get("warning") == "summary_truncated":
+            on_info("[Warning: compaction summary was truncated]")
+        elif payload.get("retry_in") is not None:
+            on_info(f"[Compact retrying in {payload['retry_in']:.0f}s: {payload.get('error', '')}]")
+        else:
+            on_info(f"[compacting part {payload.get('part')}/{payload.get('total')}…]")
+    elif phase == "end":
+        if payload.get("ok"):
+            before = payload.get("before_tokens", 0)
+            after = payload.get("after_tokens", 0)
+            on_info(f"[compacted: ~{before:,} -> ~{after:,} tokens]")
+            separator = "─" * 60
+            lines = [separator]
+            for line in str(payload.get("summary") or "").splitlines():
+                lines.append(f"  {line}")
+            lines.append(separator)
+            on_info("\n".join(lines))
+        elif payload.get("notice"):
+            # The emitter stamps ``notice`` on failed ends (suppressing
+            # error-reason ends — already printed red through on_error —
+            # plus superseded and cancelled-auto ends).  A payload without
+            # the field (an event replayed from an older node) stays
+            # silent: these notices are informational, and re-deriving the
+            # suppression here is the cross-runtime drift trap the stamp
+            # exists to kill.
+            on_info(str(payload.get("message") or ""))

turnstone/core/session.py

@@ -7799,16 +7799,52 @@ class ChatSession:
         not drive the activity-pill restore or animate a successor's card.
         ``my_generation`` is 0 on direct test invocations — falsy, so such
         events are never marked superseded (matching ``_check_cancelled``).
+
+        Failed ends additionally carry ``notice`` — the single display-
+        policy site: renderers show the failure message only when it is
+        true, instead of each re-deriving suppression from reason/trigger/
+        superseded (the cross-runtime drift trap the old hand-synced
+        cli.py/conversation.js clauses documented).  Error-reason ends are
+        suppressed because :meth:`_compaction_bailed` already fired the one
+        red ``on_error`` row; cancelled-auto ends because the surrounding
+        send prints its own "[Generation cancelled]"; superseded ends
+        because nobody is waiting on a force-abandoned compaction and its
+        notice mid-turn reads as the LIVE work being cancelled.
         """
         stale = bool(my_generation and my_generation != self._generation)
+        event: dict[str, Any] = {"compaction_id": my_generation, "superseded": stale, **payload}
+        if payload.get("phase") == "end" and not payload.get("ok"):
+            event["notice"] = (
+                not stale
+                and payload.get("reason") != "error"
+                and not (payload.get("reason") == "cancelled" and payload.get("trigger") == "auto")
+            )
         # getattr-guarded like on_generation_claimed/on_aux_usage: a
-        # duck-typed SessionUI predating the hook gets no compaction events
-        # rather than an AttributeError that wedges every long session at
-        # its first auto-compaction.
+        # duck-typed SessionUI predating the hook must not hit an
+        # AttributeError that wedges every long session at its first
+        # auto-compaction.
         emit = getattr(self.ui, "on_compaction", None)
         if emit is None:
+            # Pre-hook UIs get the classic info lines back (an
+            # auto-compaction must never swap history with zero
+            # announcement — the pre-1.8 lines reached every UI
+            # unconditionally).  Invoked for superseded events too: a
+            # superseded OK end announces a swap that really committed,
+            # and failed-end staleness is already encoded in ``notice``.
+            # ``on_info`` is getattr-guarded like the hook itself — the
+            # never-crash property is the floor; a UI with neither hook
+            # keeps compacting silently.  Deliberately NOT dual-emitted
+            # for hook-aware UIs or SSE: pre-1.8 SSE/SDK clients that
+            # ignore unknown `compaction` events lose these lines — a
+            # documented 1.8 breaking change (CHANGELOG); dual emission
+            # would double-render on every current client.
+            info = getattr(self.ui, "on_info", None)
+            if info is not None:
+                from turnstone.core.compaction_render import render_compaction_event_as_info
+
+                render_compaction_event_as_info(event, info)
             return None
-        result = emit({"compaction_id": my_generation, "superseded": stale, **payload})
+        result = emit(event)
         # Duck-typed hooks aren't bound to the protocol's return type; the
         # marker-stamp consumer needs int-or-None, nothing else.
         return result if isinstance(result, int) else None
@@ -9043,9 +9079,10 @@ class ChatSession:
         queue from BEFORE their window (a dying send worker's closing race)
         — the /compact worker's exit seam is the caller.  Messages sent
         DURING a command window never reach this queue: the /send route
-        parks them while ``worker_kind == "command"`` and dispatches them
-        as ordinary sends afterwards.  Must only be called by the thread
-        that owns the worker slot — it mutates ``self.messages``.
+        defers them (``ws._pending_sends``) while ``worker_kind ==
+        "command"`` and the drain task dispatches them as ordinary sends
+        afterwards.  Must only be called by the thread that owns the
+        worker slot — it mutates ``self.messages``.
         """
         return self._flush_queued_messages()
 
@@ -17347,9 +17384,10 @@ class ChatSession:
 
             # Flush any stranded queued text BEFORE the identity swap so it
             # is persisted into the workstream it was ADDRESSED to.  Sends
-            # during the command window itself park in the /send route and
-            # never queue; this covers only a message stranded by a dying
-            # send worker's closing race before this command started.
+            # during the command window itself defer in the /send route
+            # (ws._pending_sends) and never queue; this covers only a
+            # message stranded by a dying send worker's closing race
+            # before this command started.
             self._flush_queued_messages()
             self.messages.clear()
             self._read_files.clear()

turnstone/core/session_routes.py

@@ -103,8 +103,10 @@ AttachmentOwnerResolver = Callable[
 ]
 # (request, ui) — kind's spawn-time bookkeeping. Interactive bumps
 # ``_metrics.record_message_sent`` + per-UI message counters; coord
-# has no analog and wires ``None``.
-SpawnMetricsHook = Callable[["Request", Any], None]
+# has no analog and wires ``None``.  The request is ``None`` when the
+# pending-send drain dispatches a deferred entry (no live request
+# exists by then) — both installed impls ignore the argument.
+SpawnMetricsHook = Callable[["Request | None", Any], None]
 
 
 class CancelForensics(Protocol):
@@ -1369,13 +1371,16 @@ def make_cancel_handler(
                 #
                 # Documented bet — force-cancelling a wedged QUICK command
                 # (worker_kind == "command", e.g. /resume stuck in storage
-                # I/O): clearing the flag releases any parked /send, whose
-                # fresh worker can then run while the abandoned command
-                # thread finishes its in-place mutation — quick commands
-                # have no generation checkpoints to retire them (compact_now
-                # does).  Same blast radius as force-abandoning a send
-                # worker mid-tool; accepted because force-cancel is the
-                # operator escape hatch for an already-wedged session, not a
+                # I/O): clearing the flag releases the pending-send
+                # drain's park (_drain_pending_sends polls the same
+                # (_worker_running, worker_kind) pair the parked /send
+                # used to), so a deferred message's fresh worker can then
+                # run while the abandoned command thread finishes its
+                # in-place mutation — quick commands have no generation
+                # checkpoints to retire them (compact_now does).  Same
+                # blast radius as force-abandoning a send worker
+                # mid-tool; accepted because force-cancel is the operator
+                # escape hatch for an already-wedged session, not a
                 # routine path.  Revisit if commands ever gain generation
                 # discipline.
                 with ws._lock:
@@ -3942,6 +3947,313 @@ def make_detail_handler(cfg: SessionEndpointConfig) -> Handler:
     return detail
 
 
+# ---------------------------------------------------------------------------
+# Deferred sends (command windows)
+# ---------------------------------------------------------------------------
+
+
+@dataclass
+class _PendingSend:
+    """One send deferred during a command window.
+
+    A /send that lands while a slash-command worker holds the slot
+    (``ws.worker_kind == "command"`` — a manual /compact can hold it for
+    minutes) is answered ``{"status": "queued", "msg_id": ...}``
+    immediately and registered here; :func:`_drain_pending_sends`
+    dispatches it full-fidelity when the window closes.  This replaces
+    the parked-POST design, which encoded "client disconnected" as
+    "message retracted" — true only for the web composers' ✕-abort;
+    every bounded caller (the coordinator client and the console proxy
+    at timeout=30, SDKs, curl) times out instead, and its message was
+    deliberately dropped for the whole window.
+
+    ``attempt`` is the prebuilt one-session-capture dispatch closure
+    (:func:`_make_dispatch_attempt` with ``defer_fidelity=True``), so
+    the drain stays endpoint-agnostic — everything kind-specific
+    (attachments, spawn metrics, UI hooks) was captured at defer time.
+    ``retracted`` is flipped under ``ws._lock`` by the DELETE dequeue
+    fall-through; the drain never dispatches a retracted entry.
+
+    Durability contract (documented in the API reference): node-local
+    and in-memory, the interjection queue's lifetime — entries die with
+    the workstream or the process, so "queued" is at-most-once intake,
+    not durable acceptance.
+    """
+
+    msg_id: str
+    priority: str
+    attempt: Callable[[ChatSession], tuple[bool, dict[str, Any]]]
+    retracted: bool = False
+
+
+def _emit_send_ui(ws: Workstream, ui: Any, hook_name: str, *args: Any) -> None:
+    """Best-effort UI hook dispatch for send-worker closures.
+
+    Each call is wrapped in try/except so a failure in one hook (e.g.
+    listener-queue full → on_error raises) doesn't suppress the others.
+    Mirrors the pre-P1.5 coord_adapter.send per-hook defense.
+    """
+    if ui is None:
+        return
+    method = getattr(ui, hook_name, None)
+    if method is None:
+        return
+    try:
+        method(*args)
+    except Exception:
+        log.debug(
+            "ws.send.ui_hook_failed ws=%s hook=%s",
+            ws.id[:8] if ws.id else "",
+            hook_name,
+            exc_info=True,
+        )
+
+
+def _make_dispatch_attempt(
+    ws: Workstream,
+    cfg: SessionEndpointConfig,
+    ui: Any,
+    *,
+    message: str,
+    resolved_atts: list[Any],
+    ordered_taken: list[str],
+    send_id: str,
+    acting_uid: str,
+    request: Request | None,
+    defer_fidelity: bool = False,
+) -> Callable[[ChatSession], tuple[bool, dict[str, Any]]]:
+    """Build one atomic queue-or-spawn attempt bound to ONE session capture.
+
+    The single dispatch implementation shared by the /send route's
+    immediate path and :func:`_drain_pending_sends` — session re-capture
+    across /resume//new identity swaps, the cross-user and attachment
+    queue guards, ``send_id`` threading (the queue path reuses it as
+    ``queue_msg_id`` so the client's DELETE targets one id either way),
+    and the spawn-path metrics all live here, once.  Callers re-capture
+    ``ws.session`` before every attempt and pass it in: closures bound
+    to a pre-swap capture would send the user's message into the wrong
+    workstream's transcript.
+
+    ``queue_outcome`` (second element of the return) is written only
+    when the dispatcher takes the live-worker reuse path; empty after a
+    fresh-spawn dispatch.
+
+    ``defer_fidelity=True`` marks a deferred entry's attempt: it was
+    answered "queued" under the full-fidelity defer contract, so the
+    interjection fallback — which truncates at ``queue_message``'s
+    2000-char cap and cannot carry attachments — is refused for
+    oversized or attachment-bearing entries (the drain waits for the
+    slot and retries into the fresh-spawn arm instead).  The refusal
+    happens inside the enqueue callback, under the same ``ws._lock``
+    acquisition as the queue-vs-spawn decision, so a turn claiming the
+    slot between the drain's poll and this dispatch can never route the
+    entry into truncation.
+
+    ``request`` is ``None`` for drain-side attempts; the spawn-metrics
+    hook tolerates it (both installed impls ignore the argument).
+    """
+    import threading
+
+    from turnstone.core import session_worker
+    from turnstone.core.session import (
+        AttachmentsNotQueueableError,
+        CrossUserInterjectionError,
+        GenerationCancelled,
+    )
+
+    def attempt(session: ChatSession) -> tuple[bool, dict[str, Any]]:
+        queue_outcome: dict[str, Any] = {}
+
+        def _enqueue() -> None:
+            # Runs under ``ws._lock`` (session_worker.send calls it inside
+            # the same acquisition that reads _worker_running), so this
+            # worker_kind read cannot race the spawn write.  A command
+            # window must NEVER reach queue_message — its cap and
+            # cross-user guard are turn semantics — so report it and let
+            # the route defer (or the drain re-park).
+            if ws.worker_kind == "command":
+                queue_outcome["rejected"] = "command_window"
+                return
+            if defer_fidelity and (resolved_atts or len(message) > 2000):
+                queue_outcome["rejected"] = "defer_full_fidelity"
+                return
+            try:
+                cleaned, priority, msg_id = session.queue_message(
+                    message,
+                    attachment_ids=list(ordered_taken),
+                    queue_msg_id=send_id or None,
+                    interjector_user_id=acting_uid,
+                )
+            except AttachmentsNotQueueableError:
+                queue_outcome["rejected"] = "attachments_busy"
+                return
+            except CrossUserInterjectionError:
+                # A different authenticated participant tried to interject
+                # into someone else's in-flight turn; folding it in would
+                # borrow the initiator's credentials and misattribute the
+                # message.  Reject so they resend as a fresh turn once the
+                # worker idles (the drain instead waits and re-attempts).
+                queue_outcome["rejected"] = "cross_user_interjection"
+                return
+            queue_outcome["cleaned"] = cleaned
+            queue_outcome["priority"] = priority
+            queue_outcome["msg_id"] = msg_id
+
+        def _run() -> None:
+            me = threading.current_thread()
+            try:
+                kwargs: dict[str, Any] = {}
+                if resolved_atts:
+                    kwargs["attachments"] = resolved_atts
+                if send_id:
+                    kwargs["send_id"] = send_id
+                # Fresh turn: rebind per-user MCP credentials to the
+                # authenticated sender. Bound here (not via a send()
+                # kwarg) so per-kind session stubs with explicit send
+                # signatures keep working; getattr-guarded for the same
+                # reason. The queue path above never rebinds.
+                bind = getattr(session, "bind_acting_user", None)
+                if acting_uid and callable(bind):
+                    bind(acting_uid)
+                session.send(message, **kwargs)
+            except GenerationCancelled:
+                # Safety net — send() normally handles this internally.
+                # If this thread was force-abandoned, ws.worker_thread
+                # was set to None — don't emit spurious events.
+                if ws.worker_thread is me:
+                    _emit_send_ui(ws, ui, "on_stream_end")
+                    _emit_send_ui(ws, ui, "on_state_change", "idle")
+            except Exception:
+                # Undrained staged uploads aren't locked (the buffer is a
+                # peek, not a reservation) — they expire on the buffer TTL
+                # — so the only cleanup owed here is the UI streaming
+                # hook: ``session.send()`` already fired ``on_error``
+                # (with sanitized text), persisted ``last_error``, and
+                # emitted ``state='error'`` via
+                # :meth:`ChatSession._record_fatal_error` before
+                # re-raising.
+                if ws.worker_thread is me:
+                    _emit_send_ui(ws, ui, "on_stream_end")
+
+        ok = session_worker.send(
+            ws,
+            enqueue=_enqueue,
+            run=_run,
+            thread_name=f"send-worker-{ws.id[:8]}",
+        )
+        if ok and not queue_outcome and cfg.spawn_metrics is not None:
+            # Fresh spawn — the kind's per-turn metrics fire exactly once,
+            # from the shared attempt so the drain's dispatches count too.
+            try:
+                cfg.spawn_metrics(request, ui)
+            except Exception:
+                log.debug(
+                    "ws.send.spawn_metrics_failed ws=%s",
+                    ws.id[:8] if ws.id else "",
+                    exc_info=True,
+                )
+        return ok, queue_outcome
+
+    return attempt
+
+
+def _drain_pending_sends(ws: Workstream) -> None:
+    """Dispatch a workstream's deferred sends once its command window closes.
+
+    Per-workstream single-flight (``ws._pending_drain``), started by the
+    /send route when it defers an entry and run on a small daemon thread
+    (the dispatch machinery is synchronous and thread-shaped like every
+    other worker here, and a thread's lifetime is independent of any
+    event loop's — the request loop owes this drain nothing once the
+    route has answered).  It owns the waiting the parked POST used to do
+    — but server-side, so a client timeout or abort can no longer become
+    message loss.  Entries dispatch in arrival order via their prebuilt
+    attempt closures, re-capturing ``ws.session`` per attempt (a /resume
+    or /new that swapped the session mid-window routes the message into
+    the post-swap session, exactly as the park did).
+
+    Terminal outcomes per entry: dispatched (fresh spawn — or, for a
+    queue-shaped entry, the interjection fallback into a live turn,
+    msg_id preserved so the client's DELETE still targets it),
+    retracted (dismissed before dispatch), or dropped because the
+    workstream closed.  There is deliberately no give-up bound: an entry
+    acknowledged "queued" is never silently dropped while the workstream
+    lives — rejections wait for the slot to free and retry into the
+    fresh-spawn arm.
+
+    Claim discipline: an entry is popped under ``ws._lock`` immediately
+    before its dispatch attempt and re-inserted on rejection, so the
+    DELETE fall-through (which marks only in-list entries) can never
+    "remove" a message whose dispatch already left the station — a
+    claimed entry answers ``not_found`` ("already sent"), which its
+    eventual dispatch makes true.
+    """
+    import time
+
+    try:
+        while True:
+            with ws._lock:
+                pending = ws._pending_sends
+                while pending and pending[0].retracted:
+                    pending.pop(0)
+                if ws._closed:
+                    if pending:
+                        log.warning(
+                            "ws.send.pending_dropped_on_close ws=%s count=%d",
+                            ws.id[:8],
+                            len(pending),
+                        )
+                        pending.clear()
+                    ws._pending_drain = None
+                    return
+                if not pending:
+                    ws._pending_drain = None
+                    return
+                entry = pending[0]
+            if ws._worker_running and ws.worker_kind == "command":
+                # The park, relocated server-side: the poll cadence
+                # matches the old request-handler loop's.
+                time.sleep(0.25)
+                continue
+            session_now = ws.session
+            if session_now is None:
+                # Mid-swap / partial-construction gap; the loop-top close
+                # check terminates this if it's a close in progress.
+                time.sleep(0.25)
+                continue
+            with ws._lock:
+                if not ws._pending_sends or ws._pending_sends[0] is not entry:
+                    continue  # list reshaped under us — re-evaluate
+                if entry.retracted:
+                    ws._pending_sends.pop(0)
+                    continue
+                ws._pending_sends.pop(0)  # claim
+            ok, outcome = entry.attempt(session_now)
+            if not ok or outcome.get("rejected") in (
+                "command_window",
+                "defer_full_fidelity",
+                "attachments_busy",
+                "cross_user_interjection",
+            ):
+                # Window re-claimed between the poll and the dispatch, a
+                # live turn holds the slot against a full-fidelity entry,
+                # the interjection queue is saturated
+                # (session_worker.send → False on queue.Full), or the ws
+                # closed (resolved at the loop top).  Unclaim and wait.
+                with ws._lock:
+                    ws._pending_sends.insert(0, entry)
+                time.sleep(0.25)
+                continue
+            # Dispatched: fresh spawn (empty outcome) or interjection
+            # fallback (msg_id preserved) — this entry is done.
+    except Exception:
+        # Never die holding the single-flight slot — a wedged drain would
+        # strand every future deferred send for this workstream.
+        log.exception("ws.send.pending_drain_failed ws=%s", ws.id[:8] if ws.id else "")
+        with ws._lock:
+            ws._pending_drain = None
+
+
 def make_send_handler(cfg: SessionEndpointConfig) -> Handler:
     """Lifted body for ``POST {prefix}/{ws_id}/send`` — message dispatch.
 
@@ -3977,8 +4289,13 @@ def make_send_handler(cfg: SessionEndpointConfig) -> Handler:
       requested attachments that landed (may be a strict subset on
       reservation race losses).
     - 200 ``{"status": "queued", "priority", "msg_id", "attached_ids",
-      "dropped_attachment_ids"}`` — reused live worker; queued for
-      injection at the next tool-result seam.
+      "dropped_attachment_ids"}`` — reused live worker (queued for
+      injection at the next tool-result seam), OR deferred during a
+      slash-command window (registered on ``ws._pending_sends`` and
+      dispatched full-fidelity by :func:`_drain_pending_sends` when the
+      window closes — see :class:`_PendingSend` for the durability
+      contract).  ``DELETE {prefix}/{ws_id}/send`` with the ``msg_id``
+      retracts either kind before dispatch.
     - 200 ``{"status": "queue_full", "attached_ids",
       "dropped_attachment_ids"}`` — live worker's queue at
       capacity; reservations released. Caller should retry. The
@@ -3991,12 +4308,7 @@ def make_send_handler(cfg: SessionEndpointConfig) -> Handler:
     import threading
     import uuid
 
-    from turnstone.core import session_worker
-    from turnstone.core.session import (
-        AttachmentsNotQueueableError,
-        CrossUserInterjectionError,
-        GenerationCancelled,
-    )
+    from turnstone.core.tool_advisory import parse_priority
     from turnstone.core.web_helpers import auth_user_id, read_json_or_400
 
     async def send(request: Request) -> Response:
@@ -4093,162 +4405,94 @@ def make_send_handler(cfg: SessionEndpointConfig) -> Handler:
         if ws.session is None:
             return JSONResponse({"error": "No session"}, status_code=500)
 
-        def _dispatch_once(session: ChatSession) -> tuple[bool, dict[str, Any]]:
-            """One atomic queue-or-spawn attempt bound to ONE session capture.
-
-            The park loop below re-captures ``ws.session`` before every
-            attempt — a /resume or /new that ran while we parked swaps the
-            session's workstream identity in place, and closures bound to
-            the pre-swap capture would send the user's message into the
-            wrong workstream's transcript.  Binding happens here, in one
-            place, per attempt.  ``queue_outcome`` is written only when the
-            dispatcher takes the live-worker reuse path; empty after a
-            fresh-spawn dispatch.
-            """
-            queue_outcome: dict[str, Any] = {}
-
-            def _enqueue() -> None:
-                # Runs under ``ws._lock`` (session_worker.send calls it
-                # inside the same acquisition that reads _worker_running),
-                # so this worker_kind read cannot race the spawn write.  A
-                # command window must NEVER reach queue_message — its cap
-                # and cross-user guard are turn semantics — so refuse and
-                # let the route loop back to the park.
-                if ws.worker_kind == "command":
-                    queue_outcome["rejected"] = "command_window"
-                    return
-                try:
-                    cleaned, priority, msg_id = session.queue_message(
-                        message,
-                        attachment_ids=list(ordered_taken),
-                        queue_msg_id=send_id or None,
-                        interjector_user_id=acting_uid,
-                    )
-                except AttachmentsNotQueueableError:
-                    queue_outcome["rejected"] = "attachments_busy"
-                    return
-                except CrossUserInterjectionError:
-                    # A different authenticated participant tried to interject into
-                    # someone else's in-flight turn; folding it in would borrow the
-                    # initiator's credentials and misattribute the message. Reject
-                    # so they resend as a fresh turn once the worker idles.
-                    queue_outcome["rejected"] = "cross_user_interjection"
-                    return
-                queue_outcome["cleaned"] = cleaned
-                queue_outcome["priority"] = priority
-                queue_outcome["msg_id"] = msg_id
-
-            def _run() -> None:
-                me = threading.current_thread()
-                try:
-                    kwargs: dict[str, Any] = {}
-                    if resolved_atts:
-                        kwargs["attachments"] = resolved_atts
-                    if send_id:
-                        kwargs["send_id"] = send_id
-                    # Fresh turn: rebind per-user MCP credentials to the
-                    # authenticated sender. Bound here (not via a send()
-                    # kwarg) so per-kind session stubs with explicit send
-                    # signatures keep working; getattr-guarded for the same
-                    # reason. The queue path above never rebinds.
-                    bind = getattr(session, "bind_acting_user", None)
-                    if acting_uid and callable(bind):
-                        bind(acting_uid)
-                    session.send(message, **kwargs)
-                except GenerationCancelled:
-                    # Safety net — send() normally handles this internally.
-                    # If this thread was force-abandoned, ws.worker_thread
-                    # was set to None — don't emit spurious events.
-                    if ws.worker_thread is me:
-                        _emit_ui("on_stream_end")
-                        _emit_ui("on_state_change", "idle")
-                except Exception:
-                    # Undrained staged uploads aren't locked (the buffer is a peek,
-                    # not a reservation) — they expire on the buffer TTL — so the
-                    # only cleanup owed here is the UI streaming hook.
-                    if ws.worker_thread is me:
-                        # ``session.send()`` already fired ``on_error``
-                        # (with sanitized text), persisted ``last_error``,
-                        # and emitted ``state='error'`` via
-                        # :meth:`ChatSession._record_fatal_error` before
-                        # re-raising.  The route handler only needs the
-                        # streaming-cleanup hook the worker contract owes
-                        # the UI listeners.
-                        _emit_ui("on_stream_end")
-
-            ok = session_worker.send(
-                ws,
-                enqueue=_enqueue,
-                run=_run,
-                thread_name=f"send-worker-{ws.id[:8]}",
-            )
-            return ok, queue_outcome
-
-        def _emit_ui(hook_name: str, *args: Any) -> None:
-            """Best-effort UI hook dispatch.
-
-            Each call is wrapped in try/except so a failure in one
-            hook (e.g. listener-queue full → on_error raises) doesn't
-            suppress the others. Mirrors the pre-P1.5
-            coord_adapter.send per-hook defense.  Defined after
-            ``_dispatch_once`` but resolved late (when a worker runs) —
-            every dispatch happens strictly below this line.
-            """
-            if ui is None:
-                return
-            method = getattr(ui, hook_name, None)
-            if method is None:
-                return
-            try:
-                method(*args)
-            except Exception:
-                log.debug(
-                    "ws.send.ui_hook_failed ws=%s hook=%s",
-                    ws.id[:8] if ws.id else "",
-                    hook_name,
-                    exc_info=True,
-                )
-
-        # Park-then-dispatch.  While a slash-command worker holds the slot
-        # (a manual /compact can hold it for MINUTES), a send must not take
+        # Defer-and-drain.  While a slash-command worker holds the slot (a
+        # manual /compact can hold it for MINUTES), a send must not take
         # the interjection-queue path — its 2000-char cap and cross-user
         # guard are mid-TURN semantics, and a queued message would cross a
-        # /resume//new identity swap into the wrong workstream.  Parking
-        # reproduces the pre-worker-dispatch observable behavior (the
-        # request waited out the then-blocked event loop, then ran as a
-        # normal full-fidelity send under the sender's own identity) with
-        # the loop free: SSE keeps streaming the compaction progress card
-        # while we wait.  The inner park is deliberately unbounded — the
-        # composer bounds its POST at ~15s and aborts, and we poll
-        # ``is_disconnected`` because starlette does NOT cancel a running
-        # handler on client disconnect: dispatching after the client gave
-        # up would surprise the user with a duplicate turn when they
-        # resend.  The outer bound only caps command-window FLAPPING
-        # (back-to-back commands re-claiming the slot between our park
-        # exit and dispatch).
-        ok = False
-        queue_outcome: dict[str, Any] = {}
-        for _ in range(20):
-            while ws._worker_running and ws.worker_kind == "command":
-                if await request.is_disconnected():
-                    # Client abandoned the send mid-park; the response is
-                    # discarded — the status is for the access log only.
-                    return JSONResponse({"status": "abandoned"})
-                await asyncio.sleep(0.25)
-            session_now = ws.session
-            if session_now is None:
-                return JSONResponse({"error": "No session"}, status_code=500)
-            ok, queue_outcome = _dispatch_once(session_now)
-            if queue_outcome.get("rejected") != "command_window":
-                break
-        else:
-            # 20 consecutive command-window collisions — surface the same
-            # retryable backpressure shape as a saturated queue.
+        # /resume//new identity swap into the wrong workstream.  Instead
+        # of parking THIS request until the window closes (which encoded
+        # "client disconnected" as "message retracted" — deterministic
+        # message loss for every bounded caller: the coordinator client
+        # and console proxy time out at 30s, and the web composers' long
+        # abort bound raced the compaction card), the send is answered
+        # "queued" immediately and registered on ``ws._pending_sends``;
+        # the per-workstream drain task dispatches it full-fidelity when
+        # the window closes.  Dismissal is the same DELETE /send
+        # {msg_id} the interjection queue uses — server-confirmed, no
+        # POST-abort side channel.
+        attempt = _make_dispatch_attempt(
+            ws,
+            cfg,
+            ui,
+            message=message,
+            resolved_atts=resolved_atts,
+            ordered_taken=ordered_taken,
+            send_id=send_id,
+            acting_uid=acting_uid,
+            request=request,
+        )
+        session_now = ws.session
+        if session_now is None:
+            return JSONResponse({"error": "No session"}, status_code=500)
+        ok, queue_outcome = attempt(session_now)
+        if queue_outcome.get("rejected") == "command_window":
+            # ``send_id`` is minted only when attachments are enabled —
+            # the deferred entry needs a truthy id regardless: it is the
+            # client's dismiss/bind handle and the drain's
+            # queue_msg_id/send_id thread.
+            pending_msg_id = send_id or uuid.uuid4().hex
+            cleaned_display, pending_priority = parse_priority(message)
+            entry = _PendingSend(
+                msg_id=pending_msg_id,
+                priority=pending_priority,
+                attempt=_make_dispatch_attempt(
+                    ws,
+                    cfg,
+                    ui,
+                    message=message,
+                    resolved_atts=resolved_atts,
+                    ordered_taken=ordered_taken,
+                    send_id=pending_msg_id,
+                    acting_uid=acting_uid,
+                    request=None,
+                    defer_fidelity=True,
+                ),
+            )
+            with ws._lock:
+                if ws._closed:
+                    # Mirror the dispatch-refusal 404 below: a "queued"
+                    # answer for a workstream whose next resolution 404s
+                    # would promise a dispatch that can never happen.
+                    return JSONResponse({"error": cfg.not_found_label}, status_code=404)
+                ws._pending_sends.append(entry)
+                drain = ws._pending_drain
+                if drain is None or not drain.is_alive():
+                    t = threading.Thread(
+                        target=_drain_pending_sends,
+                        args=(ws,),
+                        name=f"pending-drain-{ws.id[:8]}",
+                        daemon=True,
+                    )
+                    ws._pending_drain = t
+                    t.start()
+            if cfg.emit_message_queued and hasattr(ui, "_enqueue"):
+                ui._enqueue(
+                    {
+                        "type": "message_queued",
+                        "message": cleaned_display,
+                        "priority": pending_priority,
+                        "msg_id": pending_msg_id,
+                    }
+                )
             return JSONResponse(
                 {
-                    "status": "queue_full",
-                    "attached_ids": [],
-                    "dropped_attachment_ids": list(requested_ids),
+                    "status": "queued",
+                    "priority": pending_priority,
+                    "msg_id": pending_msg_id,
+                    "attached_ids": list(ordered_taken),
+                    "dropped_attachment_ids": [
+                        aid for aid in requested_ids if aid not in taken_set
+                    ],
                 }
             )
         if not ok:
@@ -4324,16 +4568,9 @@ def make_send_handler(cfg: SessionEndpointConfig) -> Handler:
                 }
             )
 
-        # Spawned a fresh worker — kind's metrics fire once per turn.
-        if cfg.spawn_metrics is not None:
-            try:
-                cfg.spawn_metrics(request, ui)
-            except Exception:
-                log.debug(
-                    "ws.send.spawn_metrics_failed ws=%s",
-                    ws_id[:8] if ws_id else "",
-                    exc_info=True,
-                )
+        # Spawned a fresh worker — the kind's per-turn metrics fired inside
+        # the shared attempt (see _make_dispatch_attempt), where the drain
+        # task's dispatches fire them too.
         return JSONResponse(
             {
                 "status": "ok",
@@ -4633,12 +4870,17 @@ def make_attachment_handlers(cfg: SessionEndpointConfig) -> AttachmentHandlers:
 def make_dequeue_handler(cfg: SessionEndpointConfig) -> Handler:
     """Lifted body for ``DELETE {prefix}/{ws_id}/send`` — cancel a queued message.
 
-    Removes a previously-queued message identified by ``msg_id`` from
-    the workstream's pending queue. Returns ``status: removed`` when
-    the queue had the entry and ``status: not_found`` otherwise.
-    Queued messages don't carry attachments (see
-    :class:`AttachmentsNotQueueableError`), so there's no reservation
-    side-effect to undo here.
+    Removes a previously-queued message identified by ``msg_id`` —
+    first from the session's interjection queue, then (fall-through)
+    from the workstream's deferred-send list (``ws._pending_sends``,
+    sends answered "queued" during a command window).  Returns
+    ``status: removed`` when either held the entry and
+    ``status: not_found`` otherwise.  Interjection-queued messages
+    don't carry attachments (see :class:`AttachmentsNotQueueableError`)
+    and a deferred entry's resolved attachments die with it (the staged
+    bytes were peeked, not reserved — they expire on the buffer TTL),
+    so there's no reservation side-effect to undo here; the client
+    surfaces the discarded-attachments consequence to the user.
     """
     from turnstone.core.web_helpers import read_json_or_400
 
@@ -4677,6 +4919,21 @@ def make_dequeue_handler(cfg: SessionEndpointConfig) -> Handler:
         if ws.session is None:
             return JSONResponse({"error": "No session"}, status_code=400)
         removed = ws.session.dequeue_message(msg_id)
+        if not removed:
+            # Fall through to the deferred-send list: a send answered
+            # "queued" during a command window lives on the workstream
+            # (see _PendingSend), not in the session's interjection
+            # queue.  Marked under the same lock the drain claims under,
+            # so a retracted entry can never dispatch; an entry the
+            # drain already claimed is gone from the list and correctly
+            # answers not_found ("already sent" — its dispatch is in
+            # flight).
+            with ws._lock:
+                for entry in ws._pending_sends:
+                    if entry.msg_id == msg_id and not entry.retracted:
+                        entry.retracted = True
+                        removed = True
+                        break
         return JSONResponse({"status": "removed" if removed else "not_found"})
 
     return dequeue

turnstone/core/session_worker.py

@@ -113,24 +113,29 @@ def send(
     ``ws.worker_kind`` in the same lock acquisition as the
     ``(worker_thread, _worker_running)`` pair.  This is the ONLY site
     that sets ``_worker_running=True``, so the classification cannot be
-    bypassed by a new dispatch caller.  The /send route parks while a
+    bypassed by a new dispatch caller.  The /send route DEFERS while a
     command holds the slot instead of taking the interjection-queue
-    path (whose length cap / cross-user guard are turn semantics); an
-    ``enqueue`` callback that can fire during a command window (the
-    coordinator adapter's, the init race's) must refuse rather than
-    queue — see the command-window refusals at those closures.
+    path (whose length cap / cross-user guard are turn semantics): its
+    enqueue closure reports the window and the route registers the send
+    on ``ws._pending_sends`` for the drain task to dispatch when the
+    window closes.  An ``enqueue`` callback that can fire during a
+    command window (the coordinator adapter's, the init race's) must
+    refuse rather than queue — see the command-window refusals at those
+    closures.
 
-    The refusal is DELIBERATELY not centralized here despite the three
+    The refusal is DELIBERATELY not centralized here despite the
     hand-written guards: each surface needs a different refusal channel
-    (the /send route signals "re-park" via its ``queue_outcome`` flag;
-    the coordinator adapter and the init race raise ``queue.Full`` into
-    their existing backpressure statuses), and a central refusal inside
-    this function can only return ``False`` — indistinguishable from
-    queue-full/closed for the route's re-park decision and mislabeled by
-    the init path's status derivation.  Making it distinguishable means
-    a tri-state contract change across every dispatch caller, which is
-    more surface than three four-line guards.  If you add a NEW enqueue
-    closure that can queue turn work, copy the guard.
+    (the /send route's closure signals "command window" via its
+    ``queue_outcome`` flag — the defer trigger; the coordinator adapter
+    and the init race raise ``queue.Full`` into their existing
+    backpressure statuses), and a central refusal inside this function
+    can only return ``False`` — indistinguishable from queue-full/closed
+    exactly where the route must distinguish "defer this" from "drop
+    this".  Making it distinguishable means a tri-state contract change
+    across every dispatch caller — more surface than the guards it
+    replaces.  (A check inside this function's locked section would be
+    race-free; the cost is the contract change, not atomicity.)  If you
+    add a NEW enqueue closure that can queue turn work, copy the guard.
 
     Returns:
         ``True`` on successful enqueue (existing worker accepted) or

turnstone/core/workstream.py

@@ -13,7 +13,7 @@ import threading
 import time
 import uuid
 from dataclasses import dataclass, field
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Any
 
 if TYPE_CHECKING:
     from turnstone.core.session import ChatSession, SessionUI
@@ -146,11 +146,24 @@ class Workstream:
     # sets ``worker_thread``/``_worker_running``, so readers gating on
     # the running flag see a coherent triple.  A stale value after the
     # worker exits is harmless — every reader conjoins
-    # ``_worker_running``.  The /send route parks (never queues) while
+    # ``_worker_running``.  The /send route defers (never queues) while
     # this reads "command": the mid-turn interjection queue is
     # turn-shaped (length cap, cross-user guard) and must be
-    # unreachable during command windows.
+    # unreachable during command windows — deferred entries live on
+    # ``_pending_sends`` below.
     worker_kind: str = field(default="", repr=False)
+    # Sends deferred during a command window (full-fidelity pending
+    # entries — see ``session_routes._PendingSend``), dispatched in
+    # arrival order by the per-workstream drain task when the window
+    # closes.  Appends, retract-marks and claims all happen under
+    # ``_lock``.  Node-local and in-memory: entries die with the
+    # workstream or the process (the interjection queue's lifetime) —
+    # the /send contract documents the at-most-once consequence.
+    _pending_sends: list[Any] = field(default_factory=list, repr=False)
+    # Single-flight guard for the drain (a daemon ``threading.Thread``
+    # while one is live).  Written under ``_lock``; the drain clears it
+    # before exiting so a later deferred send starts a fresh one.
+    _pending_drain: Any = field(default=None, repr=False)
     # True once ``SessionManager.commit_create`` (or the non-deferred
     # path through ``SessionManager.create``) has fired the lifecycle
     # ``emit_created`` event for this workstream. Used by

turnstone/sdk/events.py

@@ -314,13 +314,17 @@ class CompactionEvent(ServerEvent):
     events additionally carry ``superseded``: True marks a force-abandoned
     compaction retiring after a successor generation took over — clients
     should skip failure notices for those (an OK end's result still
-    stands; the history swap happened).
+    stands; the history swap happened).  Failed ends carry ``notice``:
+    the emitter-computed display verdict — show ``message`` only when it
+    is True, instead of re-deriving suppression from
+    reason/trigger/superseded client-side.
     """
 
     type: str = "compaction"
     phase: str = ""
     compaction_id: int = 0
     superseded: bool = False
+    notice: bool = False
     trigger: str = ""
     where: str = ""
     pct: int | None = None

turnstone/server.py

@@ -1737,11 +1737,13 @@ async def command(request: Request) -> JSONResponse:
             One envelope for both command branches: the unknown-workstream
             404 and the busy refusal differ only in the retry hint.  The
             worker slot is claimed with ``worker_kind="command"`` — the
-            /send route PARKS (never queues) while that kind holds the
+            /send route DEFERS (never queues) while that kind holds the
             slot, so the mid-turn interjection queue and its turn-shaped
             semantics (length cap, cross-user guard) are unreachable for
-            the whole command window; messages sent mid-command dispatch
-            as ordinary full-fidelity sends when the window closes.
+            the whole command window; messages sent mid-command are
+            answered ``queued`` immediately and dispatched as ordinary
+            full-fidelity sends by the pending-send drain when the window
+            closes.
             """
             dispatched = session_worker.send(
                 ws,
@@ -1790,6 +1792,15 @@ async def command(request: Request) -> JSONResponse:
 
             def _run_compact() -> None:
                 me = threading.current_thread()
+                # Snapshot for the exit restore: with the slot free to
+                # claim, only IDLE or ERROR are reachable here (the live
+                # states imply a held slot and _dispatch_command refuses
+                # busy).  ERROR is the user-investigatable badge (same
+                # carve-out the orphan reaper honors) and /compact neither
+                # retries nor resolves the failed turn — the old inline
+                # /compact never touched ws.state — so the badge must
+                # survive the window; everything else exits to idle.
+                prev_state = ws.state
                 try:
                     cmd_ui.on_state_change("thinking")
                     session.compact_now()
@@ -1805,7 +1816,7 @@ async def command(request: Request) -> JSONResponse:
                     if ws.worker_thread is me:
                         try:
                             # Backstop only: sends during the command window
-                            # PARK in the /send route (they cannot reach the
+                            # DEFER in the /send route (they cannot reach the
                             # interjection queue — see _dispatch_command), so
                             # anything found here predates the window (a
                             # message stranded by a dying send worker's
@@ -1815,7 +1826,9 @@ async def command(request: Request) -> JSONResponse:
                                 log.warning("ws.compact.stranded_queue_flushed ws=%s", ws.id[:8])
                         except Exception:
                             log.exception("ws.compact.exit_seam_failed ws=%s", ws.id[:8])
-                        cmd_ui.on_state_change("idle")
+                        cmd_ui.on_state_change(
+                            "error" if prev_state is WorkstreamState.ERROR else "idle"
+                        )
 
             refusal = _dispatch_command(
                 _run_compact, f"compact-worker-{ws.id[:8]}", "run /compact again"
@@ -1875,11 +1888,12 @@ async def command(request: Request) -> JSONResponse:
                 # Unblock the endpoint response.  Suppress the loop-closed
                 # RuntimeError (process shutdown mid-command): nobody is
                 # waiting anymore.  No queue drain here: sends during the
-                # command window PARK in the /send route (the interjection
+                # command window DEFER in the /send route (the interjection
                 # queue is unreachable while worker_kind == "command"), so
-                # they dispatch as ordinary sends when this worker exits —
-                # the stranded-message backstop lives on the /compact seam,
-                # the only command window long enough to matter.
+                # the pending-send drain dispatches them as ordinary sends
+                # when this worker exits — the stranded-message backstop
+                # lives on the /compact seam, the only command window long
+                # enough to matter.
                 with contextlib.suppress(RuntimeError):
                     loop.call_soon_threadsafe(done.set)
 
@@ -2458,15 +2472,23 @@ async def _interactive_create_post_install(
                     ws.ui.on_stream_end()
                     ws.ui.on_state_change("idle")
             finally:
-                # Owner only — an abandoned init's late notify would
-                # duplicate the successor's.  (Guard, not early-return: a
-                # return in a finally silences in-flight exceptions.)
-                if ws.worker_thread is me:
-                    try:
-                        last_content = _extract_last_assistant_content(session)
-                        _fire_notify_targets(ws, last_content)
-                    except Exception:
-                        log.warning("notify_completion.hook_error", ws_id=ws.id, exc_info=True)
+                # Deliberately NOT owner-gated, unlike the except arm above
+                # and the _run_cmd/_run_compact follow-ups: those mutate
+                # live slot/UI state a successor now owns, while the notify
+                # is workstream-scoped — an outward signal that this
+                # workstream's initial turn ran to completion and its
+                # answer is in the transcript.  _fire_notify_targets has
+                # exactly ONE call site (here); successor turns never
+                # notify, so there is no successor duplicate for an owner
+                # gate to prevent — gating it turned force-cancel into
+                # permanent notification loss for scheduled/unattended
+                # workstreams (the empty-content fallback covers the
+                # error/cancel exits, as it always did).
+                try:
+                    last_content = _extract_last_assistant_content(session)
+                    _fire_notify_targets(ws, last_content)
+                except Exception:
+                    log.warning("notify_completion.hook_error", ws_id=ws.id, exc_info=True)
 
         init_enqueued = False
 
@@ -2492,7 +2514,7 @@ async def _interactive_create_post_install(
                 # raced slot: the interjection queue is turn-shaped (length
                 # cap, and the text would cross a /resume identity swap) and
                 # must stay unreachable during command windows — same rule
-                # as the /send route's park.  queue.Full is the existing
+                # as the /send route's defer.  queue.Full is the existing
                 # backpressure surface: the create reports the first message
                 # as undelivered (``queue_full``) and the client retries.
                 raise queue.Full()
@@ -4497,7 +4519,7 @@ def create_app(
         magic-mocked ``ws.user_id``."""
         return _require_ws_access(request, ws_id)
 
-    def _interactive_spawn_metrics(_request: Request, ui: Any) -> None:
+    def _interactive_spawn_metrics(_request: Request | None, ui: Any) -> None:
         """Per-conversation metrics fired once per send that spawns a
         fresh worker. Coord wires its own
         :func:`turnstone.console.server._coord_spawn_metrics` (the

turnstone/shared_static/composer_queue.js

@@ -265,6 +265,16 @@ export function createQueueController(opts) {
         var status = data && data.status;
         if (status === "removed") {
           el.remove();
+          // A deferred send (command-window defer) can carry attachments;
+          // retracting it discards them — the chips were consumed when the
+          // send was accepted and a retract does not re-stage the bytes.
+          // Say so instead of letting the files silently expire.
+          if (el._deferredAttachments > 0 && onNotice)
+            onNotice(
+              "Message removed. Its " +
+                el._deferredAttachments +
+                " attachment(s) were discarded — re-attach them to send again.",
+            );
           // `removed` is the only verdict that mutated server-side queue
           // state, so it's the only one worth re-syncing composer state for.
           if (onAfterDequeue) onAfterDequeue();
@@ -297,11 +307,6 @@ export function createQueueController(opts) {
     if (el.getAttribute("aria-busy") === "true") return;
     el.dataset.dismissAttempted = "1";
     _setDismissing(el, true);
-    // A send whose POST is still in flight (parked server-side during a
-    // command window — possibly for minutes) attaches an abort hook: the ×
-    // must kill the request itself, or the dismissed message dispatches
-    // anyway when the window closes.  Post-response the hook is a no-op.
-    if (typeof el._sendAbort === "function") el._sendAbort();
     var msgId = el.dataset.msgId;
     if (!msgId) {
       // Pre-bind: bind() confirms once the server returns the msg_id (it

turnstone/shared_static/conversation.js

@@ -246,11 +246,7 @@ export function applyCompactionEvent(holder, evt, hooks) {
     return;
   }
   if (evt.phase === "end") {
-    if (owns && holder.card) {
-      holder.card.remove();
-      holder.card = null;
-      holder.cid = null;
-    }
+    if (owns) resetCompactionHolder(holder);
     if (evt.ok) {
       // The persisted marker row is stamped with THIS event's id, so
       // whichever of /history repaint or live/replayed event renders
@@ -269,48 +265,27 @@ export function applyCompactionEvent(holder, evt, hooks) {
         ),
       );
       if (eid) hooks.renderedIds.add(eid);
-    } else if (
-      owns &&
-      !evt.superseded &&
-      evt.reason !== "error" &&
-      !(evt.reason === "cancelled" && evt.trigger === "auto")
-    ) {
+    } else if (owns && evt.notice) {
       // cancelled / not_enough_messages / irreducible / empty_summary —
-      // informational, not an error state.  Three suppressions: a stale
-      // end's notice would narrate a dead compaction underneath a live
-      // one; a superseded end (a force-abandoned compaction retiring
-      // late, flagged by the backend) is one nobody is waiting on — its
-      // notice mid-turn reads as the LIVE work being cancelled; and an
-      // auto-compaction's cancel is just part of cancelling the
-      // surrounding turn — the send loop emits its own "[Generation
-      // cancelled]" info line, so a second line here stacked two notices
-      // for one Stop click.  (A superseded OK end above still renders
-      // its result card — the history swap really happened.)
-      // HAND-SYNCED SIBLING: cli.py TerminalUI.on_compaction implements
-      // this same three-clause policy for the terminal — change both or
-      // they drift (two runtimes, no shared code path).
+      // informational, not an error state.  Whether the message is shown
+      // is the emitter's call: the backend stamps `notice` on failed ends
+      // (suppressing error-reason / superseded / cancelled-auto ends —
+      // see ChatSession._compaction_event, the single policy site), so
+      // this arm stays mechanical.  `owns` is the one pane-local clause
+      // the emitter cannot compute — card ownership via compaction_id vs
+      // holder.cid — and it guards a reachable divergence: a /resume
+      // swaps sessions and restarts generation counters, so an abandoned
+      // old-session end can arrive superseded=false against the new
+      // session's live card.  An end without the field (replayed from an
+      // older node) stays silent — these notices are informational.
+      // (A superseded OK end above still renders its result card — the
+      // history swap really happened.)
       hooks.onNotice(evt.message || "Compaction skipped.");
     }
     hooks.scroll(true);
   }
 }
 
-// Send-POST abort bound for a pane, selected off its compaction holder.
-// Sends during a slash-command window PARK server-side and dispatch when
-// the window closes — a manual /compact legitimately runs for minutes, so
-// while ITS progress card is live the composer must not abort the parked
-// POST at the wedged-node default (~15s) and silently drop the message.
-// The card is the one cross-tab signal that a long window is in progress
-// (SSE-driven via applyCompactionEvent); with no card the short default
-// stands — a WEDGED quick command past 15s should fail loudly, not hang
-// the composer for 10 minutes.  Deployment note: reverse proxies bound
-// the effective park at their own read timeout (documented in the API
-// reference).  Shared by the interactive composer and the coordinator
-// viewer so the policy can't drift between panes.
-export function sendAbortMs(holder) {
-  return holder && holder.card ? 600000 : 15000;
-}
-
 // Retire a pane's in-progress compaction card (if any) and clear the
 // holder.  The teardown half of the lifecycle the reducer above owns —
 // exported so the panes' stream_end handlers (force-stop abandons the

turnstone/shared_static/interactive.js

@@ -26,7 +26,6 @@ import {
   buildCompactionCard,
   applyCompactionEvent,
   resetCompactionHolder,
-  sendAbortMs,
   buildSystemNudgeMarker,
   buildConvBatchShell,
   buildConvRow,
@@ -269,6 +268,12 @@ class Pane {
     // (conversation.applyCompactionEvent); `card` is the in-progress card
     // between start and end, nulled wherever the transcript DOM is wiped.
     this._compaction = { card: null, cid: null };
+    // Rendered-event-id dedup for system turns and compaction markers
+    // (/history repaint vs live/replayed event — whichever renders first
+    // wins).  replayHistory assigns a FRESH Set on every transcript wipe
+    // so pre-wipe ids can't suppress re-painted rows; this is the
+    // first-load init.
+    this._renderedSystemEventIds = new Set();
     this._retryHolderEl = null;
     this._toolRowIndex = new Map();
     this._streamElIndex = new Map();
@@ -560,9 +565,6 @@ class Pane {
     // repaint or live/replayed event renders first wins.  reason="error"
     // ends render through the paired `error` event (red row), not here.
     this.removeEmptyState();
-    if (!this._renderedSystemEventIds) {
-      this._renderedSystemEventIds = new Set();
-    }
     applyCompactionEvent(this._compaction, evt, {
       container: this.messagesEl,
       renderedIds: this._renderedSystemEventIds,
@@ -1918,11 +1920,7 @@ class Pane {
         // the resume-cursor fix this shouldn't recur, but the guard keeps the
         // /history+replay seam idempotent for system turns regardless.
         const sysEid = evt._event_id != null ? String(evt._event_id) : null;
-        if (
-          sysEid &&
-          this._renderedSystemEventIds &&
-          this._renderedSystemEventIds.has(sysEid)
-        ) {
+        if (sysEid && this._renderedSystemEventIds.has(sysEid)) {
           break;
         }
         this.addSystemContext(
@@ -1930,11 +1928,7 @@ class Pane {
           evt.source || "",
           evt.meta || null,
         );
-        if (sysEid) {
-          if (!this._renderedSystemEventIds)
-            this._renderedSystemEventIds = new Set();
-          this._renderedSystemEventIds.add(sysEid);
-        }
+        if (sysEid) this._renderedSystemEventIds.add(sysEid);
         break;
       }
 
@@ -3769,17 +3763,12 @@ class Pane {
     let sendTimer = null;
     if (sendCtrl) {
       sendInit.signal = sendCtrl.signal;
-      // Long bound while a compaction card is live (the send is parked
-      // server-side for the command window); wedged-node default
-      // otherwise — see sendAbortMs.
-      sendTimer = setTimeout(
-        () => sendCtrl.abort(),
-        sendAbortMs(this._compaction),
-      );
-      // An × on the queued bubble BEFORE the response arrives (pre-bind)
-      // must also kill the parked POST — otherwise the dismissed message
-      // dispatches anyway when the command window closes, minutes later.
-      if (queuedEl) queuedEl._sendAbort = () => sendCtrl.abort();
+      // Flat wedged-node bound: every /send answers within RTT now —
+      // dispatched, queued, or deferred-with-msg_id during a command
+      // window (the server parks nothing against this POST), so a
+      // response slower than this means a wedged node, not a long
+      // command.  Dismissal is bind() → DELETE, never a POST abort.
+      sendTimer = setTimeout(() => sendCtrl.abort(), 15000);
     }
     let sendReq = authFetch(
       this._base +
@@ -3836,8 +3825,15 @@ class Pane {
           // optimistic user bubble is already in the log and the server
           // still delivers the message on worker drain — accept the
           // small UX gap (no in-UI dismiss for THIS message).
-          if (queuedEl) this.queue.bind(queuedEl, data.msg_id);
-          else this.setBusy(true);
+          if (queuedEl) {
+            // A deferred send (command-window defer) can carry
+            // attachments — stash the count BEFORE bind (a pre-bind ✕
+            // confirms inside bind) so the dismiss path can tell the
+            // user the attachments were discarded: the chips are
+            // consumed below and a retract does not re-stage them.
+            queuedEl._deferredAttachments = (data.attached_ids || []).length;
+            this.queue.bind(queuedEl, data.msg_id);
+          } else this.setBusy(true);
           this.attachments.consume(
             data.attached_ids,
             data.dropped_attachment_ids,