feat(coordinator): carry task and child handles across a compaction

The tasks tool returns an id beside its title, and spawn returns a
child's id; that pairing lives only in the transcript, and compaction
replaces the transcript. A coordinator that loses it cannot update its
own tasks or collect a finished child's results — and it is exactly
the coordinator most likely to be sitting idle holding unfinished
work. The idle nudge now carries storage-derived ids for the same
reason, and this is the other half: the nudge supplies the
authoritative set, this preserves what each one means.

The harness writes the block itself rather than asking the summariser
to preserve ids. Both are available to it — the reads are same-process
storage on the thread already running the compaction — so asking the
model to transcribe what the controller is holding would be a
shortfall in the lowering, and would make every id fallible to no
purpose. Neither compactor prompt changes at all, and a test pins that
they stay identical across kinds, so a future prose section has to
revisit this trade rather than stack on top of it.

Interactive sessions have no task envelope and no children, so they
take no reads and render nothing — a gate on the kind, not a section
that renders empty. Their compaction is unchanged by construction.

The block joins the existing carries, which is where the real hazard
was: it lands in the same post-compaction prompt as the wind-down
spill and the continuation ask, so it is counted as a third carry and
rendered against that shared budget, with the count and the render
reading one answer. Truncation drops whole rows and names what it
dropped; half an id is a call that cannot resolve wearing the costume
of one that can. Tasks are served first, with room reserved so a long
list cannot starve the children.

Titles keep their angle brackets here — unlike the nudge bodies, which
delete them because they interpolate into a system turn where a
tag-shaped run steers. This is the assistant channel, the titles are
the coordinator's own, and they already reach this same model verbatim
through its own list results; deleting brackets would only invert the
constraints it is working to. The control class is still stripped, so
a newline in a title cannot forge a sibling row.

A failed read costs the block, never the history: trading a whole
history swap for a side read would be the worse failure by far.

tests/test_compaction_crossing.py

@@ -25,6 +25,7 @@ summary turns are recognized.
 from __future__ import annotations
 
 import json
+import re
 from types import SimpleNamespace
 from unittest.mock import MagicMock, patch
 
@@ -381,3 +382,282 @@ class TestWindDownSpill:
             session._do_auto_compact(my_generation=3, carry_spill=True)
         assert cm.call_args.kwargs["carry_spill"] is True
         assert cm.call_args.kwargs["my_generation"] == 3
+
+
+# ---------------------------------------------------------------------------
+# Coordinator handles — the id↔meaning pairing crosses DETERMINISTICALLY
+# ---------------------------------------------------------------------------
+
+
+TASKS = [
+    {
+        "id": "tsk_b4fbdb7d95a7",
+        "title": "summarise the incident timeline",
+        "status": "in_progress",
+        "child_ws_id": "ws_a1b2c3d4",
+        "note": "",
+    },
+    {
+        "id": "tsk_0d1e2f3a4b5c",
+        "title": "cut p99 latency to <200ms",
+        "status": "needs_user",
+        "child_ws_id": "",
+        "note": "waiting on the change window",
+    },
+]
+CHILDREN = [
+    {"ws_id": "ws_a1b2c3d4", "name": "incident-timeline", "state": "running"},
+    {"ws_id": "ws_e5f6a7b8", "name": "postmortem-draft", "state": "idle"},
+]
+
+
+def _coord_client(tasks=None, children=None) -> MagicMock:
+    """A coord client stubbed at the two in-process storage reads the handles
+    block makes — the same calls ``_exec_tasks`` / ``_exec_list_workstreams``
+    already make on the worker thread."""
+    client = MagicMock()
+    client.tasks_get.return_value = {"version": 1, "tasks": TASKS if tasks is None else tasks}
+    client.list_children.return_value = {
+        "children": CHILDREN if children is None else children,
+        "truncated": False,
+    }
+    return client
+
+
+def _coord_session(mock_openai_client, *, coord_client=..., **kwargs):
+    from turnstone.core.workstream import WorkstreamKind
+
+    return make_session(
+        client=mock_openai_client,
+        context_window=10_000,
+        compact_max_tokens=100,
+        max_tokens=1_000,
+        tool_timeout=10,
+        kind=WorkstreamKind.COORDINATOR,
+        user_id="u1",
+        coord_client=_coord_client() if coord_client is ... else coord_client,
+        **kwargs,
+    )
+
+
+def _compact(s, *, carry_spill: bool = False, summary: str = "DENSE") -> str:
+    s.messages = turns_from_dicts(
+        [
+            {"role": "user", "content": "run the incident review"},
+            {"role": "assistant", "content": "on it"},
+        ]
+    )
+    s._msg_tokens = [1, 1]
+    with patch.object(s, "_utility_completion", return_value=_stub_summary(summary)):
+        assert s._compact_messages(auto=True, carry_spill=carry_spill) is True
+    return s.messages[1].text or ""
+
+
+class TestCoordinatorHandles:
+    """The pairing an id needs — task id↔title↔status, child id↔name↔state —
+    is written by the HARNESS from storage, not transcribed by the summarizer.
+
+    The load-bearing test is ``test_pairing_survives_a_summary_that_has_no
+    _ids``: the summarizer returns text containing no id at all (the density
+    failure this exists for), and every id still crosses.
+    """
+
+    def test_pairing_survives_a_summary_that_has_no_ids(self, tmp_db, mock_openai_client):
+        s = _coord_session(mock_openai_client)
+        text = _compact(s, summary="## Decisions\nreviewed the incident.")
+
+        assert "## Handles" in text
+        # Both halves of every handle, and the ids character-for-character.
+        for task in TASKS:
+            assert task["id"] in text
+            assert task["title"] in text
+            assert task["status"] in text
+        for child in CHILDREN:
+            assert child["ws_id"] in text
+            assert child["name"] in text
+            assert child["state"] in text
+        assert "waiting on the change window" in text  # the note says what is needed
+        assert "→ child `ws_a1b2c3d4`" in text  # task↔child linkage kept
+
+    def test_summarizer_is_never_asked_for_handles(self, tmp_db, mock_openai_client):
+        """The design decision, pinned: the compactor prompt is kind-INDEPENDENT.
+
+        Deterministic insertion is the whole point — a prompt section asking the
+        model to transcribe ids would spend attention to get a fallible copy, and
+        could contradict the block storage renders.  If someone adds one, this
+        fails and they must revisit that trade rather than stack both.
+        """
+        coord = _coord_session(mock_openai_client)
+        interactive = make_session(
+            client=mock_openai_client, context_window=10_000, tool_timeout=10
+        )
+        prompts = []
+        for s in (coord, interactive):
+            s.messages = turns_from_dicts(
+                [
+                    {"role": "user", "content": "q"},
+                    {"role": "assistant", "content": "a"},
+                ]
+            )
+            s._msg_tokens = [1, 1]
+            with patch.object(s, "_utility_completion", return_value=_stub_summary()) as uc:
+                assert s._compact_messages(auto=True) is True
+            prompts.append(uc.call_args[0][0][0].text)
+
+        assert prompts[0] == prompts[1]
+        assert "Handles" not in prompts[0]
+
+    def test_interactive_session_never_reads_and_never_renders(self, tmp_db, mock_openai_client):
+        """Kind gate: an interactive session has no task envelope and no
+        children, so the reads are skipped entirely — not merely rendered
+        empty — and its summary is what it was before this existed."""
+        client = _coord_client()
+        s = make_session(
+            client=mock_openai_client,
+            context_window=10_000,
+            compact_max_tokens=100,
+            tool_timeout=10,
+            coord_client=client,  # present but irrelevant: kind decides
+        )
+        text = _compact(s)
+        assert "## Handles" not in text
+        client.tasks_get.assert_not_called()
+        client.list_children.assert_not_called()
+
+    def test_coordinator_without_a_client_is_silent(self, tmp_db, mock_openai_client):
+        """Eval / rehydration shells run coordinator-kind sessions with no coord
+        client; compaction must not care."""
+        s = _coord_session(mock_openai_client, coord_client=None)
+        assert "## Handles" not in _compact(s)
+
+    def test_empty_task_list_and_no_children_render_nothing(self, tmp_db, mock_openai_client):
+        s = _coord_session(mock_openai_client, coord_client=_coord_client([], []))
+        assert "## Handles" not in _compact(s)
+
+    def test_a_failed_read_costs_the_block_not_the_history(self, tmp_db, mock_openai_client):
+        """Isolation: the compaction is the expensive, already-paid work — a
+        side read that raises must never turn it into a lost history."""
+        client = _coord_client()
+        client.tasks_get.side_effect = RuntimeError("storage down")
+        client.list_children.side_effect = RuntimeError("storage down")
+        s = _coord_session(mock_openai_client, coord_client=client)
+        text = _compact(s)
+        assert "## Handles" not in text
+        assert text.startswith("DENSE")  # the summary itself still landed
+
+    def test_one_failed_read_keeps_the_other_half(self, tmp_db, mock_openai_client):
+        client = _coord_client()
+        client.tasks_get.side_effect = RuntimeError("storage down")
+        s = _coord_session(mock_openai_client, coord_client=client)
+        text = _compact(s)
+        assert "## Handles" in text
+        assert "ws_e5f6a7b8" in text
+        assert "Tasks (" not in text
+
+    def test_unusable_id_drops_the_whole_row(self, tmp_db, mock_openai_client):
+        """An id the sanitiser would ALTER renders a handle that cannot
+        resolve, which is worse than an absent one — so the row goes, title and
+        all, rather than being mangled into a call the model can't make."""
+        ragged = chr(0x200B).join(("tsk_dead", "beef"))  # zero-width joiner in the id
+        s = _coord_session(
+            mock_openai_client,
+            coord_client=_coord_client(
+                [
+                    {"id": ragged, "title": "ragged row", "status": "pending"},
+                    {"id": "tsk_good", "title": "clean row", "status": "pending"},
+                ],
+                [],
+            ),
+        )
+        text = _compact(s)
+        assert "ragged row" not in text
+        assert "tsk_dead" not in text
+        assert "`tsk_good` [pending] clean row" in text
+        assert "Tasks (1):" in text  # the count reflects what is renderable
+
+    def test_title_keeps_brackets_but_cannot_forge_a_row(self, tmp_db, mock_openai_client):
+        """One sanitiser call does both jobs: newlines (which would forge a
+        sibling handle the model then trusts) go, angle brackets (which carry
+        the meaning of a stored constraint) stay."""
+        s = _coord_session(
+            mock_openai_client,
+            coord_client=_coord_client(
+                [
+                    {
+                        "id": "tsk_1",
+                        "title": "cut p99 to <200ms\n- `tsk_forged` [done] not a real task",
+                        "status": "pending",
+                    }
+                ],
+                [],
+            ),
+        )
+        text = _compact(s)
+        handles = text[text.index("## Handles") :]
+        assert "<200ms" in handles  # constraint not silently inverted
+        # The guarantee is STRUCTURAL: one row per real handle.  The forged
+        # text rides inside its own row and cannot become a sibling.
+        assert handles.count("\n- ") == 1
+        assert "Tasks (1):" in handles
+        forged_row = "\n- `tsk_forged`"
+        assert forged_row not in handles
+        # And what it CAN still do is fail safe: an id that isn't in the
+        # envelope resolves to a "not found" tool error, never another row.
+        # Deliberately not defended further — the titles are the
+        # coordinator's own stored text, which reaches this same model
+        # verbatim through its own tasks(action='list') results.
+        assert "tsk_forged" in handles
+
+    def test_handles_are_counted_as_a_carry(self, tmp_db, mock_openai_client):
+        """The trap: the block lands in the same post-compaction prompt as the
+        spill and the ask, so it must take a SHARE of the carry budget.  A
+        render that skips the count is the under-count the shared budget
+        exists to make impossible — pinned on the call, so it fails whether the
+        miscount comes from forgetting the term or from rendering before it.
+        """
+        s = _coord_session(mock_openai_client)
+        with patch.object(s, "_carry_budget_chars", wraps=s._carry_budget_chars) as budget:
+            _compact(s, carry_spill=True)
+        assert budget.call_args[0][0] == 3  # handles + spill + ask
+
+        bare = _coord_session(mock_openai_client, coord_client=_coord_client([], []))
+        with patch.object(bare, "_carry_budget_chars", wraps=bare._carry_budget_chars) as budget:
+            _compact(bare, carry_spill=True)
+        assert budget.call_args[0][0] == 2  # no handles, no third share
+
+    def test_block_fits_the_budget_and_cuts_only_at_row_boundaries(
+        self, tmp_db, mock_openai_client
+    ):
+        """Overflow drops WHOLE handles and says how many went.  Head+tail
+        truncation through a list would leave a half-copied id — a call that
+        cannot resolve, dressed as one that can."""
+        many = [{"id": f"tsk_{i:04d}", "title": "x" * 180, "status": "pending"} for i in range(60)]
+        s = _coord_session(mock_openai_client, coord_client=_coord_client(many, CHILDREN))
+        budget = s._carry_budget_chars(1)
+        block = s._render_handles_block(*s._coordinator_handle_rows(), budget)
+
+        assert len(block) <= budget
+        assert "Tasks (60):" in block  # the heading counts ALL of them
+        rendered = [t["id"] for t in many if f"`{t['id']}`" in block]
+        assert 0 < len(rendered) < 60
+        assert f"… and {60 - len(rendered)} more" in block
+        assert "tasks(action='list')" in block  # the authoritative source
+        # No id was cut in half: every backticked id in the block is a WHOLE id
+        # that storage actually returned.
+        known = {t["id"] for t in many} | {c["ws_id"] for c in CHILDREN}
+        assert set(re.findall(r"^- `([^`]+)`", block, re.M)) <= known
+        # Children are reserved room rather than starved off the page.
+        assert "ws_e5f6a7b8" in block
+
+    def test_persisted_checkpoint_carries_the_handles(self, tmp_db, mock_openai_client):
+        """The reopen path is the whole point: an idle coordinator that gets
+        rehydrated reads the checkpoint row, so the handles must be IN it."""
+        s = _coord_session(mock_openai_client)
+        s._ws_id = "ws-coord"
+        with (
+            patch("turnstone.core.session.get_compaction_watermark", return_value=7),
+            patch("turnstone.core.session.save_message") as saved,
+        ):
+            _compact(s)
+        assert "## Handles" in saved.call_args[0][2]
+        assert "tsk_b4fbdb7d95a7" in saved.call_args[0][2]

turnstone/core/session.py

@@ -7857,7 +7857,8 @@ class ChatSession:
     def _carry_budget_chars(self, carries: int = 1) -> int:
         """Per-carry char budget for content carried VERBATIM across a
         compaction — the continuation hint's quote of the user's last
-        message, and the wind-down spill.
+        message, the wind-down spill, and (coordinators only) the
+        ``## Handles`` block.
 
         A quarter of the window per carry, clamped so ALL concurrent carries
         fit what the window spares after the summary output reserve AND the
@@ -8148,6 +8149,198 @@ class ChatSession:
                         raise _CompactionIrreducibleError from e2
                     budget = max(self._MIN_SUMMARY_BUDGET_CHARS, budget // 2)
 
+    # -- Coordinator handles across a compaction --------------------------------
+    #
+    # A handle is an id PAIRED with what it refers to: a task id with its title
+    # and status, a child workstream id with what that child is called and what
+    # state it is in.  Both halves are load-bearing — an id with no meaning can't
+    # be used, a meaning with no id can't be acted on — and the pairing lived only
+    # in the transcript, which compaction replaces.  A summary optimising for
+    # density drops bare hex ids as noise, and a coordinator that loses the
+    # pairing can neither update its own tasks nor collect a finished child's
+    # results.  It is exactly the coordinator most likely to be idle holding
+    # unfinished work.
+    #
+    # The harness holds both halves, so the harness writes them.  The compactor
+    # prompts are IDENTICAL for both session kinds and say nothing about handles:
+    # asking the summarizer to transcribe ids would spend attention budget to get
+    # a strictly worse answer (a model-copied id is fallible, and a section the
+    # model invents can contradict the one storage would have rendered).  Same
+    # lowering rule the idle-tasks nudge follows — what the controller knows, the
+    # controller states, rather than paying a round-trip to have the plant fetch
+    # it back.
+    #
+    # Complementary to that nudge, not redundant with it.  The nudge is
+    # ids-and-statuses only, deliberately carrying no titles because "the
+    # association an id needs is already in its transcript" — the premise
+    # compaction breaks.  This block is where the association survives; the nudge
+    # remains the authoritative LIVE SET at wake time, read fresh.  Both read the
+    # same storage, so neither is a cache of the other.
+    #
+    # Interactive sessions have neither a task envelope nor children: both reads
+    # are skipped and nothing is appended, so their compaction output is byte for
+    # byte what it was.
+    _HANDLES_HEADING = "\n\n## Handles\n"
+    _HANDLES_PROVENANCE = (
+        "Read from storage at compaction time, not carried over from the summary "
+        "above — these ids are exact.\n"
+    )
+    _HANDLES_TASKS_MORE = "  … and {n} more — call `tasks(action='list')` for the full list.\n"
+    _HANDLES_CHILDREN_MORE = "  … and {n} more — call `list_workstreams` for the full list.\n"
+
+    def _coordinator_handle_rows(self) -> tuple[list[str], list[str]]:
+        """Render this coordinator's handles as ``(task_lines, child_lines)``.
+
+        ``([], [])`` whenever there can be no handles — an interactive session
+        (no task envelope, no children), a coordinator with no coord client
+        (eval / rehydration shells), or both reads coming back empty.
+
+        Both reads are same-process storage reads on the worker thread that is
+        already running the compaction (``tasks_get`` decodes the workstream's
+        config row; ``list_children`` is a single ``list_workstreams`` query), so
+        this costs no round-trip and no lock the tool path doesn't already take.
+
+        NEVER raises.  A failed read costs this block's precision, never the
+        compaction: the summary is correct without it, and trading a whole
+        history swap for a side read would be the worse failure by far.  Same
+        isolation the idle observer's children probe keeps.
+
+        Sanitiser ruling, one oracle for the whole block — ``sanitize_display``:
+
+        * Model-authored text (titles, notes, child names) is sanitised.  The
+          control class is what this render actually needs: these are single-line
+          list rows, and a newline inside a title would forge a sibling row the
+          model then reads as a real task.  That class is identical in both
+          sanitisers, so nothing about the structural guarantee turns on the
+          choice.  The guarantee is exactly that — STRUCTURAL, one row per real
+          handle.  A title can still mention an id-shaped string inline, which is
+          left alone because it fails safe: an id no envelope holds resolves to
+          "not found", never to another row.
+        * Angle brackets are KEPT, which is the whole difference from
+          ``sanitize_name``.  Deleting them silently rewrites ordinary planning
+          text ("cut p99 to <200ms" → "cut p99 to 200ms"), inverting a constraint
+          the coordinator is working to — and it would buy nothing here: these
+          titles are the coordinator's own, and already reach this same model
+          verbatim through its own ``tasks(action='list')`` results on this same
+          assistant channel.  The nudge bodies delete brackets because they are
+          interpolated into a SYSTEM turn; this block is not.
+        * Ids take the sanitiser as an ALTERATION CHECK rather than a filter: a
+          row whose id would be altered is DROPPED, never mangled, because a
+          mangled id renders a handle that cannot resolve — worse than an absent
+          one.  Task ids are server-minted (``tsk_`` + token_hex) but read back
+          out of a JSON blob a hand-edited DB can leave ragged; ws_ids are
+          primary keys and can't be ragged, but they take the same check so the
+          block has one rule rather than two.
+        """
+        if self._kind != WorkstreamKind.COORDINATOR or self._coord_client is None:
+            return [], []
+
+        def _clean(value: Any) -> str:
+            return sanitize_display(str(value or "").strip())
+
+        def _id(value: Any) -> str:
+            """The alteration check: '' for an id that isn't usable as-is."""
+            raw = str(value or "").strip()
+            return raw if raw and sanitize_display(raw) == raw else ""
+
+        task_lines: list[str] = []
+        child_lines: list[str] = []
+        try:
+            envelope = self._coord_client.tasks_get(self._ws_id)
+            for task in envelope.get("tasks", []):
+                if not isinstance(task, dict):
+                    continue
+                task_id = _id(task.get("id"))
+                if not task_id:
+                    continue
+                line = f"- `{task_id}` [{_clean(task.get('status')) or 'unknown'}] "
+                line += _clean(task.get("title"))
+                child_ws_id = _id(task.get("child_ws_id"))
+                if child_ws_id:
+                    line += f" → child `{child_ws_id}`"
+                note = _clean(task.get("note"))
+                if note:
+                    line += f" (note: {note})"
+                task_lines.append(line + "\n")
+        except Exception:
+            log.warning("compaction.handles_read_failed", source="tasks", exc_info=True)
+        try:
+            children = self._coord_client.list_children(self._ws_id).get("children", [])
+            for child in children:
+                if not isinstance(child, dict):
+                    continue
+                ws_id = _id(child.get("ws_id"))
+                if not ws_id:
+                    continue
+                state = _clean(child.get("state")) or "unknown"
+                child_lines.append(f"- `{ws_id}` [{state}] {_clean(child.get('name'))}\n")
+        except Exception:
+            log.warning("compaction.handles_read_failed", source="children", exc_info=True)
+        return task_lines, child_lines
+
+    @staticmethod
+    def _fit_handle_rows(heading: str, rows: list[str], budget: int, more: str) -> str:
+        """One handles section, fitted to ``budget`` chars — never mid-row.
+
+        Rows are whole handles, so truncation drops them at their boundary and
+        names how many went: cutting head+tail through a list the way
+        :meth:`_truncate_block` does would leave a half-copied id, which is worse
+        than an absent one (it renders a call that can't resolve).  ``more``
+        formats the count with an authoritative-source pointer, so the block is
+        never the last word on what exists.
+
+        The returned text is always ``<= budget`` (``""`` when even the heading
+        plus pointer won't fit): each kept row was admitted only after the
+        pointer covering everything still behind it was accounted for.
+        """
+        if not rows or budget <= 0:
+            return ""
+        kept: list[str] = []
+        used = len(heading)
+        for i, row in enumerate(rows):
+            behind = len(rows) - (i + 1)
+            pointer = more.format(n=behind) if behind else ""
+            if used + len(row) + len(pointer) > budget:
+                break
+            kept.append(row)
+            used += len(row)
+        dropped = len(rows) - len(kept)
+        if not kept:
+            pointer = more.format(n=dropped)
+            return heading + pointer if len(heading) + len(pointer) <= budget else ""
+        return heading + "".join(kept) + (more.format(n=dropped) if dropped else "")
+
+    def _render_handles_block(
+        self, task_lines: list[str], child_lines: list[str], budget: int
+    ) -> str:
+        """Assemble the ``## Handles`` block within ``budget`` chars.
+
+        Tasks are served first — the coordinator's own plan, and the only place
+        the id→title pairing survives at all — but children are reserved up to
+        half the room so a long task list can't starve them off the page; what
+        they don't need goes back to the tasks.  Headings carry the TOTAL count,
+        so a truncated section still tells the model how much it isn't seeing.
+        """
+        prefix = self._HANDLES_HEADING + self._HANDLES_PROVENANCE
+        spare = budget - len(prefix)
+        if spare <= 0:
+            return ""
+        child_heading = f"\nChild workstreams ({len(child_lines)}):\n"
+        child_full = len(child_heading) + sum(len(line) for line in child_lines)
+        reserved = min(child_full, spare // 2) if child_lines else 0
+        tasks = self._fit_handle_rows(
+            f"\nTasks ({len(task_lines)}):\n",
+            task_lines,
+            spare - reserved,
+            self._HANDLES_TASKS_MORE,
+        )
+        children = self._fit_handle_rows(
+            child_heading, child_lines, spare - len(tasks), self._HANDLES_CHILDREN_MORE
+        )
+        if not tasks and not children:
+            return ""
+        return prefix + tasks + children
+
     def _compact_messages(
         self,
         auto: bool = False,
@@ -8377,6 +8570,12 @@ class ChatSession:
         record, and the plan must cross a compaction copied, not paraphrased —
         the summarizer also reads the spill, but its paraphrase must not be
         the only survivor.
+
+        A coordinator additionally gets a ``## Handles`` block appended by the
+        same shell concatenation — its task and child-workstream ids paired with
+        what they refer to, read from storage here rather than transcribed by the
+        summarizer (:meth:`_coordinator_handle_rows`).  Interactive sessions have
+        no handles and get nothing.
         """
         # Presentation label derived from the one semantic flag — deriving
         # locally makes an auto=True/trigger="manual" drift impossible.
@@ -8470,11 +8669,25 @@ class ChatSession:
             spill = to_summarize[-1]
             if spill.role is Role.ASSISTANT:
                 spill_text = (spill.text or "").strip()
-        carries = (1 if spill_text else 0) + (1 if last_user_content else 0)
+
+        # The coordinator's handles are the third carry.  They land in the same
+        # post-compaction prompt as the other two, so they take a share of the
+        # same budget rather than a private one — a block sized outside that
+        # split is the same stacking the spill/hint pair was sized to prevent.
+        # Read BEFORE the count so the count and the render see one answer: a
+        # block that exists but wasn't counted is precisely the under-count the
+        # shared budget exists to make impossible.
+        task_lines, child_lines = self._coordinator_handle_rows()
+        handles = bool(task_lines or child_lines)
+        carries = (1 if spill_text else 0) + (1 if last_user_content else 0) + (1 if handles else 0)
         carry_budget = self._carry_budget_chars(carries) if carries else 0
 
-        # Wind-down first, then how to resume — the summary reads: sections,
+        # Handles first (state the harness knows exactly), then wind-down, then
+        # how to resume — the summary reads: sections, the ids still in play,
         # what the model recorded, then the ask to continue from.
+        if handles:
+            summary += self._render_handles_block(task_lines, child_lines, carry_budget)
+
         carry_truncated = False
         if spill_text:
             carry_truncated = len(spill_text) > carry_budget