From d2f5db092d510bd397d7ef2c2a2a1045ae8214ef Mon Sep 17 00:00:00 2001 From: Patrick Buckley Date: Mon, 18 May 2026 19:19:01 -0700 Subject: [PATCH] feat(coord): three-tier compression for inspect_workstream output MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A coord doing a fan-out wave of inspect_workstream calls against tool-heavy children could blow the context budget on raw output alone (one child with a 100 KB bash result × N children). The previous safety net was ``_truncate_output``'s head+tail strategy, which silently drops *middle* messages — exactly the wrong shape for a coordinator trying to understand a child's trajectory (the LAST message tells the model what the child concluded; the FIRST sets the brief; the middle is the connective tissue). Three-tier degradation modeled on the search tool's pattern at ``session.py:_format_search_results``: Tier 1 (full): every message verbatim — used when size fits. Tier 2 (compact): per-message head/tail-snipped content (600/300 chars) plus snipped ``tool_calls.arguments`` (300/100 chars). When content snipping alone doesn't fit, fall through a message-list trim ladder ((20,30) → (10,20) → (5,10)) that keeps head + tail messages and elides the middle as ``{"_omitted": N}``. Tier 3 (skeleton): no messages — counts + role distribution + verdicts-by-risk + last assistant preview. Budget 32 KB (matches ``_SEARCH_OUTPUT_BUDGET``). First emission whose JSON serialization fits the budget wins. ``_tier`` lands on every non-error emission so the coordinator LLM and audit readers can see which compression rung was selected; ``_tier_note`` carries actionable advice (re-call with a smaller ``message_limit`` etc.). Error-shape results bypass tiering — they're already small. Bug fixes caught during review: - ``_compact_message`` now preserves the assistant-side ``tool_calls`` list with snipped ``function.arguments``; the pre-fix shape left audit readers with tool-result orphans against invisible calls. - The intermediate Tier-2 list-trim ladder fixes a size-monotonicity bug where Tier-2 with un-snippable content (per-message body under the 964-char threshold) plus the added ``_tier_note`` came out STRICTLY larger than Tier-1, falling through to skeleton when a head+tail trim would have preserved dozens of messages. - ``_inspect_skeleton`` reads ``result["skill_id"]`` (production storage row key) with a ``skill`` fallback; pre-fix it read ``skill`` only and emitted ``null`` for every real workstream. --- tests/test_coordinator_client.py | 359 ++++++++++++++++++++++++ turnstone/console/coordinator_client.py | 286 +++++++++++++++++++ turnstone/core/session.py | 11 +- turnstone/tools/inspect_workstream.json | 2 +- 4 files changed, 655 insertions(+), 3 deletions(-) diff --git a/tests/test_coordinator_client.py b/tests/test_coordinator_client.py index 1104f6aa..4cbe8241 100644 --- a/tests/test_coordinator_client.py +++ b/tests/test_coordinator_client.py @@ -2659,3 +2659,362 @@ def test_cleanup_dead_task_child_refs_storage_batch_failure_swallows(populated_s populated_storage.get_workstreams_batch = _boom # type: ignore[method-assign] assert client.cleanup_dead_task_child_refs("coord-1") == 0 + + +# --------------------------------------------------------------------------- +# inspect_workstream — three-tier output compression +# --------------------------------------------------------------------------- +# +# A coord doing a fan-out wave against tool-heavy children would +# otherwise blow the context budget on raw output alone. Mirrors the +# search tool's Tier-1/Tier-2/Tier-3 ladder. + + +def _make_inspect_result( + *, ws_id: str = "ws-test", state: str = "running", n_messages: int = 5 +) -> dict[str, Any]: + """Build an inspect-result dict shaped like ``coordinator_client.inspect()``. + + Production output keys (``ws_id``, ``skill_id``) mirror the storage + row that ``inspect()`` spreads from ``get_workstream``. Tests that + synthesize an inspect result must match these keys — otherwise a + formatter that looks at the production keys silently emits null + values against a fixture that uses different ones (real bug-1 + regression source: skeleton tier read ``skill`` from a fixture + that wrote ``skill`` while production wrote ``skill_id``). + """ + return { + "ws_id": ws_id, + "state": state, + "title": "test workstream", + "skill_id": "researcher", + "messages": [ + {"role": "user" if i % 2 == 0 else "assistant", "content": f"msg {i} content"} + for i in range(n_messages) + ], + "verdicts": [], + } + + +def test_format_inspect_tiered_full_fits_returns_full_tier(): + """Small payloads pass through with `_tier='full'` — no compression.""" + from turnstone.console.coordinator_client import _format_inspect_tiered + + result = _make_inspect_result(n_messages=3) + out = _format_inspect_tiered(result) + parsed = json.loads(out) + assert parsed["_tier"] == "full" + # Every message verbatim. + assert len(parsed["messages"]) == 3 + assert parsed["messages"][0]["content"] == "msg 0 content" + + +def test_format_inspect_tiered_compact_when_full_exceeds_budget(): + """Large messages trigger the compact tier — head/tail-snipped + content with the rest of the row intact.""" + from turnstone.console.coordinator_client import ( + _INSPECT_MSG_CONTENT_HEAD, + _INSPECT_MSG_CONTENT_TAIL, + _INSPECT_OUTPUT_BUDGET, + _format_inspect_tiered, + ) + + # Each message ~5KB; with 20 messages, full tier blows the 32KB budget. + fat = "X" * 5000 + result = { + "id": "ws-fat", + "state": "running", + "messages": [{"role": "assistant", "content": fat} for _ in range(20)], + "verdicts": [], + } + out = _format_inspect_tiered(result) + parsed = json.loads(out) + assert parsed["_tier"] == "compact" + # Every message preserved (compact keeps the count, just snips content). + assert len(parsed["messages"]) == 20 + # Head/tail snip kicked in. + msg_content = parsed["messages"][0]["content"] + assert msg_content.startswith("X" * _INSPECT_MSG_CONTENT_HEAD) + assert msg_content.endswith("X" * _INSPECT_MSG_CONTENT_TAIL) + assert "chars elided" in msg_content + # Budget invariant — the load-bearing contract of the formatter. + # Without this assertion, a future change to ``_tier_note`` or + # ``_compact_message`` could push the output over budget and the + # ``_truncate_output`` head+tail safety net would silently mask + # the regression, re-introducing the middle-message-drop pathology. + assert len(out) <= _INSPECT_OUTPUT_BUDGET + + +def test_format_inspect_tiered_compact_when_content_below_snip_threshold(): + """When per-message content is below the snip threshold but the + message COUNT alone overflows the budget, compact tier must still + stay within budget — by trimming the message list (head + tail of + messages) rather than degrading straight to skeleton. Bug-3 + regression cover: with 400 × 100-char messages, the original + formatter fell through to skeleton because adding ``_tier_note`` + to an un-snipped tier-2 produced output strictly larger than + tier-1 (both over budget). The fix preserves messages from both + ends of the list and inserts an ``_omitted`` sentinel.""" + from turnstone.console.coordinator_client import ( + _INSPECT_OUTPUT_BUDGET, + _format_inspect_tiered, + ) + + # 400 × ~100 chars → Tier-1 ~53 KB (over budget), per-message + # content under the 964-char snip threshold so content-snipping + # saves nothing. Without the list-trim rung the formatter would + # fall to skeleton and drop all 400 messages. + smallish = "S" * 100 + result = { + "ws_id": "ws-many-small", + "state": "running", + "messages": [ + {"role": "assistant" if i % 2 == 0 else "user", "content": smallish} for i in range(400) + ], + "verdicts": [], + } + out = _format_inspect_tiered(result) + parsed = json.loads(out) + # Should NOT fall through to skeleton — message-list trim preserves + # head + tail of the conversation. + assert parsed["_tier"] == "compact" + assert "messages" in parsed + # Some messages must survive; the trim shape is head + tail with an + # ``_omitted`` sentinel between them. + assert len(parsed["messages"]) > 0 + assert len(parsed["messages"]) < 400 + # Budget invariant. + assert len(out) <= _INSPECT_OUTPUT_BUDGET + + +def test_format_inspect_tiered_skeleton_when_compact_also_exceeds_budget(): + """Tier 3 fallback: counts + last assistant preview only. Trigger by + flooding with messages whose content is a multi-block list — the + snipper correctly leaves non-string content unchanged (mirrors + Anthropic/OpenAI multi-block content shape), so even after the + (5, 10) message-list trim the surviving 15 messages don't fit in + the 32 KB budget.""" + from turnstone.console.coordinator_client import ( + _INSPECT_OUTPUT_BUDGET, + _format_inspect_tiered, + ) + + # 50 messages × multi-block content (~30 KB each — list-shape + # content bypasses the head/tail string snipper because lists + # aren't strings). Even (5, 10) trim leaves 15 × 30 KB which + # blows the 32 KB budget — forces skeleton. + fat_block = {"type": "text", "text": "Y" * 3000} + result = { + "ws_id": "ws-flood", + "state": "running", + "title": "flood", + "skill_id": "researcher", + "messages": [ + { + "role": "assistant" if i % 2 == 0 else "user", + "content": [fat_block] * 10, + } + for i in range(50) + ], + "verdicts": [], + } + out = _format_inspect_tiered(result) + parsed = json.loads(out) + assert parsed["_tier"] == "skeleton" + assert parsed["message_count"] == 50 + # Role distribution surfaces — the "what shape of activity" signal. + assert parsed["roles"]["assistant"] == 25 + assert parsed["roles"]["user"] == 25 + # No `messages` field at skeleton tier — only the aggregate signal. + assert "messages" not in parsed + # Budget invariant. + assert len(out) <= _INSPECT_OUTPUT_BUDGET + + +def test_format_inspect_tiered_skeleton_keeps_terminal_state_fields(): + """``close_reason`` / ``last_error`` survive the skeleton fall — they're + small, load-bearing, and the operator needs them to understand WHY + a terminal child landed in its state.""" + from turnstone.console.coordinator_client import ( + _INSPECT_OUTPUT_BUDGET, + _format_inspect_tiered, + ) + + # Same flood pattern as the bare-skeleton test (multi-block content + # bypasses the string snipper) — paired with terminal-state fields + # that must survive the skeleton fall. + fat_block = {"type": "text", "text": "Z" * 3000} + result = { + "ws_id": "ws-closed", + "state": "closed", + "title": "done", + "skill_id": "researcher", + "messages": [{"role": "user", "content": [fat_block] * 10} for _ in range(50)], + "verdicts": [], + "close_reason": "task complete: report attached", + "live": None, # filtered by truthy check + } + out = _format_inspect_tiered(result) + parsed = json.loads(out) + assert parsed["_tier"] == "skeleton" + assert parsed["close_reason"] == "task complete: report attached" + # Falsy ``live`` doesn't bleed through. + assert "live" not in parsed + assert len(out) <= _INSPECT_OUTPUT_BUDGET + + +def test_format_inspect_tiered_error_shapes_bypass_tiering(): + """Cross-tenant / not-found responses keep their original shape — they + carry no messages, are already tiny, and changing them would break + callers that key on the ``error`` field.""" + from turnstone.console.coordinator_client import _format_inspect_tiered + + result = {"error": "workstream not found", "ws_id": "ws-foreign"} + out = _format_inspect_tiered(result) + parsed = json.loads(out) + assert parsed == {"error": "workstream not found", "ws_id": "ws-foreign"} + # No `_tier` annotation — error shapes are self-describing. + assert "_tier" not in parsed + + +def test_format_inspect_tiered_compact_preserves_tool_call_linkage(): + """Compact tier keeps ``tool_name`` / ``tool_call_id`` / ``name`` so a + model reading the snipped trace can still pair a tool call to its + response — the linkage is load-bearing for "what happened" signal.""" + from turnstone.console.coordinator_client import _format_inspect_tiered + + fat = "Q" * 5000 + result = { + "ws_id": "ws-tools", + "state": "running", + "messages": [ + { + "role": "assistant", + "content": fat, + "tool_name": "bash", + "tool_call_id": "call-1", + } + for _ in range(20) + ], + "verdicts": [], + } + out = _format_inspect_tiered(result) + parsed = json.loads(out) + assert parsed["_tier"] == "compact" + first = parsed["messages"][0] + assert first["tool_name"] == "bash" + assert first["tool_call_id"] == "call-1" + + +def test_format_inspect_tiered_compact_preserves_assistant_tool_calls(): + """Compact tier must preserve the assistant-side ``tool_calls`` list + (OpenAI shape: ``[{id, type, function: {name, arguments}}]``) so a + model reading the snipped trace can see WHICH tool was called and + pair it with the corresponding result row via ``id`` ↔ ``tool_call_id``. + Bug-2 regression cover: the pre-fix compactor stripped ``tool_calls``, + leaving the audit reader with a tool-result orphan against an + invisible call. + + ``function.arguments`` strings are snipped head/tail (analogous to + content) because they can be multi-KB JSON; ``id`` and + ``function.name`` are preserved verbatim — they're the linkage.""" + from turnstone.console.coordinator_client import ( + _INSPECT_TOOL_ARG_HEAD, + _INSPECT_TOOL_ARG_TAIL, + _format_inspect_tiered, + ) + + fat_content = "C" * 5000 # forces compact tier + fat_args = "A" * 5000 # forces argument snipping + tool_calls = [ + { + "id": "call-abc-123", + "type": "function", + "function": {"name": "bash", "arguments": fat_args}, + }, + { + "id": "call-def-456", + "type": "function", + "function": {"name": "read_file", "arguments": fat_args}, + }, + ] + result = { + "ws_id": "ws-tool-calls", + "state": "running", + "messages": [ + {"role": "assistant", "content": fat_content, "tool_calls": tool_calls} + for _ in range(20) + ], + "verdicts": [], + } + out = _format_inspect_tiered(result) + parsed = json.loads(out) + assert parsed["_tier"] == "compact" + first = parsed["messages"][0] + # tool_calls survives compaction. + assert "tool_calls" in first + assert len(first["tool_calls"]) == 2 + # Linkage fields verbatim. + assert first["tool_calls"][0]["id"] == "call-abc-123" + assert first["tool_calls"][0]["function"]["name"] == "bash" + assert first["tool_calls"][1]["id"] == "call-def-456" + assert first["tool_calls"][1]["function"]["name"] == "read_file" + # arguments snipped head/tail — both prefix and suffix preserved. + snipped_args = first["tool_calls"][0]["function"]["arguments"] + assert snipped_args.startswith("A" * _INSPECT_TOOL_ARG_HEAD) + assert snipped_args.endswith("A" * _INSPECT_TOOL_ARG_TAIL) + assert "chars elided" in snipped_args + + +def test_format_inspect_tiered_compact_passes_small_messages_through_unsnipped(): + """Messages under the snip threshold pass through verbatim at compact + tier — snipping a 100-byte message costs more bytes (the elision + marker) than it saves.""" + from turnstone.console.coordinator_client import _format_inspect_tiered + + # Mix: a few large messages force compact tier; small messages must + # not be snipped. + big = "B" * 5000 + small = "S" * 50 + result = { + "id": "ws-mixed", + "state": "running", + "messages": [{"role": "assistant", "content": big} for _ in range(15)] + + [{"role": "user", "content": small}], + "verdicts": [], + } + out = _format_inspect_tiered(result) + parsed = json.loads(out) + assert parsed["_tier"] == "compact" + # The trailing small message is exact, not snipped. + assert parsed["messages"][-1]["content"] == small + + +def test_format_inspect_tiered_emits_tier_note_when_compressed(): + """The ``_tier_note`` advisory tells the LLM how to ask for a tighter + or fuller view next time — actionable feedback rather than a bare + "we compressed your output" signal.""" + from turnstone.console.coordinator_client import _format_inspect_tiered + + fat = "F" * 5000 + result = { + "id": "ws-noted", + "state": "running", + "messages": [{"role": "assistant", "content": fat} for _ in range(20)], + "verdicts": [], + } + out = _format_inspect_tiered(result) + parsed = json.loads(out) + assert "_tier_note" in parsed + assert "message_limit" in parsed["_tier_note"] + + +def test_format_inspect_tiered_full_tier_omits_tier_note(): + """When the full tier fits, no note is emitted — the absence of a + note is the signal that nothing was compressed.""" + from turnstone.console.coordinator_client import _format_inspect_tiered + + out = _format_inspect_tiered(_make_inspect_result(n_messages=2)) + parsed = json.loads(out) + assert parsed["_tier"] == "full" + assert "_tier_note" not in parsed diff --git a/turnstone/console/coordinator_client.py b/turnstone/console/coordinator_client.py index 0d4db9d9..1faa19d9 100644 --- a/turnstone/console/coordinator_client.py +++ b/turnstone/console/coordinator_client.py @@ -1799,6 +1799,292 @@ def _serialize_verdicts(rows: list[Any]) -> list[dict[str, Any]]: return out +# --------------------------------------------------------------------------- +# inspect_workstream — tiered output compression +# --------------------------------------------------------------------------- +# +# A coord doing a fan-out wave of inspect_workstream calls against +# tool-heavy children can blow the context budget on raw output alone +# (one child with a 100 KB bash result × N children). The previous +# safety net was ``_truncate_output``'s head+tail strategy, which +# silently drops *middle* messages — exactly the wrong shape for a +# coordinator trying to understand a child's trajectory (the LAST +# message tells the model what the child concluded; the FIRST sets +# the brief; the middle is the connective tissue). +# +# The three-tier degradation pattern matches the ``search`` tool's +# Tier-1/Tier-2/Tier-3 ladder at ``session.py:_format_search_results``. +# First tier whose serialized size fits the budget wins; the LLM +# learns which tier it got via the ``_tier`` field in the response +# (no API change to the coordinator tool). +# +# Budget chosen well under ``tool_truncation`` (typically 256 KB+) so +# the head+tail safety net never fires for inspect_workstream — that +# strategy silently drops middle messages, which is exactly the +# pathology this formatter exists to avoid. + +_INSPECT_OUTPUT_BUDGET: int = 32_768 +# Per-message head/tail snip when Tier 2 needs to compress content. +# Head dominates because the first ~600 chars of an assistant message +# usually contains the conclusion / direction; the tail is the +# follow-through. Tool results compress similarly: head shows what +# the tool was asked / what it found at the top; tail shows the final +# state / error suffix. +_INSPECT_MSG_CONTENT_HEAD: int = 600 +_INSPECT_MSG_CONTENT_TAIL: int = 300 +# Threshold below which a message's content passes through unsnipped +# even in Tier 2. Snipping a sub-1KB message costs more bytes (the +# elision marker) than it saves. +_INSPECT_MSG_SNIP_THRESHOLD: int = _INSPECT_MSG_CONTENT_HEAD + _INSPECT_MSG_CONTENT_TAIL + 64 +# Skeleton-tier preview length on the last assistant message. Single +# value because the skeleton wants ONE meaningful signal ("what did +# the child last say"), not a head/tail snip. +_INSPECT_SKELETON_LAST_PREVIEW: int = 400 + +# Snip lengths for tool-call ``function.arguments`` strings on +# assistant turns. Tighter than content snipping because tool calls +# often appear in clusters (10+ per turn for a fan-out) and the +# arguments JSON is dense — keep just enough to see what was invoked +# and the head of the args structure. +_INSPECT_TOOL_ARG_HEAD: int = 300 +_INSPECT_TOOL_ARG_TAIL: int = 100 +_INSPECT_TOOL_ARG_SNIP_THRESHOLD: int = _INSPECT_TOOL_ARG_HEAD + _INSPECT_TOOL_ARG_TAIL + 64 + +# Message-list trim ladder for the compact tier when per-message +# content snipping alone doesn't free enough budget. Each rung is +# ``(head_count, tail_count)`` — keep the first N + last M messages, +# elide the middle as ``{"_omitted": K}``. Tail-weighted because the +# last assistant turn carries the load-bearing "what did the child +# conclude" signal (same rationale as ``_inspect_skeleton``'s +# last-assistant preview). Tried in order; first rung whose +# serialized emission fits the budget wins. Mirrors the per-file +# sample ladder in ``_format_search_results`` at session.py:254. +_INSPECT_LIST_TRIM_LADDER: tuple[tuple[int, int], ...] = ((20, 30), (10, 20), (5, 10)) + + +def _snip_head_tail(text: str, head: int, tail: int) -> str: + """Head/tail snip with elision marker; passthrough when shorter than threshold.""" + if not isinstance(text, str) or len(text) <= head + tail + 64: + return text + elided = len(text) - head - tail + return text[:head] + f"\n...[{elided} chars elided]...\n" + text[-tail:] + + +def _compact_tool_calls(tool_calls: Any) -> Any: + """Snip ``function.arguments`` on each tool-call entry; keep ``id`` and + ``function.name`` verbatim. + + OpenAI shape: ``[{"id": ..., "type": "function", "function": + {"name": ..., "arguments": ""}}, ...]``. The + arguments string is the dominant size term on a fan-out turn that + issued many tool calls with multi-KB JSON arguments each; + preserving them verbatim re-opens the same size pressure the + compact tier is trying to relieve. Non-list / non-dict entries + pass through so a future shape change doesn't crash the formatter. + """ + if not isinstance(tool_calls, list): + return tool_calls + out: list[Any] = [] + for call in tool_calls: + if not isinstance(call, dict): + out.append(call) + continue + compact_call: dict[str, Any] = {} + for k in ("id", "type"): + v = call.get(k) + if v: + compact_call[k] = v + func = call.get("function") + if isinstance(func, dict): + compact_func: dict[str, Any] = {} + name = func.get("name") + if name: + compact_func["name"] = name + args = func.get("arguments", "") + if args: + compact_func["arguments"] = _snip_head_tail( + args, _INSPECT_TOOL_ARG_HEAD, _INSPECT_TOOL_ARG_TAIL + ) + compact_call["function"] = compact_func + out.append(compact_call) + return out + + +def _compact_message(msg: dict[str, Any]) -> dict[str, Any]: + """Tier-2 per-message projection: keep role + identifier keys, snip content + tool_calls. + + Tool-call linkage is the load-bearing "what happened" signal: + ``tool_call_id`` on the result side matches an ``id`` in + ``tool_calls`` on the issuing assistant turn. Stripping + ``tool_calls`` (the pre-fix shape) left tool results dangling + against an invisible call — the audit reader could see "bash + returned X" but not "the assistant asked for ``ls /tmp``". The + ``arguments`` string is the size offender, so we snip it head/tail + rather than dropping the call entirely. + """ + content = msg.get("content", "") + snipped = _snip_head_tail(content, _INSPECT_MSG_CONTENT_HEAD, _INSPECT_MSG_CONTENT_TAIL) + compact: dict[str, Any] = {"role": msg.get("role"), "content": snipped} + # Tool-result linkage (result-side keys). + for k in ("tool_name", "tool_call_id", "name"): + v = msg.get(k) + if v: + compact[k] = v + # Tool-call request linkage (issuing-side list), snipped per-call. + tool_calls = msg.get("tool_calls") + if tool_calls: + compact["tool_calls"] = _compact_tool_calls(tool_calls) + return compact + + +def _inspect_skeleton(result: dict[str, Any]) -> dict[str, Any]: + """Tier-3 fallback: state + counts + last assistant preview + terminal info. + + Drops every message, keeping only aggregate signal: state, message + count, role distribution, verdict count + risk distribution, and a + short preview of the most recent assistant turn (the "what did this + child last say" signal). Terminal-state fields (``close_reason``, + ``last_error``) and the ``live`` block pass through unchanged + because they're already small and load-bearing. + """ + messages = result.get("messages") or [] + verdicts = result.get("verdicts") or [] + role_counts: dict[str, int] = {} + for m in messages: + role = m.get("role") if isinstance(m, dict) else None + if role: + role_counts[role] = role_counts.get(role, 0) + 1 + verdicts_by_risk: dict[str, int] = {} + for v in verdicts: + if isinstance(v, dict): + risk = v.get("risk_level") or "unknown" + verdicts_by_risk[risk] = verdicts_by_risk.get(risk, 0) + 1 + last_preview = "" + for m in reversed(messages): + if not isinstance(m, dict) or m.get("role") != "assistant": + continue + c = m.get("content", "") + if isinstance(c, str) and c: + last_preview = c[:_INSPECT_SKELETON_LAST_PREVIEW] + if len(c) > _INSPECT_SKELETON_LAST_PREVIEW: + last_preview += "..." + break + skeleton: dict[str, Any] = { + # Storage row keys verbatim from ``get_workstreams_batch`` + # (the projection backing ``get_workstream`` → ``inspect()``): + # ``ws_id``, ``skill_id``. No fallback to ``id`` / ``skill`` + # — fail loud on storage column drift rather than silently + # emitting null. + "ws_id": result["ws_id"], + "state": result.get("state"), + "title": result.get("title"), + "skill": result["skill_id"], + "message_count": len(messages), + "roles": role_counts, + "verdict_count": len(verdicts), + "verdicts_by_risk": verdicts_by_risk, + "last_assistant_preview": last_preview, + "_tier": "skeleton", + "_tier_note": ( + "Output exceeded the inspect_workstream budget at both full and compact " + "tiers; skeleton-only. Re-call with a smaller ``message_limit`` to fit " + "the compact tier, or read individual messages via the storage admin path." + ), + } + for k in ("close_reason", "last_error", "live"): + v = result.get(k) + if v: + skeleton[k] = v + return skeleton + + +def _format_inspect_tiered(result: dict[str, Any], *, budget: int = _INSPECT_OUTPUT_BUDGET) -> str: + """Serialize an ``inspect_workstream`` result with tiered degradation. + + Tier 1 (full): every message verbatim — used when the size fits. + Tier 2 (compact): per-message ``{role, head/tail-snipped content, + tool linkage, snipped tool_calls.arguments}`` for + every message, then a head+tail message-list trim + ladder when content snipping alone doesn't free + enough budget. + Tier 3 (skeleton): no messages — counts + last assistant preview only. + + First emission whose JSON serialization fits ``budget`` wins. + ``_tier`` appears on every non-error emission so the coordinator + LLM (and any audit reader) can see which compression rung the + output landed on without inferring from length. Error-shape + results (missing or cross-tenant ws_id) bypass tiering entirely — + they're already small and the ``error`` key signals the shape. + + The intermediate Tier-2 list-trim rungs exist because content + snipping alone fails on workloads where many small messages + overflow the budget by sheer count (``message_limit=200`` × a few + hundred chars each). In that regime, dropping content-snipping + saves zero bytes per message, so without the list-trim ladder + Tier-2 produces output strictly larger than Tier-1 (added + ``_tier_note``) and the formatter fell through to skeleton — + losing every message when a head+tail message-list trim would + have preserved dozens. Mirrors the per-file sample ladder in + ``_format_search_results`` (session.py:_SEARCH_TIER2_SAMPLE_LADDER). + """ + if "error" in result: + # Cross-tenant guard / not-found responses — pass through. + return json.dumps(result, default=str, separators=(",", ":")) + tier1 = {**result, "_tier": "full"} + out1 = json.dumps(tier1, default=str, separators=(",", ":")) + if len(out1) <= budget: + return out1 + messages = result.get("messages") or [] + compact_msgs = [_compact_message(m) if isinstance(m, dict) else m for m in messages] + tier2_note_full = ( + "Output exceeded the inspect_workstream budget at the full tier; messages " + "are head/tail-snipped at " + f"{_INSPECT_MSG_CONTENT_HEAD}/{_INSPECT_MSG_CONTENT_TAIL} chars. Re-call " + "with a smaller ``message_limit`` for a tighter tail, or include_provider_" + "content=False if it was on." + ) + tier2 = { + **result, + "messages": compact_msgs, + "_tier": "compact", + "_tier_note": tier2_note_full, + } + out2 = json.dumps(tier2, default=str, separators=(",", ":")) + if len(out2) <= budget: + return out2 + # Tier-2 list-trim ladder: keep head N + tail M, elide the middle. + # Tail-weighted because the recent turns carry the load-bearing + # signal ("what did the child conclude") — same reason + # ``_inspect_skeleton`` keeps a last-assistant preview rather than + # a first-user preview. + total = len(compact_msgs) + for head_n, tail_n in _INSPECT_LIST_TRIM_LADDER: + if head_n + tail_n >= total: + # Rung doesn't actually trim — would re-emit Tier-2 verbatim. + continue + omitted = total - head_n - tail_n + trimmed: list[Any] = ( + compact_msgs[:head_n] + [{"_omitted": omitted}] + compact_msgs[-tail_n:] + ) + tier2_trim_note = ( + f"Output exceeded the inspect_workstream budget at the compact tier; " + f"keeping first {head_n} + last {tail_n} of {total} messages, eliding " + f"{omitted} middle messages. Re-call with a smaller ``message_limit`` " + "to fit the full compact tier." + ) + tier2_trim = { + **result, + "messages": trimmed, + "_tier": "compact", + "_tier_note": tier2_trim_note, + } + out2_trim = json.dumps(tier2_trim, default=str, separators=(",", ":")) + if len(out2_trim) <= budget: + return out2_trim + skeleton = _inspect_skeleton(result) + return json.dumps(skeleton, default=str, separators=(",", ":")) + + # --------------------------------------------------------------------------- # wait_for_workstream — last-message extraction # --------------------------------------------------------------------------- diff --git a/turnstone/core/session.py b/turnstone/core/session.py index 170d2b20..61dce672 100644 --- a/turnstone/core/session.py +++ b/turnstone/core/session.py @@ -7252,6 +7252,8 @@ class ChatSession: } def _exec_inspect_workstream(self, item: dict[str, Any]) -> tuple[str, str]: + from turnstone.console.coordinator_client import _format_inspect_tiered + call_id = item["call_id"] ws_id = item["ws_id"] try: @@ -7264,8 +7266,13 @@ class ChatSession: msg = f"Error: inspect_workstream failed: {e}" self._report_tool_result(call_id, "inspect_workstream", msg, is_error=True) return call_id, msg - output = json.dumps(result, default=str, separators=(",", ":")) - # Summary for UI: state + message count + # Tiered output: full → compact (head/tail-snipped messages) → + # skeleton (counts + last-assistant preview). First tier that + # fits the budget wins; the LLM sees a ``_tier`` field on every + # non-error response. ``_truncate_output`` remains the safety + # net for the (rare) skeleton-exceeds-budget case — guarding + # against a single-field blowup we didn't anticipate. + output = _format_inspect_tiered(result) desc = f"{result.get('state', '?')} ({len(result.get('messages', []))} msgs)" self._report_tool_result(call_id, "inspect_workstream", desc) return call_id, self._truncate_output(output) diff --git a/turnstone/tools/inspect_workstream.json b/turnstone/tools/inspect_workstream.json index b19f10e0..f1b17218 100644 --- a/turnstone/tools/inspect_workstream.json +++ b/turnstone/tools/inspect_workstream.json @@ -1,6 +1,6 @@ { "name": "inspect_workstream", - "description": "Read a workstream's persisted state: state, title, skill, timestamps, last N messages. A `live` block from the owning node merges in when available (current tokens, activity, `pending_approval`) and reflects the node currently holding in-memory state, which can differ from the stored `node_id` if cluster membership shifted since spawn. `close_reason` surfaces when the workstream was closed with one. Provider-native content blocks are stripped by default for compactness; pass `include_provider_content=true` for the full-fidelity payload (replay tooling).", + "description": "Read a workstream's persisted state: state, title, skill, timestamps, last N messages. A `live` block from the owning node merges in when available (current tokens, activity, `pending_approval`) and reflects the node currently holding in-memory state, which can differ from the stored `node_id` if cluster membership shifted since spawn. `close_reason` surfaces when the workstream was closed with one. Provider-native content blocks are stripped by default for compactness; pass `include_provider_content=true` for the full-fidelity payload (replay tooling). Output uses three-tier compression to stay within a ~32 KB budget: `_tier=\"full\"` (every message verbatim), `_tier=\"compact\"` (messages head/tail-snipped at 600/300 chars), or `_tier=\"skeleton\"` (counts + last assistant preview only, when even compact didn't fit). Check the `_tier` field to know which shape you got; re-call with a smaller `message_limit` if you landed on skeleton and need more detail.", "parameters": { "type": "object", "properties": {