feat: rework the TUI transcript into collapsible episodes

raven/agent/loop/main.py

@@ -1433,6 +1433,7 @@ async def _run_agent_loop(
         on_token_delta: Callable[[str], Awaitable[None]] | None = None,
         on_reasoning_delta: Callable[[str], Awaitable[None]] | None = None,
         on_tool_event: Callable[[str, dict], Awaitable[None]] | None = None,
+        on_episode_start: Callable[[int], Awaitable[None]] | None = None,
         usage_sink: dict[str, Any] | None = None,
         drain: Drain | None = None,
     ) -> tuple[str | None, list[str], list[dict], TurnOutcome]:
@@ -1483,6 +1484,11 @@ async def _run_agent_loop(
                 effective_model,
             )
 
+            # Mark the episode boundary (one per model call) so an outlet can
+            # group this call's reasoning + text + tools into a single step.
+            if on_episode_start is not None:
+                await on_episode_start(iteration - 1)
+
             # Merge any INJECT-ed user messages (BusyPolicy.INJECT) before this
             # iteration's LLM call. Media-carrying injects keep their file
             # paths in the text so nothing is silently dropped.
@@ -1601,38 +1607,49 @@ async def _run_agent_loop(
                     # second emit here would double it.
                     emit_tool_event = on_tool_event is not None and tool_call.name != "message"
                     if emit_tool_event:
+                        _tool = self.tools.get(tool_call.name)
                         await on_tool_event(
                             "start",
                             {
                                 "tool_call_id": tool_call.id,
                                 "name": tool_call.name,
                                 "arguments": tool_call.arguments,
+                                # Tool-authored call label; None -> UI derives one.
+                                "display": _tool.display_call(tool_call.arguments) if _tool else None,
                             },
                         )
                     tool_t0 = time.monotonic()
                     result = await self.tools.execute(tool_call.name, tool_call.arguments)
                     duration_ms = int((time.monotonic() - tool_t0) * 1000)
-                    result_str = str(result)
-                    preview = result_str.replace("\n", " ")[:200]
+                    # The registry already unwrapped any ToolResult: `result` is
+                    # the model-facing text, with the optional display string
+                    # riding along on it (ToolOutput). The model always gets the
+                    # model text; the UI preview prefers the display string.
+                    model_text = str(result)
+                    display_src = getattr(result, "display_text", None) or model_text
+                    # The log stays one line; the UI event keeps newlines so a
+                    # tool that reports several items (e.g. ask_user's
+                    # question -> answer pairs) renders one row each.
+                    preview = display_src[:200]
                     logger.info(
                         "Tool result: {} duration={}ms result={}",
                         tool_call.name,
                         duration_ms,
-                        preview,
+                        preview.replace("\n", " ")[:200],
                     )
                     if emit_tool_event:
                         await on_tool_event(
                             "complete",
                             {
                                 "tool_call_id": tool_call.id,
                                 "result_preview": preview,
-                                "truncated": len(result_str) > 200,
+                                "truncated": len(display_src) > 200,
                             },
                         )
-                    messages = self.context.add_tool_result(messages, tool_call.id, tool_call.name, result)
+                    messages = self.context.add_tool_result(messages, tool_call.id, tool_call.name, model_text)
                     # #1b Track consecutive same-tool deterministic failures
                     # (transient errors excluded — a retry would clear those).
-                    if _is_hard_tool_failure(result):
+                    if _is_hard_tool_failure(model_text):
                         if tool_call.name == loop_fail_tool:
                             loop_fail_streak += 1
                         else:
@@ -1872,6 +1889,7 @@ async def _process_message(
         on_token_delta: Callable[[str], Awaitable[None]] | None = None,
         on_reasoning_delta: Callable[[str], Awaitable[None]] | None = None,
         on_tool_event: Callable[[str, dict], Awaitable[None]] | None = None,
+        on_episode_start: Callable[[int], Awaitable[None]] | None = None,
         usage_sink: dict[str, Any] | None = None,
         origin: Origin | None = None,
         drain: Drain | None = None,
@@ -2109,6 +2127,7 @@ async def _extract():
             on_token_delta=on_token_delta,
             on_reasoning_delta=on_reasoning_delta,
             on_tool_event=on_tool_event,
+            on_episode_start=on_episode_start,
             usage_sink=usage_sink,
             drain=drain,
         )
@@ -2282,6 +2301,7 @@ async def run_turn(
         """
         from raven.proactive_engine.schedulers.cron.tool import CronTool
         from raven.spine.events import (
+            EpisodeStart,
             MediaOut,
             Notice,
             NoticeKind,
@@ -2329,6 +2349,9 @@ async def on_reasoning(text: str) -> None:
             if text:
                 await emit(Reasoning(content=text))
 
+        async def on_episode(index: int) -> None:
+            await emit(EpisodeStart(index=index))
+
         async def on_tool(phase: str, info: dict[str, Any]) -> None:
             if phase == "start":
                 await emit(
@@ -2337,6 +2360,7 @@ async def on_tool(phase: str, info: dict[str, Any]) -> None:
                         tool_call_id=info["tool_call_id"],
                         name=info["name"],
                         arguments=info["arguments"],
+                        display=info.get("display"),
                     )
                 )
             else:
@@ -2439,6 +2463,7 @@ async def _route_deep_research(kind: str, text: str) -> None:
                 on_token_delta=on_token if stream else None,
                 on_reasoning_delta=on_reasoning if stream else None,
                 on_tool_event=on_tool,
+                on_episode_start=on_episode if stream else None,
                 usage_sink=usage_sink,
                 origin=req.origin,
                 drain=drain,

raven/agent/tools/ask_user.py

@@ -11,7 +11,7 @@
 from contextvars import ContextVar
 from typing import Any
 
-from raven.agent.tools.base import Tool
+from raven.agent.tools.base import Tool, ToolResult
 from raven.tui_rpc.question_broker import QuestionBroker
 
 
@@ -102,7 +102,19 @@ def parameters(self) -> dict[str, Any]:
             "required": ["questions"],
         }
 
-    async def execute(self, questions: list[dict[str, Any]], **kwargs: Any) -> str:
+    def display_call(self, args: dict[str, Any]) -> str | None:
+        """Show the question itself, not the raw arguments blob. A batch keeps
+        every question visible (joined) so the row still says what was asked;
+        the UI elides whatever does not fit."""
+        questions = [str(q.get("question", "")).strip() for q in args.get("questions") or []]
+        questions = [q for q in questions if q]
+        if not questions:
+            return None
+        if len(questions) == 1:
+            return questions[0]
+        return " | ".join(questions)
+
+    async def execute(self, questions: list[dict[str, Any]], **kwargs: Any) -> "str | ToolResult":
         cid = self._cid.get()
         if not self._broker:
             return "Error: ask_user not configured (no question broker)"
@@ -111,7 +123,11 @@ async def execute(self, questions: list[dict[str, Any]], **kwargs: Any) -> str:
         if not questions:
             return "Error: ask_user requires at least one question"
 
-        results: list[str] = []
+        told: list[str] = []  # model-facing
+        # Human-facing display: one "question -> answer" line per question, so a
+        # batch shows which answer belongs to which question. The UI renders each
+        # line as its own row.
+        picks: list[str] = []
         for entry in questions:
             question = str(entry.get("question", "")).strip()
             if not question:
@@ -124,10 +140,15 @@ async def execute(self, questions: list[dict[str, Any]], **kwargs: Any) -> str:
                 choices=[str(o) for o in options],
             )
             if answer:
-                results.append(f'User answered: "{question}" -> "{answer}".')
+                told.append(f'User answered: "{question}" -> "{answer}".')
+                picks.append(f"{question} -> {answer}" if len(questions) > 1 else str(answer))
             else:
-                results.append(f'For "{question}": (user did not answer; proceed with best judgment).')
+                told.append(f'For "{question}": (user did not answer; proceed with best judgment).')
+                picks.append(f"{question} -> (no answer)" if len(questions) > 1 else "(no answer)")
 
-        if not results:
+        if not told:
             return "Error: ask_user requires at least one non-empty question"
-        return " ".join(results) + " Continue."
+        return ToolResult(
+            model_text=" ".join(told) + " Continue.",
+            display_text="\n".join(picks) if len(picks) > 1 else f"answered: {picks[0]}",
+        )

raven/agent/tools/base.py

@@ -1,9 +1,46 @@
 """Base class for agent tools."""
 
 from abc import ABC, abstractmethod
+from dataclasses import dataclass
 from typing import Any
 
 
+@dataclass
+class ToolResult:
+    """A tool's output split into the model-facing text and an optional
+    human-facing display string.
+
+    ``execute`` may return a bare ``str`` (model text only — the UI falls back
+    to a generic preview of it) or this, when the tool wants a cleaner
+    transcript rendering than what it feeds the model. ``display_text`` must be
+    built from the tool's own execution data, not by re-parsing ``model_text``.
+    """
+
+    model_text: str
+    display_text: str | None = None
+
+
+class ToolOutput(str):
+    """What :meth:`ToolRegistry.execute` hands back: the model-facing text with
+    the optional display string attached.
+
+    A ``str`` subclass on purpose. Every caller of the registry boundary --
+    the sentinel action executor, the subagent manager, the context curator,
+    tracing -- puts the return value straight into a message, a preview or an
+    artifact, so the boundary has to return something that *is* a str; handing
+    them a :class:`ToolResult` would format its repr into model context and
+    user-facing replies. The agent loop reads ``display_text`` off it to render
+    the transcript row.
+    """
+
+    display_text: str | None
+
+    def __new__(cls, model_text: str, display_text: str | None = None) -> "ToolOutput":
+        out = super().__new__(cls, model_text)
+        out.display_text = display_text
+        return out
+
+
 class Tool(ABC):
     """
     Abstract base class for agent tools.
@@ -52,18 +89,29 @@ def parameters(self) -> dict[str, Any]:
         pass
 
     @abstractmethod
-    async def execute(self, **kwargs: Any) -> str:
+    async def execute(self, **kwargs: Any) -> "str | ToolResult":
         """
         Execute the tool with given parameters.
 
         Args:
             **kwargs: Tool-specific parameters.
 
         Returns:
-            String result of the tool execution.
+            The model-facing result as a ``str``, or a :class:`ToolResult` when
+            the tool wants a distinct human-facing display string.
         """
         pass
 
+    def display_call(self, args: dict[str, Any]) -> str | None:
+        """Human-facing one-line summary of a call to this tool.
+
+        ``None`` (the default) lets the UI derive a generic summary from the
+        arguments. Override only when a tool wants a cleaner label than the
+        generic one (e.g. ask_user showing just its question, not the raw
+        arguments blob).
+        """
+        return None
+
     def cast_params(self, params: dict[str, Any]) -> dict[str, Any]:
         """Apply safe schema-driven casts before validation."""
         schema = self.parameters or {}

raven/agent/tools/deep_research.py

@@ -311,7 +311,11 @@ def _receipt(status: str, report_ref: str | None) -> str:
 # well-aligned model treats tool output as data and refuses embedded commands (a
 # strong "do NOT answer / reply with only this" was rejected as a prompt injection
 # in testing). (The older receipt/ack strings below keep mild imperatives.)
-_USE_REGULAR = "deep_research did not run: the user chose a regular web search instead."
+_USE_REGULAR = (
+    "The user chose a regular web search for this query, not the deep-research engine. "
+    "Go ahead and answer it with a normal web search -- this is the user's chosen path, "
+    "not a failure or a fallback."
+)
 # Shown to the USER directly via the broker prompt below. Command FIRST so it
 # survives the scrollback's truncated tool-call line (cut at ~60 chars); the live
 # picker wraps and shows it all anyway.

raven/agent/tools/registry.py

@@ -3,7 +3,7 @@
 import asyncio
 from typing import Any
 
-from raven.agent.tools.base import Tool
+from raven.agent.tools.base import Tool, ToolOutput, ToolResult
 from raven.tracing import semconv, trace
 
 
@@ -67,9 +67,17 @@ async def execute(self, name: str, params: dict[str, Any]) -> str:
             else:
                 result = await asyncio.wait_for(tool.execute(**params), timeout=ceiling)
 
-            if isinstance(result, str) and result.startswith("Error"):
-                return result + _hint
-            return result
+            # Unwrap ToolResult here, at the boundary: `execute` promises model
+            # text to every caller, and only the agent loop wants the display
+            # string (which rides along on ToolOutput).
+            if isinstance(result, ToolResult):
+                model_text, display_text = result.model_text, result.display_text
+            else:
+                model_text, display_text = str(result), None
+
+            if model_text.startswith("Error"):
+                return ToolOutput(model_text + _hint, display_text)
+            return ToolOutput(model_text, display_text)
         except asyncio.TimeoutError:
             return f"Error: Tool '{name}' timed out after {ceiling:.0f}s." + _hint
         except Exception as e:

raven/spine/__init__.py

@@ -7,6 +7,7 @@
 
 from raven.spine.events import (
     Deliverable,
+    EpisodeStart,
     MediaOut,
     Notice,
     NoticeKind,
@@ -32,6 +33,7 @@
     "ChatType",
     "Deliverable",
     "Emit",
+    "EpisodeStart",
     "Media",
     "MediaOut",
     "Notice",

raven/spine/events.py

@@ -66,6 +66,8 @@ class ToolEvent:
     tool_call_id: str
     name: str = ""
     arguments: dict[str, Any] | None = None
+    # Tool-authored call label for the start phase; None -> UI derives one.
+    display: str | None = None
     result_preview: str = ""
     truncated: bool = False
     source: Source | None = None
@@ -110,7 +112,18 @@ class Notice:
     conversation_id: str | None = None
 
 
-RunnerEvent = ToolEvent | Text | MediaOut | StreamDelta | Reasoning | Notice
+@dataclass(frozen=True)
+class EpisodeStart:
+    """Boundary marker: a new model call (episode) begins. ``index`` is the
+    0-based step within the turn. Outlets that group a turn into per-call
+    episodes use it to start a fresh bucket; others ignore it."""
+
+    index: int
+    source: Source | None = None
+    conversation_id: str | None = None
+
+
+RunnerEvent = ToolEvent | Text | MediaOut | StreamDelta | Reasoning | Notice | EpisodeStart
 # Same union, named for its delivery role: what the hub routes and an Outlet renders.
 Deliverable = RunnerEvent
 TurnEvent = TurnStarted | TurnFailed | TurnEnded | RunnerEvent

raven/tui_rpc/models.py

@@ -139,6 +139,15 @@ class MessageStartEvent(_Strict):
     payload: MessageStartPayload
 
 
+class EpisodeStartPayload(_Strict):
+    index: int
+
+
+class EpisodeStartEvent(_Strict):
+    type: Literal["episode.start"]
+    payload: EpisodeStartPayload
+
+
 class TokenDeltaPayload(_Strict):
     text: str
 
@@ -161,6 +170,7 @@ class ToolStartPayload(_Strict):
     tool_call_id: str
     name: str
     arguments: dict[str, JsonValue]
+    display: str | None = None
 
 
 class ToolStartEvent(_Strict):
@@ -226,6 +236,7 @@ class CronDeliveredEvent(_Strict):
 TurnEvent = Annotated[
     Union[
         MessageStartEvent,
+        EpisodeStartEvent,
         TokenDeltaEvent,
         ThinkingDeltaEvent,
         ToolStartEvent,
@@ -897,6 +908,7 @@ class ToolsConfigureParams(_Strict):
     "SessionExportParams",
     "SessionExportResult",
     "MessageStartEvent",
+    "EpisodeStartEvent",
     "TokenDeltaEvent",
     "ThinkingDeltaEvent",
     "ToolStartEvent",