fix(tui): preserve reasoning and cancellation boundaries

Author: elkaix

src/pythinker_code/ui/shell/visualize/_blocks.py

@@ -608,6 +608,7 @@ def _tail_lines(text: str, n: int) -> str:
 class _ThinkingSegment:
     summary_index: int | None
     text: str = ""
+    encrypted_boundary: bool = False
 
 
 class _StyleOverrideRenderable:
@@ -762,11 +763,21 @@ def render_expanded(self) -> RenderableType:
         finally:
             self._report_update.set_expanded(was_expanded)
 
-    def append(self, content: str, *, summary_index: int | None = None) -> None:
+    def append(
+        self,
+        content: str,
+        *,
+        summary_index: int | None = None,
+        encrypted: str | None = None,
+    ) -> None:
         self.raw_text += content
         self._token_count += _estimate_tokens(content)
-        if self.is_think and content:
-            self._append_thinking_segment(content, summary_index=summary_index)
+        if self.is_think and (content or encrypted):
+            self._append_thinking_segment(
+                content,
+                summary_index=summary_index,
+                encrypted=encrypted,
+            )
         self._invalidate_preview_cache()
         if self._paced:
             # Reveal is paced by reveal_tick() for smooth streaming; just buffer
@@ -970,11 +981,30 @@ def take_committed_renderables(self) -> list[RenderableType]:
 
     # -- Private -------------------------------------------------------------
 
-    def _append_thinking_segment(self, content: str, *, summary_index: int | None) -> None:
-        if self._thinking_segments and self._thinking_segments[-1].summary_index == summary_index:
-            self._thinking_segments[-1].text += content
+    def _append_thinking_segment(
+        self,
+        content: str,
+        *,
+        summary_index: int | None,
+        encrypted: str | None,
+    ) -> None:
+        encrypted_boundary = bool(encrypted)
+        if (
+            self._thinking_segments
+            and self._thinking_segments[-1].summary_index == summary_index
+            and not self._thinking_segments[-1].encrypted_boundary
+        ):
+            segment = self._thinking_segments[-1]
+            segment.text += content
+            segment.encrypted_boundary = encrypted_boundary
             return
-        self._thinking_segments.append(_ThinkingSegment(summary_index=summary_index, text=content))
+        self._thinking_segments.append(
+            _ThinkingSegment(
+                summary_index=summary_index,
+                text=content,
+                encrypted_boundary=encrypted_boundary,
+            )
+        )
 
     def _pending_text(self) -> str:
         return self.raw_text[self._committed_len : self._revealed_len]

src/pythinker_code/ui/shell/visualize/_live_view.py

@@ -1014,6 +1014,9 @@ def _working_indicator(self, *, hide_tips: bool = False) -> RenderableType:
                 elapsed_s=elapsed,
                 tokens=get_turn_output_tokens(),
                 token_rate=self._turn_token_rate(now),
+                interrupt_hint=(
+                    "esc to interrupt" if getattr(self, "_cancel_event", None) is not None else ""
+                ),
             ),
             width=width,
         )
@@ -1695,13 +1698,14 @@ def flush_notifications(self) -> None:
 
     def append_content(self, part: ContentPart) -> None:
         match part:
-            case ThinkPart(think=text, summary_index=summary_index):
+            case ThinkPart(think=text, encrypted=encrypted, summary_index=summary_index):
                 is_think = True
             case TextPart(text=text):
                 if not text:
                     return
                 is_think = False
                 summary_index = None
+                encrypted = None
             case ImageURLPart():
                 self._append_content_label("[image]")
                 return
@@ -1739,8 +1743,12 @@ def append_content(self, part: ContentPart) -> None:
                 paced=self._stream_pacing,
             )
             self.refresh_soon()
-        if text:
-            self._current_content_block.append(text, summary_index=summary_index)
+        if text or encrypted:
+            self._current_content_block.append(
+                text,
+                summary_index=summary_index,
+                encrypted=encrypted,
+            )
             self.refresh_soon()
 
     def _append_content_label(self, label: str, *, unknown: bool = False) -> None:

tests/e2e/test_shell_pty_e2e.py

@@ -927,7 +927,7 @@ def test_shell_cancel_running_command_kills_process_and_recovers(tmp_path: Path)
     scripts = [
         build_shell_tool_call(
             "tc-c1",
-            "printf started > cancel_started.txt && sleep 30 && "
+            "printf '%s' \"$$\" > cancel_pgid.txt && sleep 30 && "
             "printf should-not-exist > cancel_output.txt",
         ),
         "text: Cancel recovery completed.",
@@ -948,19 +948,14 @@ def test_shell_cancel_running_command_kills_process_and_recovers(tmp_path: Path)
 
         cancel_mark = shell.mark()
         shell.send_line("start cancellable command")
-        started_path = work_dir / "cancel_started.txt"
+        pgid_path = work_dir / "cancel_pgid.txt"
         started_deadline = time.monotonic() + 10.0
-        while not started_path.exists():
+        while not pgid_path.exists():
             if time.monotonic() >= started_deadline:
                 raise AssertionError("Timed out waiting for cancellable command to start.")
             shell.read_available(timeout=0.05)
-        # The child can begin while prompt_toolkit is still switching from the
-        # submitted prompt to the running-turn delegate that owns Escape. Keep
-        # the command alive well beyond this short stabilization window so the
-        # key cannot land in the transition and be discarded.
-        stabilization_deadline = time.monotonic() + 1.0
-        while time.monotonic() < stabilization_deadline:
-            shell.read_available(timeout=0.05)
+        command_pgid = int(pgid_path.read_text(encoding="utf-8"))
+        shell.read_until_contains("esc)", after=cancel_mark, timeout=10.0)
         shell.send_key("escape")
         # The "Interrupted by user" acknowledgement only prints after the soul
         # re-raises the cancellation, which first awaits a shielded, disk-first
@@ -973,7 +968,8 @@ def test_shell_cancel_running_command_kills_process_and_recovers(tmp_path: Path)
         cancel_prompt_mark = shell.mark()
         _read_until_prompt(shell, after=cancel_prompt_mark)
 
-        time.sleep(5.3)
+        with pytest.raises(ProcessLookupError):
+            os.killpg(command_pgid, 0)
         assert not (work_dir / "cancel_output.txt").exists()
 
         recovery_mark = shell.mark()

tests/ui_and_conv/test_live_content_parts.py

@@ -92,6 +92,24 @@ def test_live_view_dispatch_preserves_reasoning_summary_boundaries_and_style(
         assert all(style.italic for style in styles)
 
 
+def test_live_view_preserves_encrypted_only_reasoning_boundary(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    emitted = _capture_scrollback(monkeypatch)
+    view = _LiveView(StatusUpdate(context_tokens=1000), show_thinking_stream=True)
+
+    view.dispatch_wire_message(ThinkPart(think="**Planning**", summary_index=0))
+    view.dispatch_wire_message(ThinkPart(think="", encrypted="signature", summary_index=0))
+    view.dispatch_wire_message(ThinkPart(think="**Executing**", summary_index=0))
+    view.flush_content()
+
+    output = _render(emitted)
+    lines = [line for line in output.splitlines() if line.strip()]
+    assert len(lines) == 2
+    assert "Planning" in lines[0]
+    assert "Executing" in lines[1]
+
+
 def test_text_media_text_flushes_at_stable_boundaries(
     monkeypatch: pytest.MonkeyPatch,
 ) -> None:

tests/ui_and_conv/test_live_view_notifications.py

@@ -1,5 +1,7 @@
 from __future__ import annotations
 
+import asyncio
+
 from pythinker_core.message import ToolCall
 from pythinker_core.tooling import ToolResult, ToolReturnValue
 from rich.console import Console, Group
@@ -158,6 +160,15 @@ def test_working_indicator_uses_turn_elapsed_time(monkeypatch):
     assert "4h" not in rendered
 
 
+def test_working_indicator_exposes_escape_when_turn_is_cancellable():
+    view = _LiveView(StatusUpdate(), asyncio.Event())
+    view.dispatch_wire_message(TurnBegin(user_input="scan"))
+
+    rendered = _render(view._working_indicator())
+
+    assert "esc)" in rendered
+
+
 def test_working_indicator_uses_rotating_thinking_words(monkeypatch):
     now = 90.0
     monkeypatch.setattr(live_view_module.time, "monotonic", lambda: now)

tests/ui_and_conv/test_subagent_live_stream.py

@@ -680,6 +680,7 @@ def test_run_agents_background_result_keeps_one_semantic_live_tree_after_nested_
     assert "searching…" in output
     assert "2 queued" in output
     assert "agent-alpha-raw-id" not in output
+    assert "agent-beta-raw-id" not in output
     for line in output.splitlines():
         assert cell_width(line) <= 80
 

tests/ui_and_conv/test_tool_call_block.py

@@ -523,9 +523,8 @@ def test_run_agents_background_launch_keeps_live_agent_activity():
     assert "Run Agents completed" not in output
 
 
-def test_run_agents_background_launch_keeps_live_agent_activity_in_card_style(
-    _card_style_with_builtin_renderers,
-):
+@pytest.mark.usefixtures("_card_style_with_builtin_renderers")
+def test_run_agents_background_launch_keeps_live_agent_activity_in_card_style():
     block = _ToolCallBlock(
         _tool_call(
             "RunAgents",
@@ -551,6 +550,7 @@ def test_run_agents_background_launch_keeps_live_agent_activity_in_card_style(
     output = _plain(block.compose())
 
     assert "waiting Explore Audit the renderer" in output
+    assert "Run Agents completed" not in output
 
 
 def test_run_agents_foreground_completion_is_not_background_pending():