feat(coord): three-tier compression for inspect_workstream output

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

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

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": "<json-string>"}}, ...]``.  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
 # ---------------------------------------------------------------------------

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)

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