| """Transcript event application and open-block tracking.""" |
|
|
| from typing import Any |
|
|
| from .context import RenderCtx |
| from .renderer import render_segments |
| from .segments import ( |
| ErrorSegment, |
| Segment, |
| SubagentSegment, |
| TextSegment, |
| ThinkingSegment, |
| ToolCallSegment, |
| ToolResultSegment, |
| ) |
| from .subagents import SubagentState, task_heading_from_input |
|
|
| _SUBAGENT_SUPPRESSED_EVENTS = frozenset( |
| { |
| "thinking_start", |
| "thinking_delta", |
| "thinking_chunk", |
| "text_start", |
| "text_delta", |
| "text_chunk", |
| } |
| ) |
|
|
|
|
| class TranscriptBuffer: |
| """Maintains an ordered, truncatable transcript of parsed CLI events.""" |
|
|
| def __init__( |
| self, |
| *, |
| show_tool_results: bool = True, |
| debug_subagent_stack: bool = False, |
| ) -> None: |
| self._segments: list[Segment] = [] |
| self._open_thinking_by_index: dict[int, ThinkingSegment] = {} |
| self._open_text_by_index: dict[int, TextSegment] = {} |
| self._open_tools_by_index: dict[int, ToolCallSegment] = {} |
| self._tool_name_by_id: dict[str, str] = {} |
| self._show_tool_results = bool(show_tool_results) |
| self._subagents = SubagentState(debug=debug_subagent_stack) |
|
|
| def apply(self, event: dict[str, Any]) -> None: |
| """Apply a parsed CLI transcript event.""" |
| event_type = event.get("type") |
| if self._subagents.in_subagent() and event_type in _SUBAGENT_SUPPRESSED_EVENTS: |
| return |
|
|
| if event_type == "thinking_start": |
| self._start_thinking(_event_index(event)) |
| return |
| if event_type in ("thinking_delta", "thinking_chunk"): |
| self._append_thinking(_event_index(event), str(event.get("text", ""))) |
| return |
| if event_type == "thinking_stop": |
| self._open_thinking_by_index.pop(_event_index(event), None) |
| return |
|
|
| if event_type == "text_start": |
| self._start_text(_event_index(event)) |
| return |
| if event_type in ("text_delta", "text_chunk"): |
| self._append_text(_event_index(event), str(event.get("text", ""))) |
| return |
| if event_type == "text_stop": |
| self._open_text_by_index.pop(_event_index(event), None) |
| return |
|
|
| if event_type == "tool_use_start": |
| self._start_tool_use(event) |
| return |
| if event_type == "tool_use_delta": |
| return |
| if event_type == "tool_use_stop": |
| segment = self._open_tools_by_index.pop(_event_index(event), None) |
| if segment is not None: |
| segment.closed = True |
| return |
|
|
| if event_type == "block_stop": |
| self._close_block(_event_index(event)) |
| return |
| if event_type == "tool_use": |
| self._append_complete_tool_use(event) |
| return |
| if event_type == "tool_result": |
| self._append_tool_result(event) |
| return |
| if event_type == "error": |
| self._segments.append(ErrorSegment(str(event.get("message", "")))) |
|
|
| def render(self, ctx: RenderCtx, *, limit_chars: int, status: str | None) -> str: |
| return render_segments( |
| self._segments, |
| ctx, |
| limit_chars=limit_chars, |
| status=status, |
| ) |
|
|
| def _start_thinking(self, index: int) -> None: |
| if index >= 0: |
| self._close_block(index) |
| segment = ThinkingSegment() |
| self._segments.append(segment) |
| if index >= 0: |
| self._open_thinking_by_index[index] = segment |
|
|
| def _append_thinking(self, index: int, text: str) -> None: |
| segment = self._open_thinking_by_index.get(index) |
| if segment is None: |
| segment = ThinkingSegment() |
| self._segments.append(segment) |
| if index >= 0: |
| self._open_thinking_by_index[index] = segment |
| segment.append(text) |
|
|
| def _start_text(self, index: int) -> None: |
| if index >= 0: |
| self._close_block(index) |
| segment = TextSegment() |
| self._segments.append(segment) |
| if index >= 0: |
| self._open_text_by_index[index] = segment |
|
|
| def _append_text(self, index: int, text: str) -> None: |
| segment = self._open_text_by_index.get(index) |
| if segment is None: |
| segment = TextSegment() |
| self._segments.append(segment) |
| if index >= 0: |
| self._open_text_by_index[index] = segment |
| segment.append(text) |
|
|
| def _start_tool_use(self, event: dict[str, Any]) -> None: |
| index = _event_index(event) |
| if index >= 0: |
| self._close_block(index) |
|
|
| tool_id = _event_tool_id(event, "id") |
| name = str(event.get("name", "") or "tool") |
| if tool_id: |
| self._tool_name_by_id[tool_id] = name |
|
|
| if name == "Task": |
| segment = SubagentSegment(task_heading_from_input(event.get("input"))) |
| self._segments.append(segment) |
| self._subagents.push(tool_id, segment) |
| return |
|
|
| segment = self._append_tool_call(tool_id, name) |
| if index >= 0: |
| self._open_tools_by_index[index] = segment |
|
|
| def _append_complete_tool_use(self, event: dict[str, Any]) -> None: |
| tool_id = _event_tool_id(event, "id") |
| name = str(event.get("name", "") or "tool") |
| if tool_id: |
| self._tool_name_by_id[tool_id] = name |
|
|
| if name == "Task": |
| segment = SubagentSegment(task_heading_from_input(event.get("input"))) |
| self._segments.append(segment) |
| self._subagents.push(tool_id, segment) |
| return |
|
|
| segment = self._append_tool_call(tool_id, name) |
| segment.closed = True |
|
|
| def _append_tool_call(self, tool_id: str, name: str) -> ToolCallSegment: |
| if self._subagents.in_subagent(): |
| parent = self._subagents.current_segment() |
| if parent is not None: |
| return parent.set_current_tool_call(tool_id, name) |
|
|
| segment = ToolCallSegment(tool_id, name) |
| self._segments.append(segment) |
| return segment |
|
|
| def _append_tool_result(self, event: dict[str, Any]) -> None: |
| tool_id = _event_tool_id(event, "tool_use_id") |
| name = self._tool_name_by_id.get(tool_id) |
|
|
| if self._subagents.in_subagent(): |
| self._subagents.close_for_tool_result(tool_id, tool_name=name) |
|
|
| if not self._show_tool_results: |
| return |
|
|
| self._segments.append( |
| ToolResultSegment( |
| tool_id, |
| event.get("content"), |
| name=name, |
| is_error=bool(event.get("is_error", False)), |
| ) |
| ) |
|
|
| def _close_block(self, index: int) -> None: |
| if index in self._open_tools_by_index: |
| segment = self._open_tools_by_index.pop(index, None) |
| if segment is not None: |
| segment.closed = True |
| return |
| if index in self._open_thinking_by_index: |
| self._open_thinking_by_index.pop(index, None) |
| return |
| if index in self._open_text_by_index: |
| self._open_text_by_index.pop(index, None) |
|
|
|
|
| def _event_index(event: dict[str, Any]) -> int: |
| return int(event.get("index", -1)) |
|
|
|
|
| def _event_tool_id(event: dict[str, Any], key: str) -> str: |
| return str(event.get(key, "") or "").strip() |
|
|
