hermes-agent/tools/delegate_tool_progress.py

"""Child progress relay, console formatting and child system-prompt construction for delegate_task."""

from __future__ import annotations

import logging
import enum
import os
import threading
from contextlib import contextmanager
from typing import Any, Dict, List, Optional
from tools.delegate_tool_registry import _active_subagents, _active_subagents_lock

logger = logging.getLogger("tools.delegate_tool")  # log-record parity with the origin module

# Terminal child statuses that mean "the subagent did NOT deliver a usable result". Shared by the CLI spinner echo,
# the gateway failure notice, and the parent-facing failure summary so every surface agrees.
SUBAGENT_FAILURE_STATUSES = frozenset({"failed", "error", "timeout"})

@contextmanager
def _quiet(log_message: Optional[str], *log_args: Any, exc_info: bool = False):
    """Best-effort block: any Exception is swallowed (never reaches the run) and, when ``log_message`` is given,
    logged at debug — the exception fills a trailing unsatisfied ``%s``."""
    try:
        yield
    except Exception as exc:
        if log_message is None:
            return
        if log_message.count("%") > len(log_args):
            log_args = log_args + (exc,)
        logger.debug(log_message, *log_args, exc_info=exc_info)

def _safe_progress(cb: Any, event_type: Any, *args: Any, **kwargs: Any) -> None:
    """Invoke a child progress callback; relay failures never reach the run."""
    if cb:
        with _quiet("Progress callback %s failed: %s", event_type):
            cb(event_type, *args, **kwargs)

def _clean_error_text(error: Any, max_chars: int = 200) -> str:
    """Reduce an error payload (traceback / JSON wall) to one clean line: the exception message (last line of a
    traceback) or the first non-empty line, hard-capped in length."""
    lines = [ln.strip() for ln in str(error or "").strip().splitlines() if ln.strip()]
    if not lines:
        return ""
    line = lines[-1] if lines[0].startswith("Traceback") else lines[0]
    return line[: max_chars - 3] + "..." if len(line) > max_chars else line


def describe_subagent_failure(failure_reason: Any, error: Any, max_chars: int = 200) -> str:
    """Plain-language reason for a failed child: the classified ``failure_reason`` gloss when there is one,
    else the child's own error text reduced to one clean line."""
    # One gloss table for every surface (agent/turn_failure_copy.py); the child is the subject.
    from agent.turn_failure_copy import failure_cause_gloss

    gloss = failure_cause_gloss(failure_reason, subject="it", possessive="its")
    return gloss or _clean_error_text(error, max_chars)


def _format_duration(seconds: Any) -> str:
    if not isinstance(seconds, (int, float)) or seconds <= 0:
        return ""
    return f"{round(seconds / 60)} min" if seconds >= 120 else f"{round(seconds)}s"


def format_subagent_failure_line(
    goal: Optional[str], status: Optional[str], error: Any = None, duration_seconds: Any = None,
    failure_reason: Any = None,
) -> str:
    """One clean, human-readable line describing a failed subagent, rendered directly to the user (CLI spinner
    echo, gateway platform notice). Says what happened and what to do next; never the raw exception when the
    child loop classified the failure, e.g.
    ``⚠️ Subagent failed — "research competitor pricing" after 12s: the model it was given was not found at the
    AI model service. Details: /agents, or ask me to retry with a smaller task.``"""
    goal_label = (goal or "").strip().replace("\n", " ")
    if len(goal_label) > 60:
        goal_label = goal_label[:57] + "..."
    goal_part = f' — "{goal_label}"' if goal_label else ""
    elapsed = _format_duration(duration_seconds)
    if status == "timeout":
        # The child's own timeout text repeats the duration and names mechanisms (API/tool calls); the
        # user needs the outcome and the knob.
        after = f" after {elapsed}" if elapsed else ""
        return (
            f"⚠️ Subagent timed out{goal_part}{after} without finishing. I will carry on without it; ask me to "
            "retry it, or raise delegation.child_timeout_seconds in config.yaml if these tasks legitimately "
            "take longer."
        )
    line = f"⚠️ Subagent failed{goal_part}"
    if elapsed:
        line += f" after {elapsed}"
    reason = describe_subagent_failure(failure_reason, error)
    if reason:
        line += f": {reason.rstrip('.')}"
    return line + ". Details: /agents, or ask me to retry with a smaller task."


class DelegateEvent(str, enum.Enum):
    """Formal delegation progress event types. The relay normalises incoming legacy strings (``tool.started``,
    ``_thinking``, …) to these via ``_LEGACY_EVENT_MAP``; external consumers (gateway SSE, ACP adapter, CLI) still
    receive the legacy strings during the deprecation window. TASK_SPAWNED / TASK_COMPLETED / TASK_FAILED are
    reserved for future orchestrator lifecycle events, not emitted yet."""

    TASK_SPAWNED = "delegate.task_spawned"
    TASK_PROGRESS = "delegate.task_progress"
    TASK_COMPLETED = "delegate.task_completed"
    TASK_FAILED = "delegate.task_failed"
    TASK_THINKING = "delegate.task_thinking"
    TASK_TOOL_STARTED = "delegate.tool_started"
    TASK_TOOL_COMPLETED = "delegate.tool_completed"

# Legacy child-agent event strings → DelegateEvent.
_LEGACY_EVENT_MAP: Dict[str, DelegateEvent] = {
    "_thinking": DelegateEvent.TASK_THINKING,
    "reasoning.available": DelegateEvent.TASK_THINKING,
    "tool.started": DelegateEvent.TASK_TOOL_STARTED,
    "tool.completed": DelegateEvent.TASK_TOOL_COMPLETED,
    "subagent_progress": DelegateEvent.TASK_PROGRESS,
}

# Event → _ChildProgressRelay method name. Lifecycle strings are emitted by the orchestrator itself (not
# DelegateEvent). Any other DelegateEvent (TASK_TOOL_STARTED and the reserved TASK_* values) takes the tool-started
# path; None means "recognised but ignored".
_LIFECYCLE_EVENTS = frozenset({"subagent.start", "subagent.complete", "subagent.text"})
_EVENT_HANDLERS: Dict[Any, Optional[str]] = {
    "subagent.start": "_on_start",
    "subagent.complete": "_on_complete",
    "subagent.text": "_on_text",
    DelegateEvent.TASK_THINKING: "_on_thinking",
    DelegateEvent.TASK_PROGRESS: "_on_progress",
    DelegateEvent.TASK_TOOL_COMPLETED: None,
}

def _normalize_event(event_type: Any) -> Any:
    """Lifecycle string / DelegateEvent / legacy string / ``delegate.*`` string
    → dispatch key; None for unknown events."""
    if isinstance(event_type, DelegateEvent) or (isinstance(event_type, str) and event_type in _LIFECYCLE_EVENTS):
        return event_type
    try:
        return _LEGACY_EVENT_MAP.get(event_type) or DelegateEvent(event_type)
    except (ValueError, TypeError):
        return None

_CONTEXT_FILES_INTRO = (
    "\nThe workspace's project context files are reproduced below. Their conventions and invariants are binding for "
    "your work in this workspace.\n\n"
)
_COMPLETION_INSTRUCTIONS = (
    "\nComplete this task using the tools available to you. When finished, provide a clear, concise summary of:\n"
    "- What you did\n- What you found or accomplished\n- Any files you created or modified\n- Any issues encountered\n\n"
    "Important workspace rule: Never assume a repository lives at /workspace/... or any other container-style path "
    "unless the task/context explicitly gives that path. If no exact local path is provided, discover it first before "
    "issuing git/workdir-specific commands.\n\n"
    "Keep your final summary tight: lead with outcomes, prefer bullet points over paragraphs, and don't replay your "
    "whole process. Your response is returned to the parent agent as a summary, and overlong summaries crowd out the "
    "parent's context window."
)
_ORCHESTRATOR_BLOCK = (
    "\n## Subagent Spawning (Orchestrator Role)\n"
    "You have access to the `delegate_task` tool and CAN spawn your own subagents to parallelize independent work.\n\n"
    "WHEN to delegate:\n"
    "- The goal decomposes into 2+ independent subtasks that can run in parallel (e.g. research A and B simultaneously).\n"
    "- A subtask is reasoning-heavy and would flood your context with intermediate data.\n\n"
    "WHEN NOT to delegate:\n"
    "- Single-step mechanical work — do it directly.\n"
    "- Trivial tasks you can execute in one or two tool calls.\n"
    "- Re-delegating your entire assigned goal to one worker (that's just pass-through with no value added).\n\n"
    "Coordinate your workers' results and synthesize them before reporting back to your parent. You are responsible "
    "for the final summary, not your workers.\n\n"
)
_LEAF_CHILDREN_NOTE = (
    "Your own children MUST be leaves (cannot delegate further) because they will reach the maximum nesting depth. "
    "The system manages this automatically; no parameter override is available."
)
_NESTED_CHILDREN_NOTE = (
    "Your own children can themselves delegate because depth remains. The system determines this automatically "
    "from the nesting depth; children at the maximum depth cannot delegate further. No parameter override is available."
)

def _build_child_system_prompt(
    goal: str, context: Optional[str] = None, *, workspace_path: Optional[str] = None, role: str = "leaf",
    max_spawn_depth: int = 2, child_depth: int = 1,
) -> str:
    """Focused system prompt for a child agent. role='orchestrator' appends a delegation-capability block (modeled on
    OpenClaw's buildSubagentSystemPrompt); its depth note is literal truth grounded in the passed config so the LLM
    can't confabulate nesting."""
    # The goal is the child's first user turn (see ``_ChildRun.await_child``).
    # Keeping it out of the system prompt avoids sending OAuth Anthropic the
    # same task in both roles, while preserving the normal user-turn contract.
    parts = ["You are a focused subagent working on a specific delegated task."]
    if context and context.strip():
        parts.append(f"\nCONTEXT:\n{context}")
    if workspace_path and str(workspace_path).strip():
        parts.append(
            "\nWORKSPACE PATH:\n"
            f"{workspace_path}\n"
            "Use this exact path for local repository/workdir operations unless the task explicitly says otherwise."
        )
        # Project context files (AGENTS.md / CLAUDE.md / .cursorrules ...) via the SAME discovery/priority/cap logic
        # as the main agent's prompt: children are built with skip_context_files=True, so without this a subagent
        # works in a repo blind to its conventions. SOUL.md is skipped (identity belongs to the parent).
        # workspace_path comes only from explicit sources (_resolve_workspace_hint, never bare getcwd), so the
        # install-tree-fallback leak doesn't apply. Best-effort.
        _ctx_files = ""
        with _quiet("subagent: workspace context-files load failed", exc_info=True):
            # See #64590.
            from agent.prompt_builder import build_context_files_prompt
            _ctx_files = build_context_files_prompt(cwd=str(workspace_path), skip_soul=True)
        if _ctx_files.strip():
            parts.append(_CONTEXT_FILES_INTRO + _ctx_files.strip())
    parts.append(_COMPLETION_INSTRUCTIONS)
    if role == "orchestrator":
        child_note = _LEAF_CHILDREN_NOTE if child_depth + 1 >= max_spawn_depth else _NESTED_CHILDREN_NOTE
        parts.append(
            _ORCHESTRATOR_BLOCK
            + f"NOTE: You are at depth {child_depth}. The delegation tree is capped at max_spawn_depth={max_spawn_depth}. "
            + child_note
        )
    return "\n".join(parts)

def _resolve_workspace_hint(parent_agent) -> Optional[str]:
    """Best-effort local workspace hint for child prompts: only a concrete
    absolute directory is ever injected (never a fake container path)."""
    from agent.runtime_cwd import scope_terminal_cwd
    candidates = [
        scope_terminal_cwd(), getattr(getattr(parent_agent, "_subdirectory_hints", None), "working_dir", None),
        getattr(parent_agent, "terminal_cwd", None), getattr(parent_agent, "cwd", None),
    ]
    for candidate in filter(None, candidates):
        with _quiet(None):
            text = os.path.abspath(os.path.expanduser(str(candidate)))
            if os.path.isabs(text) and os.path.isdir(text):
                return text
    return None

_BATCH_ORDINALS: Dict[str, Dict[str, int]] = {}
_BATCH_ORDINALS_LOCK = threading.Lock()

def format_batch_tag(delegation_id: Optional[str], parent_agent: Any = None) -> str:
    """Short human tag for a delegation batch: the parent's first fan-out is ``set 1``, its next distinct
    delegation id ``set 2``. Several batches (a parent's fan-out plus a child's nested fan-out, or two
    concurrent tools) print interleaved ``[n/N]`` lines to one console; without a tag ``✓ [3/3]`` and ``✓ [3/9]``
    are indistinguishable, and a raw hex slice is unreadable. Ordinals are scoped per parent conversation
    (``parent_agent.session_id``): one process hosts many conversations and every child's own fan-out, so a
    process-wide counter showed a user's second wave as ``set 20``. Empty string when no id is known so callers
    can concatenate unconditionally."""
    if not isinstance(delegation_id, str) or not delegation_id:
        return ""
    scope = str(getattr(parent_agent, "session_id", None) or "")
    with _BATCH_ORDINALS_LOCK:
        ordinals = _BATCH_ORDINALS.setdefault(scope, {})
        n = ordinals.setdefault(delegation_id, len(ordinals) + 1)
    return f"set {n}"

def _batch_prefix(delegation_id: Optional[str], task_index: int, task_count: int, parent_agent: Any = None) -> str:
    """``[set 2 · 3/9] `` for batch children, ``[set 2] `` for a lone child,
    ``[3/9] `` / ``""`` when the batch id is unknown."""
    tag = format_batch_tag(delegation_id, parent_agent)
    if task_count > 1:
        inner = f"{tag} · {task_index + 1}/{task_count}" if tag else f"{task_index + 1}/{task_count}"
        return f"[{inner}] "
    return f"[{tag}] " if tag else ""

def _emit_parent_console(parent_agent, line: str) -> None:
    """Emit a progress line through ``parent_agent._safe_print`` when available so headless stdio hosts (ACP, gateway
    API) can redirect it to stderr; a bare ``print()`` would land on stdout and corrupt JSON-RPC framing."""
    printer = getattr(parent_agent, "_safe_print", None)
    if callable(printer):
        with _quiet(None):
            printer(line)
            return
    print(line)

def _print_completion_line(parent_agent: Any, spinner_ref: Any, line: str, console_line: Optional[str] = None) -> None:
    """Above-spinner line when a spinner exists (console fallback if it raises), else console
    (``console_line`` when given, else the line indented two spaces)."""
    if spinner_ref:
        with _quiet(None):
            spinner_ref.print_above(line)
            return
    _emit_parent_console(parent_agent, f"  {line}" if console_line is None else console_line)

def _short(text: str, n: int) -> str:
    return (text[:n] + "...") if len(text) > n else text


class _ChildProgressRelay:
    """Callable relaying one child's events to the parent display. CLI: prints tree-view lines above the parent's
    delegation spinner. Gateway: batches tool names (``_BATCH_SIZE``) and relays to the parent's progress callback,
    threading the identity kwargs (subagent_id, parent_id, depth, model, toolsets) into every event so the TUI can
    rebuild the live spawn tree and route per-branch controls back by ``subagent_id``."""

    _BATCH_SIZE = 5

    def __init__(
        self, task_index: int, goal: str, spinner: Any, parent_cb: Any, task_count: int,
        subagent_id, parent_id, depth, model, toolsets, session_ref,
    ) -> None:
        self.task_index, self.task_count, self.goal_label = task_index, task_count, (goal or "").strip()
        # session_ref is a SHARED dict filled in later by the caller — keep the identity.
        self.spinner, self.parent_cb, self.session_ref = spinner, parent_cb, session_ref if session_ref is not None else {}
        self.subagent_id, self.parent_id, self.depth, self.model, self.toolsets = (
            subagent_id, parent_id, depth, model, toolsets
        )
        self.batch: List[str] = []
        self.parent_scope: Any = None  # owning parent agent; set by _build_child_progress_callback
        self.tool_count = 0  # per-subagent running counter

    def _prefix(self) -> str:
        # The batch tag is resolved lazily from session_ref: the relay is built
        # before delegate_task stamps ``_delegation_id`` on the child. The parent
        # scope rides in the same ref so ordinals match the parent's own lines.
        return _batch_prefix(
            self.session_ref.get("delegation_id"), self.task_index, self.task_count,
            parent_agent=self.session_ref.get("_parent_scope"),
        )

    def _identity_kwargs(self) -> Dict[str, Any]:
        kw: Dict[str, Any] = {"task_index": self.task_index, "task_count": self.task_count, "goal": self.goal_label}
        kw.update({k: getattr(self, k) for k in ("subagent_id", "parent_id", "depth", "model") if getattr(self, k) is not None})
        if self.toolsets is not None:
            kw["toolsets"] = list(self.toolsets)
        # child_session_id / delegation_id are filled into the shared ref once
        # the child exists, so every relayed event lets UIs open its session.
        for src, dst in (("session_id", "child_session_id"), ("delegation_id", "delegation_id")):
            if self.session_ref.get(src):
                kw[dst] = str(self.session_ref[src])
        kw["tool_count"] = self.tool_count
        return kw

    def _relay(self, event_type: str, tool_name: str = None, preview: str = None, args=None, **kwargs):
        if self.parent_cb:
            # kwargs override identity (e.g. status, duration_seconds).
            with _quiet("Parent callback failed: %s"):
                self.parent_cb(event_type, tool_name, preview, args, **{**self._identity_kwargs(), **kwargs})

    def _tree_line(self, text: str) -> None:
        """Print one tree-view line above the CLI spinner (no-op without a spinner)."""
        if self.spinner:
            with _quiet("Spinner print_above failed: %s"):
                self.spinner.print_above(f" {self._prefix()}├─ {text}")

    def _flush(self) -> None:
        """Flush remaining batched tool names to the gateway."""
        if self.parent_cb and self.batch:
            self._relay("subagent.progress", preview=f"🔀 {self._prefix()}{', '.join(self.batch)}")
            self.batch.clear()

    # ── Lifecycle events emitted by the orchestrator itself ──
    def _on_start(self, tool_name, preview, args, kwargs):
        if self.goal_label:
            self._tree_line(f"🔀 {_short(self.goal_label, 55)}")
        self._relay("subagent.start", preview=preview or self.goal_label or "", **kwargs)

    def _on_complete(self, tool_name, preview, args, kwargs):
        # Failed child: echo one clean reason line into the CLI tree so the human
        # sees WHY, not just a vanished branch (gateway renders off the relayed event).
        # The echo is an automatic diagnostic presentation: it goes through the warning
        # boundary under the parent's turn snapshot. The relayed event (the gateway's
        # producer, which classifies it) and the child result are never gated here.
        if kwargs.get("status") in SUBAGENT_FAILURE_STATUSES:
            from gateway.warning_notifications import render_notification
            parent = self.parent_scope
            render_notification(
                lambda: self._tree_line(format_subagent_failure_line(
                    self.goal_label, kwargs.get("status"), error=kwargs.get("summary") or preview,
                    duration_seconds=kwargs.get("duration_seconds"), failure_reason=kwargs.get("failure_reason"),
                )),
                platform=getattr(parent, "_notification_platform", getattr(parent, "platform", "cli")),
                user_config=getattr(parent, "_notification_config", None),
            )
        self._relay("subagent.complete", preview=preview, **kwargs)

    def _on_text(self, tool_name, preview, args, kwargs):
        # Streamed child reply text, relayed verbatim for gateway watch windows;
        # inert on CLI/TUI (their progress handlers ignore non-tool events).
        self._relay("subagent.text", preview=preview)

    # ── DelegateEvent handlers ──
    def _on_thinking(self, tool_name, preview, args, kwargs):
        text = preview or tool_name or ""
        self._tree_line(f'💭 "{_short(text, 55)}"')
        self._relay("subagent.thinking", preview=text)

    def _on_progress(self, tool_name, preview, args, kwargs):
        # Pre-batched summary from a nested orchestrator's grandchild arrives in the tool_name slot: render distinctly
        # (no tool-emoji lookup) and relay upward without re-batching.
        summary_text = tool_name or preview or ""
        if summary_text:
            self._tree_line(f"🔀 {summary_text}")
        if self.parent_cb:
            with _quiet("Parent callback relay failed: %s"):
                self.parent_cb("subagent_progress", f"{self._prefix()}{summary_text}")

    def _on_tool_started(self, tool_name, preview, args, kwargs):
        self.tool_count += 1
        if self.subagent_id is not None:
            with _active_subagents_lock:
                rec = _active_subagents.get(self.subagent_id)
                if rec is not None:
                    rec["tool_count"] = self.tool_count
                    rec["last_tool"] = tool_name or ""
        if self.spinner:
            from agent.display import get_tool_emoji
            line = f"{get_tool_emoji(tool_name or '')} {tool_name}"
            short = _short(preview, 35) if preview else ""
            self._tree_line(f'{line}  "{short}"' if short else line)
        if self.parent_cb:
            self._relay("subagent.tool", tool_name, preview, args)
            self.batch.append(tool_name or "")
            if len(self.batch) >= self._BATCH_SIZE:
                self._flush()

    def __call__(self, event_type, tool_name: str = None, preview: str = None, args=None, **kwargs):
        key = _normalize_event(event_type)
        method = None if key is None else _EVENT_HANDLERS.get(key, "_on_tool_started")
        if method is not None:
            getattr(self, method)(tool_name, preview, args, kwargs)

def _build_child_progress_callback(
    task_index: int, goal: str, parent_agent, task_count: int = 1, *, subagent_id: Optional[str] = None,
    parent_id: Optional[str] = None, depth: Optional[int] = None, model: Optional[str] = None,
    toolsets: Optional[List[str]] = None, session_ref: Optional[Dict[str, Any]] = None,
) -> Optional[callable]:
    """Relay for one child's events (see ``_ChildProgressRelay``), or None when the parent has neither a spinner nor a
    progress callback — the child then runs with no progress callback at all (zero behavior change)."""
    spinner = getattr(parent_agent, "_delegate_spinner", None)
    parent_cb = getattr(parent_agent, "tool_progress_callback", None)
    if not spinner and not parent_cb:
        return None
    if session_ref is not None:
        # Not an identity kwarg (underscore-prefixed, never relayed); only scopes the batch ordinal.
        session_ref["_parent_scope"] = parent_agent
    relay = _ChildProgressRelay(
        task_index, goal, spinner, parent_cb, task_count, subagent_id, parent_id, depth, model, toolsets, session_ref,
    )
    relay.parent_scope = parent_agent
    return relay