Spaces:
Sleeping
Sleeping
File size: 3,476 Bytes
2415446 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 | """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
@property
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)
)
|