feat(tui): improve reasoning and agent activity rendering (#231)

* fix(core): preserve reasoning summary indices

* test(core): cover completed reasoning summaries

* fix(tui): render reasoning summary boundaries

* feat(tui): render agent-centric activity tree

* fix(tui): harden RunAgents activity rendering

* fix(tui): prioritize active agent rows

* fix(tui): avoid reset ghosts on posix resize

* test(tui): tighten pty resize regressions

* fix(core): retain encrypted reasoning boundaries

* fix(tui): normalize agent activity labels

* feat(tui): extend agent activity tree to single Agent calls

Route single `Agent` tool calls through the same payload-free semantic
activity tree previously used only for `RunAgents`, so a lone subagent
renders per-agent status rows (waiting/running/thinking + reading…/
searching…/running command…) instead of raw nested tool calls, args, or
streamed output. Derive the running description from the tool `description`
argument, suppress stale pre-result activity once the parent returns, and
render the completed result via the dedicated result renderer.

Update the block, nested-lifecycle, integration, and card tests to the new
semantic contract (raw payloads never leak; first-result-wins dedup and
ancestry roll-up preserved), and fix the stale `RunAgents(`->`Agents(`
label assertion in the action-spacer test.

* fix(core): keep encrypted reasoning on summary-less responses

Address CodeRabbit review findings on the branch:

- core: a non-streaming OpenAI Responses reasoning item can return
  encrypted_content with an empty summary; the conversion loop dropped it
  entirely, diverging from the streaming output_item.done path and making
  the reasoning boundary non-replayable. Emit an encrypted ThinkPart in
  that case, mirroring the streaming behavior, with focused coverage.
- tui: render "1 agent" (singular) instead of "1 agents" in the grouped
  agents summary, and make the uniform-status check explicitly boolean.

The two other flagged reasoning findings (summary-index preservation and
cross-output merge collisions) were verified against the code and tests
and are non-issues: real deltas always carry valid indices, and encrypted
boundaries already block adjacent cross-output merges.

* fix(tui): annotate agent activity mutators

* fix(tui): keep background agent activity live

* test(tui): synchronize shell cancellation e2e

* fix(tui): preserve pending agent activity in cards

* fix(tui): preserve reasoning and cancellation boundaries

Author: elkaix

CHANGELOG.md

@@ -15,6 +15,8 @@ GitHub Releases page; `0.8.0` is the new starting line.
 
 ## Unreleased
 
+- Fixed reasoning summaries exposing Markdown delimiters and duplicate terminal rows after refocus, and redesigned agent progress (single `Agent` calls and parallel `RunAgents` fan-outs) as a compact, payload-free agent activity tree.
+- Retain a non-streaming OpenAI Responses reasoning item's encrypted content when it carries no summary, matching the streaming path so the reasoning boundary stays replayable.
 - Reduce the shell prompt session to a compatibility façade over deep `prompting/` modules (config, keybindings, completion menus, narrow shell-facing methods), with Unicode cell-width coverage and documented module ownership in the architecture guide.
 - Render image/audio/video and unknown content parts as payload-free labels in the live view, roll nested subagent activity (up to 16 levels) under the correct root tool card, and stop provider-remapped tool results from starting replay user turns.
 - **Behavior change:** `!` shell commands now run through the detected configured shell (`<shell> -c`, PowerShell `-command` on Windows) instead of the implicit `/bin/sh`/`cmd.exe`, with separate 1 MiB stdout/stderr caps and cancellation cleanup; existing `cmd.exe`-syntax commands on Windows may need updating.

packages/pythinker-core/src/pythinker_core/contrib/chat_provider/openai_responses.py

@@ -475,6 +475,18 @@ def _responses_finish_reason(response: Response) -> str | None:
     return response.status
 
 
+def _reasoning_summary_index(value: object) -> int | None:
+    if isinstance(value, bool) or not isinstance(value, int) or value < 0:
+        return None
+    return value
+
+
+def _responses_output_index(value: object) -> int | None:
+    if isinstance(value, bool) or not isinstance(value, int) or value < 0:
+        return None
+    return value
+
+
 class OpenAIResponsesStreamedMessage:
     def __init__(self, response: Response | AsyncStream[ResponseStreamEvent]):
         if isinstance(response, Response):
@@ -535,16 +547,30 @@ async def _convert_non_stream_response(
                     ),
                 )
             elif item.type == "reasoning":
-                for summary in item.summary:
+                encrypted_content = getattr(item, "encrypted_content", None)
+                emitted_summary = False
+                for summary_index, summary in enumerate(getattr(item, "summary", ())):
+                    emitted_summary = True
                     yield ThinkPart(
                         think=summary.text,
-                        encrypted=item.encrypted_content,
+                        encrypted=encrypted_content,
+                        summary_index=summary_index,
+                    )
+                if not emitted_summary and encrypted_content is not None:
+                    # Mirror the streaming `output_item.done` path: a reasoning item
+                    # can carry encrypted_content with no summary parts, and dropping
+                    # it here would make the reasoning boundary non-replayable.
+                    yield ThinkPart(
+                        think="",
+                        encrypted=encrypted_content,
+                        summary_index=None,
                     )
 
     async def _convert_stream_response(
         self, response: AsyncStream[ResponseStreamEvent]
     ) -> AsyncIterator[StreamedMessagePart]:
         """Convert streaming Responses events into message parts."""
+        reasoning_summary_indices_by_output: dict[int, int | None] = {}
         try:
             async for chunk in response:
                 if isinstance(chunk, ResponseCreatedEvent):
@@ -565,16 +591,40 @@ async def _convert_stream_response(
                 elif chunk.type == "response.output_item.done":
                     item = chunk.item
                     if item.type == "reasoning":
-                        yield ThinkPart(think="", encrypted=item.encrypted_content)
+                        output_index = _responses_output_index(getattr(chunk, "output_index", None))
+                        summary_index = (
+                            reasoning_summary_indices_by_output.pop(output_index, None)
+                            if output_index is not None
+                            else None
+                        )
+                        yield ThinkPart(
+                            think="",
+                            encrypted=item.encrypted_content,
+                            summary_index=summary_index,
+                        )
                 elif isinstance(chunk, ResponseFunctionCallArgumentsDeltaEvent):
                     yield ToolCallPart(
                         arguments_part=chunk.delta,
                         stream_index=chunk.output_index,
                     )
                 elif chunk.type == "response.reasoning_summary_part.added":
-                    yield ThinkPart(think="")
+                    summary_index = _reasoning_summary_index(getattr(chunk, "summary_index", None))
+                    output_index = _responses_output_index(getattr(chunk, "output_index", None))
+                    if output_index is not None:
+                        reasoning_summary_indices_by_output[output_index] = summary_index
+                    yield ThinkPart(
+                        think="",
+                        summary_index=summary_index,
+                    )
                 elif chunk.type == "response.reasoning_summary_text.delta":
-                    yield ThinkPart(think=chunk.delta)
+                    summary_index = _reasoning_summary_index(getattr(chunk, "summary_index", None))
+                    output_index = _responses_output_index(getattr(chunk, "output_index", None))
+                    if output_index is not None:
+                        reasoning_summary_indices_by_output[output_index] = summary_index
+                    yield ThinkPart(
+                        think=getattr(chunk, "delta", ""),
+                        summary_index=summary_index,
+                    )
                 elif isinstance(chunk, ResponseErrorEvent):
                     self._finish_reason = "failed"
                     return

packages/pythinker-core/src/pythinker_core/message.py

@@ -98,11 +98,14 @@ class ThinkPart(ContentPart):
     think: str
     encrypted: str | None = None
     """Encrypted thinking content, or signature."""
+    summary_index: int | None = Field(default=None, exclude_if=lambda value: value is None)
 
     @override
     def merge_in_place(self, other: Any) -> bool:
         if not isinstance(other, ThinkPart):
             return False
+        if self.summary_index != other.summary_index:
+            return False
         if self.encrypted:
             return False
         self.think += other.think

packages/pythinker-core/tests/test_message.py

@@ -301,6 +301,21 @@ def test_message_with_empty_list_content():
     )
 
 
+def test_think_part_summary_index_round_trips_through_pydantic() -> None:
+    part = ThinkPart(think="Plan", summary_index=3)
+    dumped = part.model_dump()
+
+    assert dumped == snapshot(
+        {
+            "type": "think",
+            "think": "Plan",
+            "encrypted": None,
+            "summary_index": 3,
+        }
+    )
+    assert ThinkPart.model_validate(dumped) == part
+
+
 def test_message_extract_text():
     message = Message(
         role="user",

packages/pythinker-core/tests/test_stream_message_assembler.py

@@ -1,7 +1,7 @@
 import pytest
 
 from pythinker_core.chat_provider import APIStreamProtocolError
-from pythinker_core.message import TextPart, ToolCall, ToolCallPart
+from pythinker_core.message import TextPart, ThinkPart, ToolCall, ToolCallPart
 from pythinker_core.stream_message_assembler import StreamMessageAssembler
 
 
@@ -364,6 +364,71 @@ def test_terminal_failure_reasons_are_rejected(finish_reason: str, category: str
     assert "private" not in str(caught.value)
 
 
+def test_same_summary_index_merges_adjacent_think_parts() -> None:
+    assembler = StreamMessageAssembler()
+    assembler.add(ThinkPart(think="Plan", summary_index=0))
+    assembler.add(ThinkPart(think=" next", summary_index=0))
+    assembler.add(ThinkPart(think="Check", summary_index=1))
+
+    message = assembler.finish(response_id=None, finish_reason="completed")
+
+    assert message.content == [
+        ThinkPart(think="Plan next", summary_index=0),
+        ThinkPart(think="Check", summary_index=1),
+    ]
+
+
+def test_different_summary_indices_stay_separate() -> None:
+    assembler = StreamMessageAssembler()
+    assembler.add(ThinkPart(think="Plan", summary_index=0))
+    assembler.add(ThinkPart(think="Check", summary_index=1))
+
+    message = assembler.finish(response_id=None, finish_reason="completed")
+
+    assert message.content == [
+        ThinkPart(think="Plan", summary_index=0),
+        ThinkPart(think="Check", summary_index=1),
+    ]
+
+
+def test_non_adjacent_duplicate_summary_indices_stay_separate() -> None:
+    assembler = StreamMessageAssembler()
+    assembler.add(ThinkPart(think="Plan", summary_index=0))
+    assembler.add(ThinkPart(think="Check", summary_index=1))
+    assembler.add(ThinkPart(think="Finish", summary_index=0))
+
+    message = assembler.finish(response_id=None, finish_reason="completed")
+
+    assert message.content == [
+        ThinkPart(think="Plan", summary_index=0),
+        ThinkPart(think="Check", summary_index=1),
+        ThinkPart(think="Finish", summary_index=0),
+    ]
+
+
+def test_unindexed_legacy_think_parts_still_merge() -> None:
+    assembler = StreamMessageAssembler()
+    assembler.add(ThinkPart(think="Plan"))
+    assembler.add(ThinkPart(think=" next"))
+
+    message = assembler.finish(response_id=None, finish_reason="completed")
+
+    assert message.content == [ThinkPart(think="Plan next")]
+
+
+def test_indexed_and_unindexed_think_parts_do_not_merge() -> None:
+    assembler = StreamMessageAssembler()
+    assembler.add(ThinkPart(think="Plan", summary_index=0))
+    assembler.add(ThinkPart(think=" legacy"))
+
+    message = assembler.finish(response_id=None, finish_reason="completed")
+
+    assert message.content == [
+        ThinkPart(think="Plan", summary_index=0),
+        ThinkPart(think=" legacy"),
+    ]
+
+
 def test_content_parts_assemble_independently_from_tool_calls() -> None:
     assembler = StreamMessageAssembler()
     assembler.add(TextPart(text="hello "))