mirror of
https://github.com/Alishahryar1/free-claude-code.git
synced 2026-07-03 14:05:26 +02:00
Ali Khokhar
111 lines
3.4 KiB
Python
111 lines
3.4 KiB
Python
"""Task/subagent display state for messaging transcripts."""
|
|
|
|
from __future__ import annotations
|
|
|
|
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)
|
|
)
|