chore(compaction): review round 10 — cleanups from the first correctness-clean round

- INTERJECTION_CAP_CHARS joins PENDING_SENDS_MAX in workstream.py: the
  2000-char interjection cap was triplicated (queue_message's truncation,
  the defer-fidelity refusal, the test fake) and already drifting in
  measurement — the defer check deliberately measures RAW text (raw >=
  cleaned since parse_priority only strips, so it can only over-refuse
  into a full-fidelity fresh spawn, never admit a truncation), now
  stated in a comment. The four unrelated 2000s (notify tool, recall
  preview, summary formatting, agent step cap) stay deliberately
  unlinked — they are different contracts.
- The changelog's ~110-line compaction bullet is split into six per-seam
  bullets matching house style, and the Breaking (1.8) compaction-event
  notice moved under "### Changed" where integrators scanning bullet
  heads will actually see it (cross-referenced both ways with the
  pre-1.8 embedder compat bullet).
- SpawnMetricsHook takes (ui) only: the request parameter was threaded
  through the whole dispatch-attempt path solely to be ignored by both
  installed impls; the stale "coord wires None" claims in the rewritten
  comment blocks are corrected too.
- The attachments tests' Mock-hardening block lives once in
  _harden_ws_mock() — deliberately excluding _worker_running, which each
  fixture chooses per scenario (one relies on the truthy auto-Mock).
- Two hand-rolled poll loops become wait_until (file convention,
  diagnostic timeout) and the orphaned time import goes with them.

CHANGELOG.md

@@ -128,6 +128,15 @@ Earlier stable lines (`stable/1.6`, `stable/1.5`) are frozen.
 
 ### Changed
 
+- **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 — dual emission would double-render on
+  every current client). Consume the `compaction` lifecycle event (see
+  the API reference and the `CompactionEvent` SDK type); embedders
+  driving `ChatSession` through a duck-typed `SessionUI` are unaffected
+  (the classic `on_info` lines are restored for them — see Fixed).
+
 - **Sampling knobs (temperature, reasoning effort) now ride one assignment
   scheme: per-model alias value → operator-stored global setting → the
   model definition's declared default (effort only) → field omitted.**
@@ -216,8 +225,11 @@ Earlier stable lines (`stable/1.6`, `stable/1.5`) are frozen.
   honored rather than silently eaten; and Stop now aborts the in-flight
   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 are **deferred**: answered
+  immediate instead of waiting out a model call.
+
+- **Sends during a command window are deferred, ordered, bounded, and
+  honestly rendered — never silently truncated or lost.** Messages sent
+  while any 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
@@ -254,7 +266,12 @@ Earlier stable lines (`stable/1.6`, `stable/1.5`) are frozen.
   chip instead of a sent-looking bubble, releases the composer (a
   deferred send has no running worker to wait on), and cleans up fully
   when the send is refused or the chip retracted instead of stranding
-  the pane in Stop mode.
+  the pane in Stop mode. 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.
+
+- **Slash commands hold the worker slot with a loud 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,
@@ -269,7 +286,10 @@ Earlier stable lines (`stable/1.6`, `stable/1.5`) are frozen.
   a proxied pane, which now surfaces it instead of silence; the
   `/command` response contract — `ok` / `running`, with busy refusals
   answering a loud HTTP 409 rather than a silent 200 — is now documented
-  in the API reference and the OpenAPI spec). Manual compaction
+  in the API reference and the OpenAPI spec).
+
+- **Compaction status stays truthful across every UI surface.** Manual
+  compaction
   success also refreshes the status line/context pill immediately (parity
   with auto-compaction), compaction failures keep feeding the typed
   `error` event and the node error counter (while a CLI Ctrl-C reports as
@@ -282,10 +302,9 @@ Earlier stable lines (`stable/1.6`, `stable/1.5`) are frozen.
   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.
-  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
+
+- **Compaction failures report exactly once, to the right owner.** 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), failed-end
   notice suppression is computed once by the emitter (a `notice` bool on
@@ -299,13 +318,13 @@ Earlier stable lines (`stable/1.6`, `stable/1.5`) are frozen.
   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`
+
+- **Pre-1.8 embedder UIs keep their compaction lines.** 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).
+  summary box — instead of silent history swaps. (See the breaking
+  event-contract note under **Changed** for SSE/SDK clients.)
 
 - **Static MCP servers: a pushed catalog change no longer wedges the shared
   session (#839).** The static-path `*/list_changed` handler awaited its

tests/test_coord_rich_ws_state_payload.py

@@ -476,7 +476,7 @@ def test_coord_spawn_metrics_increments_messages_and_resets_tool_count() -> None
     ui = ConsoleCoordinatorUI(ws_id="coord-ws", user_id="u1")
     ui._ws_messages = 5
     ui._ws_turn_tool_calls = 3
-    _coord_spawn_metrics(MagicMock(), ui)
+    _coord_spawn_metrics(ui)
     assert ui._ws_messages == 6
     assert ui._ws_turn_tool_calls == 0
 
@@ -489,7 +489,7 @@ def test_coord_spawn_metrics_tolerates_ui_without_counters() -> None:
     class _StubUI:
         pass
 
-    _coord_spawn_metrics(MagicMock(), _StubUI())  # must not raise
+    _coord_spawn_metrics(_StubUI())  # must not raise
 
 
 # ---------------------------------------------------------------------------

tests/test_server_attachments_endpoints.py

@@ -105,6 +105,28 @@ def _auth(user: str) -> dict[str, str]:
     return {"Authorization": f"Bearer {_make_jwt(user)}"}
 
 
+def _harden_ws_mock(ws) -> None:
+    """Neutralize every truthy-Mock trap the /send dispatch path reads.
+
+    A bare ``MagicMock()`` auto-creates truthy attributes and callables,
+    which the route misreads: a truthy ``_closed`` makes ``send()``
+    refuse; truthy ``_pending_sends``/``_pending_drain`` (and a truthy
+    ``send_barrier_active()`` result — the route consults the barrier as
+    a METHOD) defer every send behind a phantom order barrier.  Every
+    NEW Workstream field the dispatch path reads gets added HERE, once —
+    not appended to each fixture (missing one copy made that fixture's
+    sends defer/hang with an error pointing nowhere near the cause).
+    Deliberately does NOT set ``_worker_running``: fixtures choose that
+    per scenario (one even relies on the truthy auto-Mock to force the
+    queue path).
+    """
+    ws._closed = False
+    ws._pending_sends = []
+    ws._pending_drain = None
+    ws.send_barrier_active = lambda: False
+    ws._lock = threading.RLock()
+
+
 # ---------------------------------------------------------------------------
 # Upload
 # ---------------------------------------------------------------------------
@@ -503,17 +525,7 @@ class TestSendMessageAttachments:
         ws.session = session
         ws.worker_thread = None
         ws._worker_running = False
-        ws._closed = False  # a bare Mock attr is truthy → send() would refuse
-        # Same truthy-Mock trap as _closed: the /send order barrier reads
-        # both — a bare Mock attr would defer every send behind a phantom
-        # pending list.
-        ws._pending_sends = []
-        ws._pending_drain = None
-        # The /send route consults the order barrier as a METHOD — a bare
-        # Mock attr would be a truthy callable result and defer every send
-        # behind a phantom barrier (same class as the fields above).
-        ws.send_barrier_active = lambda: False
-        ws._lock = threading.RLock()
+        _harden_ws_mock(ws)
         mgr.get.return_value = ws
         return captured, session
 
@@ -711,16 +723,7 @@ class TestQueuedSendWithAttachments:
         ws.session = session
         ws.worker_thread = worker
         ws._worker_running = True
-        ws._closed = False  # a bare Mock attr is truthy → send() would refuse
-        # Same truthy-Mock trap: the /send order barrier reads both — a
-        # bare Mock attr would defer every send behind a phantom list.
-        ws._pending_sends = []
-        ws._pending_drain = None
-        # The /send route consults the order barrier as a METHOD — a bare
-        # Mock attr would be a truthy callable result and defer every send
-        # behind a phantom barrier (same class as the fields above).
-        ws.send_barrier_active = lambda: False
-        ws._lock = threading.RLock()
+        _harden_ws_mock(ws)
         mgr.get.return_value = ws
         return captured
 
@@ -783,16 +786,9 @@ class TestBusyWorkerAttachments:
         ws.ui = ui
         ws.session = session
         ws.worker_thread = worker
-        ws._closed = False  # a bare Mock attr is truthy → send() would refuse
-        # Same truthy-Mock trap: the /send order barrier reads both — a
-        # bare Mock attr would defer every send behind a phantom list.
-        ws._pending_sends = []
-        ws._pending_drain = None
-        # The /send route consults the order barrier as a METHOD — a bare
-        # Mock attr would be a truthy callable result and defer every send
-        # behind a phantom barrier (same class as the fields above).
-        ws.send_barrier_active = lambda: False
-        ws._lock = threading.RLock()
+        # No explicit _worker_running: the truthy auto-Mock forces the
+        # queue path (deliberate — see _harden_ws_mock's exclusion note).
+        _harden_ws_mock(ws)
         mgr.get.return_value = ws
         return ws, session
 

tests/test_server_authz.py

@@ -12,7 +12,6 @@ from __future__ import annotations
 import json
 import queue
 import threading
-import time
 from typing import Any
 from unittest.mock import MagicMock
 
@@ -20,6 +19,7 @@ import pytest
 from starlette.testclient import TestClient
 
 from tests._helpers import wait_until
+from turnstone.core.workstream import INTERJECTION_CAP_CHARS
 
 _TEST_JWT_SECRET = "test-jwt-secret-minimum-32-chars!"
 
@@ -274,7 +274,8 @@ class _FakeSession:
         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
+        cap = INTERJECTION_CAP_CHARS
+        cleaned = text[:cap] + "..." if len(text) > cap else text
         return cleaned, "notice", queue_msg_id or "m1"
 
     def dequeue_message(self, msg_id: str) -> bool:
@@ -1338,9 +1339,9 @@ class TestCompactCommandDispatch:
         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."""
+        send — never the interjection queue, whose INTERJECTION_CAP_CHARS
+        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)
@@ -1531,11 +1532,7 @@ class TestCompactCommandDispatch:
         runner = threading.Thread(target=_cmd, daemon=True)
         runner.start()
         # Wait until the command worker actually holds the slot.
-        for _ in range(100):
-            if ws._worker_running:
-                break
-            time.sleep(0.02)
-        assert ws._worker_running
+        wait_until(lambda: ws._worker_running, timeout=8.0)
         assert ws.worker_kind == "command"
         r = client.post(
             f"/v1/api/workstreams/{ws_id}/send",
@@ -1591,11 +1588,7 @@ class TestCompactCommandDispatch:
 
         runner = threading.Thread(target=_cmd, daemon=True)
         runner.start()
-        for _ in range(100):
-            if ws._worker_running:
-                break
-            time.sleep(0.02)
-        assert ws._worker_running
+        wait_until(lambda: ws._worker_running, timeout=8.0)
         worker = ws.worker_thread
         # The cancel handler's force path shape: abandon the worker.
         with ws._lock:

turnstone/console/server.py

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

turnstone/core/session.py

@@ -204,7 +204,11 @@ from turnstone.core.trajectory import (
 )
 from turnstone.core.watch import WATCH_REMINDER_OPTIONAL_KEYS
 from turnstone.core.web import check_ssrf, fetch_with_ssrf_guard, strip_html
-from turnstone.core.workstream import PENDING_SENDS_MAX, WorkstreamKind
+from turnstone.core.workstream import (
+    INTERJECTION_CAP_CHARS,
+    PENDING_SENDS_MAX,
+    WorkstreamKind,
+)
 from turnstone.prompts import (
     INTERACTIVE_CONSENT_CLIENT_TYPES,
     ClientType,
@@ -9130,9 +9134,11 @@ class ChatSession:
             )
 
         cleaned, priority = parse_priority(text)
-        # Cap individual message length to prevent context bloat
-        if len(cleaned) > 2000:
-            cleaned = cleaned[:2000] + "..."
+        # Cap individual message length to prevent context bloat (the
+        # shared bound — the /send defer-fidelity check refuses fold-ins
+        # that could hit this truncation).
+        if len(cleaned) > INTERJECTION_CAP_CHARS:
+            cleaned = cleaned[:INTERJECTION_CAP_CHARS] + "..."
         # Full UUID hex (128 bits) rather than a truncated prefix — this id is
         # the ``send_id`` tracking token threaded through the turn, so the wide
         # space keeps the birthday bound comfortable.

turnstone/core/session_routes.py

@@ -40,7 +40,11 @@ from starlette.routing import Route
 
 from turnstone.core.log import get_logger
 from turnstone.core.session_ui_base import AutoApproveReason
-from turnstone.core.workstream import PENDING_SENDS_MAX, _PendingSend
+from turnstone.core.workstream import (
+    INTERJECTION_CAP_CHARS,
+    PENDING_SENDS_MAX,
+    _PendingSend,
+)
 
 if TYPE_CHECKING:
     import threading
@@ -104,12 +108,12 @@ AttachmentOwnerResolver = Callable[
     ["Request", str, "SessionManager"],
     tuple[str, "JSONResponse | None"],
 ]
-# (request, ui) — kind's spawn-time bookkeeping. Interactive bumps
+# (ui) — kind's spawn-time bookkeeping. Interactive bumps
 # ``_metrics.record_message_sent`` + per-UI message counters; coord
-# 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]
+# wires its per-UI-counter analog.  Deliberately request-free: the
+# pending-send drain dispatches deferred entries with no live request,
+# and no impl ever needed one.
+SpawnMetricsHook = Callable[[Any], None]
 
 
 class CancelForensics(Protocol):
@@ -346,7 +350,7 @@ class SessionEndpointConfig:
     - ``spawn_metrics``: optional bookkeeping hook fired once per
       ``send`` that spawns a fresh worker (queue-reuse path skips it).
       Interactive wires its WebUI per-conversation counters; coord
-      wires ``None``.
+      wires its per-UI-counter analog.
     - ``emit_message_queued``: when ``True`` and the dispatcher takes
       the live-worker enqueue path, the lifted body emits a
       ``message_queued`` event onto the workstream's listener queue
@@ -4015,7 +4019,6 @@ def _make_dispatch_attempt(
     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.
@@ -4036,17 +4039,15 @@ def _make_dispatch_attempt(
 
     ``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
+    interjection fallback — which truncates at ``queue_message``'s cap
+    (``workstream.INTERJECTION_CAP_CHARS``) 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
 
@@ -4070,7 +4071,14 @@ def _make_dispatch_attempt(
             if ws.worker_kind == "command":
                 queue_outcome["rejected"] = "command_window"
                 return
-            if defer_fidelity and (resolved_atts or len(message) > 2000):
+            # Measures the RAW message while queue_message caps the
+            # post-!!!-strip CLEANED text — deliberate: parse_priority
+            # only strips, so raw >= cleaned and the raw measure can
+            # only ever OVER-refuse (a borderline fold-in costs one
+            # full-fidelity fresh spawn, never a truncation); measuring
+            # cleaned here would run parse_priority twice per dispatch
+            # for zero safety gain.
+            if defer_fidelity and (resolved_atts or len(message) > INTERJECTION_CAP_CHARS):
                 queue_outcome["rejected"] = "defer_full_fidelity"
                 return
             try:
@@ -4141,7 +4149,7 @@ def _make_dispatch_attempt(
             # 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)
+                cfg.spawn_metrics(ui)
             except Exception:
                 log.debug(
                     "ws.send.spawn_metrics_failed ws=%s",
@@ -4384,8 +4392,8 @@ def make_send_handler(cfg: SessionEndpointConfig) -> Handler:
       ``True`` post-P1.5; the flag exists so a kind that hasn't
       lit up its UI surface yet can defer.
     - ``spawn_metrics``: when set, fires once on the spawn path with
-      ``(request, ui)``. Interactive wires its WebUI per-conversation
-      counters here; coord wires ``None``.
+      ``(ui)``. Interactive wires its WebUI per-conversation counters
+      here; coord wires its per-UI-counter analog.
     - ``emit_message_queued``: when ``True``, the queue-reuse path
       pushes a ``message_queued`` event onto the listener queue.
 
@@ -4523,8 +4531,9 @@ def make_send_handler(cfg: SessionEndpointConfig) -> Handler:
 
         # 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
+        # the interjection-queue path — its INTERJECTION_CAP_CHARS cap and
+        # cross-user guard are mid-TURN semantics, and a queued message
+        # would cross a
         # /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
@@ -4608,7 +4617,6 @@ def make_send_handler(cfg: SessionEndpointConfig) -> Handler:
                         ordered_taken=ordered_taken,
                         send_id=pending_msg_id,
                         acting_uid=acting_uid,
-                        request=None,
                         defer_fidelity=True,
                     ),
                 )
@@ -4705,7 +4713,6 @@ def make_send_handler(cfg: SessionEndpointConfig) -> Handler:
             ordered_taken=ordered_taken,
             send_id=send_id,
             acting_uid=acting_uid,
-            request=request,
         )
         session_now = ws.session
         if session_now is None:

turnstone/core/workstream.py

@@ -119,6 +119,15 @@ BULK_CLOSE_STATE_VALUES: frozenset[str] = frozenset(
 # turn, so acceptance must be bounded.
 PENDING_SENDS_MAX = 10
 
+# The interjection queue's per-message character cap, shared by its TWO
+# consumers so they cannot drift: ``ChatSession.queue_message`` truncates
+# the cleaned text at this bound, and the /send defer-fidelity check
+# refuses the interjection fold-in for any deferred entry whose text
+# could hit it (routing the entry to a full-fidelity fresh spawn
+# instead).  A drift between those sites is exactly the silent-truncation
+# bug the defer contract exists to prevent.
+INTERJECTION_CAP_CHARS = 2000
+
 
 @dataclass
 class _PendingSend:

turnstone/server.py

@@ -4563,7 +4563,7 @@ def create_app(
         magic-mocked ``ws.user_id``."""
         return _require_ws_access(request, ws_id)
 
-    def _interactive_spawn_metrics(_request: Request | None, ui: Any) -> None:
+    def _interactive_spawn_metrics(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