fix(agent): keep interrupt-checkpoint scaffolding out of the steered transcript

A mid-stream steer persists an interrupted-turn checkpoint so the model knows
its reply was cut off. That scaffolding — "[This response was interrupted by a
user correction.]" and the "Visible response before the interruption:" header —
was written straight into message content, so every reload painted the raw
machinery as an assistant bubble (and merged it into the preceding tool-call
bubble). Steered transcripts became unreadable.

Reuse the existing display/replay split instead of inventing new surface:
- Carry the scaffolded form in the server-only api_content sidecar (the exact
  bytes replayed to the provider), keep content the user's/agent's real words.
- When nothing reached the screen there is no clean form, so mark the row
  display_kind=hidden — replayed to the model, dropped by every transcript
  surface, exactly like compaction-reference rows.
- Honor display_kind=hidden in the gateway's _history_to_messages projection
  (it only sniffed the [System: convention), so the checkpoint can't leak
  through the live/resume path to the TUI/CLI either.

The model still receives the full interrupted context on the wire; the
transcript shows the partial reply and the user's correction.

agent/conversation_loop.py

@@ -140,6 +140,19 @@ def _apply_active_turn_redirect(agent: Any, messages: List[Dict[str, Any]], text
     incomplete by definition; the model regenerates it on the retried turn.
     If a future path needs to preserve interrupted thinking, carry it in a
     provider-gated reasoning *field*, never in content.
+    INVARIANT — the scaffolding is provider-replay text, not transcript text.
+    ``[This response was interrupted by a user correction.]`` and its
+    ``Visible response before the interruption:`` header exist so the MODEL
+    understands its own reply was cut off. They are not prose the user wrote
+    or the agent said. Persisting them into ``content`` painted the raw
+    machinery as an assistant bubble on every reload (and merged it into the
+    preceding tool-call bubble), which is what made a steered transcript
+    unreadable. Carry the scaffolded form in the ``api_content`` sidecar --
+    the exact bytes replayed to the provider -- and keep ``content`` clean.
+    When nothing was on screen there is no clean form at all, so the row is
+    marked ``display_kind="hidden"``: still replayed to the model, dropped by
+    every transcript surface (desktop, TUI, CLI resume), exactly like the
+    compaction-reference rows.
     """
     visible = agent._strip_think_blocks(
         getattr(agent, "_current_streamed_assistant_text", "") or ""
@@ -162,9 +175,22 @@ def _apply_active_turn_redirect(agent: Any, messages: List[Dict[str, Any]], text
             f"{checkpoint}\n\n"
             f"{text}"
         )
-        messages.append({"role": "user", "content": correction})
+        # Transcript shows the user's own words; the provider replays the
+        # scaffolded form so it still sees the interrupted context.
+        messages.append(
+            {"role": "user", "content": text, "api_content": correction}
+        )
     else:
-        messages.append({"role": "assistant", "content": checkpoint})
+        entry: Dict[str, Any] = {
+            "role": "assistant",
+            "content": visible or checkpoint,
+            "api_content": checkpoint,
+        }
+        if not visible:
+            # Nothing reached the screen — this row carries no assistant prose
+            # at all, only the cut-off notice for the model.
+            entry["display_kind"] = "hidden"
+        messages.append(entry)
         messages.append({"role": "user", "content": text})
 
     agent._current_streamed_assistant_text = ""

apps/desktop/src/lib/chat-messages.test.ts

@@ -233,6 +233,45 @@ describe('toChatMessages', () => {
     expect(chatMessageText(message)).toBe('summarize @file:`src/main.ts` for me')
   })
 
+  it('never paints redirect scaffolding as an assistant bubble', () => {
+    // What the desktop actually receives after a mid-stream steer: the runtime
+    // keeps the interrupt scaffolding in a server-only api_content sidecar
+    // (never shipped to the client) so content is already clean, and marks a
+    // prose-free checkpoint display_kind:'hidden'. The transcript must show the
+    // partial reply and the user's correction — never
+    // "[This response was interrupted by a user correction.]".
+    const messages = toChatMessages([
+      { role: 'user', content: 'go', timestamp: 1 },
+      {
+        role: 'assistant',
+        content: 'Hey. I was mid-Figma MCP fix when we paused.',
+        timestamp: 2
+      },
+      { role: 'user', content: 'i love you', timestamp: 3 },
+      {
+        // Nothing had reached the screen — checkpoint exists only for the model.
+        role: 'assistant',
+        content: '[This response was interrupted by a user correction.]',
+        display_kind: 'hidden',
+        timestamp: 4
+      },
+      { role: 'user', content: 'keep going', timestamp: 5 }
+    ])
+
+    expect(messages.map(chatMessageText)).toEqual([
+      'go',
+      'Hey. I was mid-Figma MCP fix when we paused.',
+      'i love you',
+      'keep going'
+    ])
+
+    for (const message of messages) {
+      expect(chatMessageText(message)).not.toContain('This response was interrupted')
+      expect(chatMessageText(message)).not.toContain('Visible response before the interruption')
+      expect(chatMessageText(message)).not.toContain('Context from the interrupted assistant response')
+    }
+  })
+
   it('projects durable timeline kinds without inspecting their text', () => {
     const messages = toChatMessages([
       { role: 'user', content: 'real user turn', timestamp: 1 },

tests/run_agent/test_steer.py

@@ -242,10 +242,68 @@ class TestActiveTurnRedirectCheckpoint:
 
         assert [m["role"] for m in messages] == ["user", "assistant", "user"]
         assert messages[-1]["role"] == "user"
-        assert messages[-1]["content"].endswith("Use Postgres instead.")
+        assert messages[-1]["content"] == "Use Postgres instead."
         assert sum(1 for m in messages if m["role"] == "assistant") == 1
-        assert "Visible draft." in messages[-1]["content"]
-        assert "Context from the interrupted assistant response" in messages[-1]["content"]
+        # Scaffolding is provider-replay text, carried in the sidecar so the
+        # model still sees the interrupted context — never in the transcript.
+        replayed = messages[-1]["api_content"]
+        assert "Visible draft." in replayed
+        assert "Context from the interrupted assistant response" in replayed
+        assert replayed.endswith("Use Postgres instead.")
+
+    def test_scaffolding_never_lands_in_transcript_content(self):
+        """The checkpoint machinery is for the MODEL, not the transcript.
+
+        Persisting ``[This response was interrupted by a user correction.]``
+        into ``content`` painted raw scaffolding as an assistant bubble on
+        every reload. It must ride in ``api_content`` (replayed to the
+        provider) while ``content`` stays clean, or be marked
+        ``display_kind="hidden"`` when there is no clean form at all.
+        """
+        from agent.conversation_loop import _apply_active_turn_redirect
+
+        scaffolding = (
+            "[This response was interrupted by a user correction.]",
+            "Visible response before the interruption:",
+            "[Context from the interrupted assistant response]",
+        )
+
+        for tail_role in ("tool", "assistant"):
+            for streamed in ("Partial reply on screen.", ""):
+                agent = _bare_agent()
+                agent._current_streamed_assistant_text = streamed
+                messages = [{"role": "user", "content": "start"}]
+                if tail_role == "assistant":
+                    messages.append({"role": "assistant", "content": "committed"})
+                else:
+                    messages.append(
+                        {"role": "assistant", "tool_calls": [{"id": "a"}]}
+                    )
+                    messages.append(
+                        {"role": "tool", "content": "out", "tool_call_id": "a"}
+                    )
+
+                _apply_active_turn_redirect(agent, messages, "New direction.")
+
+                for msg in messages:
+                    if msg.get("display_kind") == "hidden":
+                        continue  # dropped by every transcript surface
+                    content = str(msg.get("content", ""))
+                    for marker in scaffolding:
+                        assert marker not in content, (
+                            f"scaffolding leaked into visible content "
+                            f"(tail={tail_role}, streamed={bool(streamed)}): {content!r}"
+                        )
+
+                # The user's correction is always shown verbatim.
+                assert messages[-1]["content"] == "New direction."
+                # ...and the model still receives the interrupted context.
+                replayed = "".join(
+                    str(m.get("api_content") or m.get("content", "")) for m in messages
+                )
+                assert "[This response was interrupted by a user correction.]" in replayed
+                if streamed:
+                    assert streamed in replayed
 
     def test_checkpoint_never_replays_chain_of_thought(self):
         """Raw CoT serialized into checkpoint content reads to Anthropic's
@@ -268,7 +326,12 @@ class TestActiveTurnRedirectCheckpoint:
 
             _apply_active_turn_redirect(agent, messages, "Change course.")
 
-            serialized = "".join(str(m.get("content", "")) for m in messages)
+            # Check BOTH the transcript content and the replayed sidecar —
+            # the sidecar is what actually reaches the provider.
+            serialized = "".join(
+                str(m.get("content", "")) + str(m.get("api_content") or "")
+                for m in messages
+            )
             assert "SECRET chain of thought." not in serialized
             assert "Reasoning shown before the interruption" not in serialized
             assert "Visible draft." in serialized
@@ -282,8 +345,14 @@ class TestActiveTurnRedirectCheckpoint:
 
         _apply_active_turn_redirect(agent, messages, "New direction.")
 
-        checkpoint = messages[-2]["content"]
-        assert checkpoint == "[This response was interrupted by a user correction.]"
+        checkpoint_row = messages[-2]
+        # Nothing was on screen, so the row exists only for the model: hidden
+        # from every transcript surface, scaffolding replayed via the sidecar.
+        assert checkpoint_row["display_kind"] == "hidden"
+        assert (
+            checkpoint_row["api_content"]
+            == "[This response was interrupted by a user correction.]"
+        )
         assert messages[-1]["content"] == "New direction."
 
 

tests/test_tui_gateway_server.py

@@ -1714,6 +1714,45 @@ def test_history_to_messages_hides_gateway_system_markers():
     ]
 
 
+def test_history_to_messages_drops_display_hidden_scaffolding():
+    # A mid-stream steer persists an interrupted-turn checkpoint. When nothing
+    # reached the screen the row carries only model-facing scaffolding and is
+    # marked display_kind="hidden"; the scaffolded bytes live in the server-only
+    # api_content sidecar for provider replay. This projection -- the single
+    # display source every client reads -- must drop the row by its declared
+    # display_kind, not just the "[System:" string convention, or the raw
+    # "[This response was interrupted by a user correction.]" paints as an
+    # assistant bubble (and api_content must never ship to a client).
+    history = [
+        {"role": "user", "content": "go"},
+        {
+            "role": "assistant",
+            "content": "[This response was interrupted by a user correction.]",
+            "api_content": "[This response was interrupted by a user correction.]",
+            "display_kind": "hidden",
+        },
+        {"role": "user", "content": "i love you"},
+        {
+            "role": "assistant",
+            "content": "Love you too",
+            "api_content": (
+                "[This response was interrupted by a user correction.]\n\n"
+                "Visible response before the interruption:\n\nLove you too"
+            ),
+        },
+    ]
+
+    projected = server._history_to_messages(history)
+
+    assert projected == [
+        {"role": "user", "text": "go"},
+        {"role": "user", "text": "i love you"},
+        {"role": "assistant", "text": "Love you too"},
+    ]
+    # Server-only sidecar never crosses the wire.
+    assert all("api_content" not in m for m in projected)
+
+
 def test_history_to_messages_keeps_real_user_bracket_text():
     # Only role=user rows whose text OPENS with the [System: marker sentinel are
     # bookkeeping notices. A genuine user turn that merely mentions the token is

tui_gateway/server.py

@@ -6247,6 +6247,13 @@ def _history_to_messages(history: list[dict]) -> list[dict]:
         role = m.get("role")
         if role not in {"user", "assistant", "tool", "system"}:
             continue
+        # An explicit display_kind="hidden" row is model-facing scaffolding
+        # (compaction references, interrupted-turn checkpoints). The string
+        # sniff below only catches the "[System:" convention; honor the
+        # declared field too, or scaffolding reaches every surface that reads
+        # this projection.
+        if m.get("display_kind") == "hidden":
+            continue
         content_text = _coerce_message_text(m.get("content"))
         if _is_display_hidden_marker(role, content_text):
             continue