Spaces:
Sleeping
Sleeping
| """Task/subagent display state for messaging transcripts.""" | |
| from typing import Any | |
| from loguru import logger | |
| from .segments import SubagentSegment | |
| class SubagentState: | |
| """Track active Task tool calls that suppress nested text/thinking output.""" | |
| def __init__(self, *, debug: bool = False) -> None: | |
| self._stack: list[str] = [] | |
| self._segments: list[SubagentSegment] = [] | |
| self._debug = debug | |
| def open_ids(self) -> tuple[str, ...]: | |
| return tuple(self._stack) | |
| def in_subagent(self) -> bool: | |
| return bool(self._stack) | |
| def current_segment(self) -> SubagentSegment | None: | |
| return self._segments[-1] if self._segments else None | |
| def push(self, tool_id: str, segment: SubagentSegment) -> None: | |
| marker = str(tool_id or "").strip() or f"__task_{len(self._stack) + 1}" | |
| self._stack.append(marker) | |
| self._segments.append(segment) | |
| if self._debug: | |
| logger.debug( | |
| "SUBAGENT_STACK: push id=%r depth=%d heading=%r", | |
| marker, | |
| len(self._stack), | |
| segment.description, | |
| ) | |
| def close_for_tool_result(self, tool_id: str, *, tool_name: str | None) -> bool: | |
| tool_id = str(tool_id or "").strip() | |
| popped = self._pop(tool_id) | |
| top = self._stack[-1] if self._stack else "" | |
| looks_like_task_id = "task" in tool_id.lower() | |
| if ( | |
| not popped | |
| and tool_id | |
| and top.startswith("__task_") | |
| and tool_name in (None, "Task") | |
| and looks_like_task_id | |
| ): | |
| return self._pop("") | |
| return popped | |
| def _pop(self, tool_id: str) -> bool: | |
| tool_id = str(tool_id or "").strip() | |
| if not self._stack: | |
| return False | |
| if tool_id: | |
| if _ids_roughly_match(self._stack[-1], tool_id): | |
| self._pop_to_depth(len(self._stack) - 1, tool_id, "LIFO") | |
| return True | |
| for idx in range(len(self._stack) - 1, -1, -1): | |
| if _ids_roughly_match(self._stack[idx], tool_id): | |
| self._pop_to_depth(idx, tool_id, "matched") | |
| return True | |
| return False | |
| if self._stack[-1].startswith("__task_"): | |
| self._pop_to_depth(len(self._stack) - 1, self._stack[-1], "synthetic") | |
| return True | |
| return False | |
| def _pop_to_depth(self, idx: int, requested_id: str, reason: str) -> None: | |
| while len(self._stack) > idx: | |
| popped = self._stack.pop() | |
| if self._segments: | |
| self._segments.pop() | |
| if self._debug: | |
| logger.debug( | |
| "SUBAGENT_STACK: pop id=%r depth=%d (%s=%r)", | |
| popped, | |
| len(self._stack), | |
| reason, | |
| requested_id, | |
| ) | |
| def task_heading_from_input(input_value: Any) -> str: | |
| if isinstance(input_value, dict): | |
| for key in ("description", "subagent_type", "type"): | |
| value = str(input_value.get(key, "") or "").strip() | |
| if value: | |
| return value | |
| return "Subagent" | |
| def _ids_roughly_match(stack_id: str, result_id: str) -> bool: | |
| if not stack_id or not result_id: | |
| return False | |
| return ( | |
| stack_id == result_id | |
| or stack_id.startswith(result_id) | |
| or result_id.startswith(stack_id) | |
| ) | |