2776813df3
The Sep 2026 decomposition (PR #102117) makes internal import paths a non-API: names now live in the focused modules that define them. This commit is the ONLY thing keeping the old paths alive, so external plugins have time to update. It is deliberately a single, unsquashed commit: git revert <this sha> removes every shim, stub and manifest at once on the announced date. Nothing in-tree may depend on these pointers: scripts/check_compat_pointers.py (wired into lint.yml) fails CI if it does. What it adds (see COMPAT_MANIFEST.md, compat_manifest.json): - 332 facade modules get one delimited `PLUGIN-COMPAT` block appended at the end of the file - 1,172 moved names resolved lazily via a module `__getattr__` (PEP 562) — never a top-level import, so no import cycles; facades that already had `__getattr__` get a chained one - 592 third-party/stdlib names the old modules used to expose, with their original import statements - 266 public definitions that had been deleted as unused, restored byte-for-byte from the pre-decomposition tree (+40 private helpers and 16 imports pulled in only because a restored definition needs them) - 3 deleted modules recreated as re-export stubs (gateway/startup_watchdog, hermes_cli/observability/ relay_runtime, tools/environments/modal_utils) - private names (`_x`) get no pointer: they were never API (3,792 skipped) Verified: all 335 touched modules import under a fresh HERMES_HOME and every manifest name resolves; the lint reports zero in-tree uses; ruff clean; targeted suites unchanged.
649 lines
36 KiB
Python
649 lines
36 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Delegate Tool -- Subagent Architecture
|
|
|
|
Spawns child AIAgent instances with a fresh conversation, their own task_id
|
|
(terminal session, file-ops cache), the parent's toolsets minus child-blocked
|
|
tools, and a focused system prompt built from goal + context. Single-task and
|
|
batch (parallel) modes; top-level model calls run in the background while
|
|
orchestrator children wait for their own workers. The parent only ever sees
|
|
the delegation call and the summary result, never the child's intermediate
|
|
tool calls or reasoning.
|
|
"""
|
|
|
|
import logging
|
|
import time
|
|
import weakref
|
|
from typing import Any, Dict, List, Optional
|
|
|
|
from tools.terminal_tool import set_approval_callback as _set_subagent_approval_cb # noqa: F401 (used via _ChildRun.await_child)
|
|
from utils import is_truthy_value
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# The delegate_tool_* siblings hold the pieces split out of this module; every name callers or patching tests reach as
|
|
# ``tools.delegate_tool.<name>`` is re-imported here. Mutable flag globals live only in their owning module.
|
|
from tools.delegate_tool_child_run import ( # noqa: F401
|
|
_ChildRun, _attach_child, _build_result_entry, _dump_subagent_timeout_diagnostic, _fabricated_entry,
|
|
_lease_child_credential, _merge_late_steer, _register_child, _start_heartbeat, _validate_child_output_schema,
|
|
)
|
|
from tools.delegate_tool_config import ( # noqa: F401
|
|
_DEFAULT_MAX_CONCURRENT_CHILDREN, _get_child_timeout, _get_max_async_children, _get_max_concurrent_children,
|
|
_get_max_spawn_depth, _get_orchestrator_enabled, _get_subagent_approval_callback, _get_worktree_isolation,
|
|
_inherit_parent_capabilities, _load_config, _merge_request_overrides, _resolve_child_credential_pool,
|
|
_resolve_child_runtime, _resolve_delegation_credentials, _subagent_auto_approve, _subagent_auto_deny,
|
|
)
|
|
from tools.delegate_tool_dispatch import _Batch, _announce_batch, _capture_origin, _run_batch
|
|
from tools.delegate_tool_progress import ( # noqa: F401
|
|
DelegateEvent, SUBAGENT_FAILURE_STATUSES, _batch_prefix, _build_child_progress_callback,
|
|
_build_child_system_prompt, _clean_error_text, _emit_parent_console, _quiet, _resolve_workspace_hint,
|
|
_safe_progress, format_batch_tag, format_subagent_failure_line,
|
|
)
|
|
from tools.delegate_tool_registry import ( # noqa: F401
|
|
_CONTROL_ACTIONS, _active_subagents, _active_subagents_lock, _capture_gateway_steer_authority,
|
|
_handle_control_action, _is_descendant_of, _owns_subagent_record, _register_subagent, _unregister_subagent,
|
|
get_subagent_attribution, interrupt_subagent, is_spawn_paused, list_active_subagents, set_spawn_paused,
|
|
steer_subagent,
|
|
)
|
|
from tools.delegate_tool_tasks import _coerce_task_schemas, _normalize_task_list
|
|
from tools.delegate_tool_toolsets import ( # noqa: F401
|
|
DELEGATE_BLOCKED_TOOLS, _expand_parent_toolsets, _resolve_child_toolsets, _strip_blocked_tools,
|
|
)
|
|
from tools.delegate_tool_results import ( # noqa: F401
|
|
_apply_summary_budget, _build_child_preserving_parent_tools, _run_child_lifecycle, _summarize_tool_arguments,
|
|
)
|
|
|
|
_ROLES = frozenset({"leaf", "orchestrator"})
|
|
|
|
# Nested delegation is granted by depth/role in _build_child_agent, never by the
|
|
# model naming toolsets (there is no model-facing toolsets argument).
|
|
def _normalize_role(r: Optional[str]) -> str:
|
|
"""'leaf' | 'orchestrator'; None/empty/unknown -> 'leaf' (unknown warns)."""
|
|
r_norm = str(r).strip().lower() if r else "leaf"
|
|
if r_norm not in _ROLES:
|
|
logger.warning("Unknown delegate_task role=%r, coercing to 'leaf'", r)
|
|
return "leaf"
|
|
return r_norm
|
|
|
|
DEFAULT_MAX_ITERATIONS = 250
|
|
_HEARTBEAT_INTERVAL = 30 # seconds between parent activity heartbeats during delegation
|
|
# Stale-heartbeat thresholds (cycles of _HEARTBEAT_INTERVAL with no progress). Progress = iteration, current_tool OR
|
|
# last_activity_ts advancing; an in-flight model wait refreshes last_activity_ts, so slow models are not "idle". Idle
|
|
# stays tight so a truly wedged child doesn't mask the gateway timeout; in-tool is much higher so legitimately long
|
|
# tools can finish.
|
|
_HEARTBEAT_STALE_CYCLES_IDLE = 15 # 450s idle between turns → stale
|
|
_HEARTBEAT_STALE_CYCLES_IN_TOOL = 40 # 1200s stuck on same tool → stale
|
|
|
|
def check_delegate_requirements() -> bool:
|
|
"""Delegation has no external requirements -- always available."""
|
|
return True
|
|
|
|
|
|
def _open_child_session_db(parent_agent) -> Any:
|
|
"""DEDICATED SessionDB handle for the child, or None: the parent's handle can be closed by its own lifecycle while
|
|
a background child still flushes (transcript silently dropped). It MUST open the same db FILE as the parent's
|
|
handle (non-launch profiles), else lineage / session_search break; released by the child's close() via
|
|
_owns_session_db."""
|
|
# Each child gets a DEDICATED SessionDB connection instead of the parent's live object. The parent's
|
|
# handle is owned by the parent's lifecycle (cron run_job's finally block, gateway session end, /new)
|
|
# and can be closed while a fire-and-forget background child is still flushing on a daemon thread —
|
|
# every subsequent flush then hits the closed handle and the child's transcript is silently dropped
|
|
# (#81267). It MUST point at the same database FILE as the parent's handle: parents can hold non-default
|
|
# per-profile handles (tui_gateway opens SessionDB(db_path=<profile>/ state.db) for non-launch
|
|
# profiles), and a bare SessionDB() would write the child's transcript into the launch profile's db,
|
|
# breaking parent_session_id lineage and session_search. AsyncSessionDB wrappers (gateway) forward
|
|
# .db_path via __getattr__, so this works through them.
|
|
parent_session_db = getattr(parent_agent, "_session_db", None)
|
|
if parent_session_db is None:
|
|
return None
|
|
with _quiet("subagent: failed to open dedicated SessionDB; child persistence disabled", exc_info=True):
|
|
from hermes_state_registry import acquire
|
|
_parent_db_path = getattr(parent_session_db, "db_path", None)
|
|
return acquire(_parent_db_path) if _parent_db_path is not None else acquire()
|
|
return None
|
|
|
|
def _build_child_agent(
|
|
task_index: int,
|
|
goal: str,
|
|
context: Optional[str],
|
|
toolsets: Optional[List[str]],
|
|
model: Optional[str],
|
|
max_iterations: int,
|
|
task_count: int,
|
|
parent_agent,
|
|
# Credential overrides from delegation config
|
|
override_provider: Optional[str] = None,
|
|
override_base_url: Optional[str] = None,
|
|
override_api_key: Optional[str] = None,
|
|
override_api_mode: Optional[str] = None,
|
|
override_request_overrides: Optional[Dict[str, Any]] = None,
|
|
override_max_tokens: Optional[int] = None,
|
|
# ACP transport overrides from trusted delegation config.
|
|
override_acp_command: Optional[str] = None,
|
|
override_acp_args: Optional[List[str]] = None,
|
|
# Legacy; accepted for wire compat but ignored (capability is depth-derived).
|
|
role: str = "leaf",
|
|
):
|
|
"""Build (don't run) a child AIAgent on the main thread. override_* (from delegation config) replace parent
|
|
inheritance so children can run on a different provider:model pair."""
|
|
import uuid as _uuid
|
|
from run_agent import AIAgent
|
|
from agent.delegation_context import delegated_child_context
|
|
# Role is depth-derived: a child may delegate iff the kill switch is on and
|
|
# depth budget remains below max_spawn_depth. The `role` arg is ignored.
|
|
child_depth = getattr(parent_agent, "_delegate_depth", 0) + 1
|
|
max_spawn = _get_max_spawn_depth()
|
|
effective_role = "orchestrator" if _get_orchestrator_enabled() and child_depth < max_spawn else "leaf"
|
|
|
|
# One subagent_id shared by the progress callback, spawn_requested event and
|
|
# the live registry; parent_id is set when THIS parent is itself a subagent.
|
|
subagent_id = f"sa-{task_index}-{_uuid.uuid4().hex[:8]}"
|
|
parent_subagent_id = getattr(parent_agent, "_subagent_id", None)
|
|
|
|
delegation_cfg = _load_config()
|
|
child_toolsets, child_disabled_toolsets = _resolve_child_toolsets(parent_agent, toolsets, effective_role)
|
|
child_prompt = _build_child_system_prompt(
|
|
goal, context, workspace_path=_resolve_workspace_hint(parent_agent), role=effective_role,
|
|
max_spawn_depth=max_spawn, child_depth=child_depth,
|
|
)
|
|
parent_api_key = getattr(parent_agent, "api_key", None)
|
|
if (not parent_api_key) and hasattr(parent_agent, "_client_kwargs"):
|
|
parent_api_key = parent_agent._client_kwargs.get("api_key")
|
|
|
|
# Shared ref: session_id once the child exists, delegation_id once
|
|
# delegate_task stamps it — both ride on every relayed event.
|
|
child_session_ref: Dict[str, Any] = {}
|
|
child_progress_cb = _build_child_progress_callback(
|
|
task_index, goal, parent_agent, task_count, subagent_id=subagent_id, parent_id=parent_subagent_id,
|
|
depth=max(0, child_depth - 1), # 0 = first-level child for the UI
|
|
model=model or getattr(parent_agent, "model", None), toolsets=child_toolsets, session_ref=child_session_ref,
|
|
)
|
|
rt = _resolve_child_runtime(
|
|
parent_agent, delegation_cfg, parent_api_key, model=model, override_provider=override_provider,
|
|
override_base_url=override_base_url, override_api_key=override_api_key, override_api_mode=override_api_mode,
|
|
override_max_tokens=override_max_tokens, override_acp_command=override_acp_command,
|
|
override_acp_args=override_acp_args,
|
|
)
|
|
if override_request_overrides is not None:
|
|
# honored whenever set, incl. the inherit branch where
|
|
# _resolve_delegation_credentials already merged OVER the parent's
|
|
request_overrides = dict(override_request_overrides)
|
|
else:
|
|
request_overrides = {} if override_provider else dict(getattr(parent_agent, "request_overrides", {}) or {})
|
|
parent_sid = getattr(parent_agent, "session_id", None)
|
|
child_session_db = _open_child_session_db(parent_agent)
|
|
with delegated_child_context():
|
|
try:
|
|
child = AIAgent(
|
|
**rt, max_iterations=max_iterations, prefill_messages=getattr(parent_agent, "prefill_messages", None),
|
|
enabled_toolsets=child_toolsets, disabled_toolsets=child_disabled_toolsets, quiet_mode=True,
|
|
ephemeral_system_prompt=child_prompt, log_prefix=f"[subagent-{task_index}]", platform="subagent",
|
|
skip_context_files=True, skip_memory=True, clarify_callback=None,
|
|
thinking_callback=(
|
|
(lambda text: _safe_progress(child_progress_cb, "_thinking", text) if text else None)
|
|
if child_progress_cb else None
|
|
),
|
|
session_db=child_session_db, parent_session_id=parent_sid, request_overrides=request_overrides,
|
|
tool_progress_callback=child_progress_cb,
|
|
iteration_budget=None, # fresh budget per subagent
|
|
)
|
|
except BaseException:
|
|
# No child close() will ever run: release the dedicated handle here.
|
|
if child_session_db is not None:
|
|
with _quiet(None):
|
|
from hermes_state_registry import release_or_close
|
|
release_or_close(child_session_db)
|
|
raise
|
|
child._print_fn = getattr(parent_agent, "_print_fn", None)
|
|
if child_session_db is not None:
|
|
child._owns_session_db = True # released by the child's close(), never by the parent
|
|
# Ownership transfer for the dedicated handle: the child's close() must release it (nothing else holds a
|
|
# reference), and no parent teardown can close it out from under a background child (#81267).
|
|
child_session_ref["session_id"] = getattr(child, "session_id", "") or ""
|
|
child._progress_identity_ref = child_session_ref
|
|
child._delegate_depth, child._delegate_role = child_depth, effective_role # post-degrade role
|
|
child._subagent_id, child._parent_subagent_id = subagent_id, parent_subagent_id
|
|
# Ownership chain for action=list/steer/stop; weakref so a finished parent
|
|
# can be collected while a detached child record lingers in the registry.
|
|
try:
|
|
child._delegate_parent_ref = weakref.ref(parent_agent)
|
|
except TypeError:
|
|
child._delegate_parent_ref = None # non-weakref-able test doubles
|
|
# Sidebar marker: subagent sessions stay out of session pickers even when a
|
|
# parent delete orphans them (mirrors /branch's ``_branched_from``).
|
|
if parent_sid and getattr(child, "_session_init_model_config", None) is not None:
|
|
child._session_init_model_config["_delegate_from"] = parent_sid
|
|
# Shared pool lets children rotate credentials on rate limits.
|
|
child_pool = _resolve_child_credential_pool(rt["provider"], parent_agent, rt["base_url"])
|
|
if child_pool is not None:
|
|
child._credential_pool = child_pool
|
|
|
|
_attach_child(parent_agent, child) # interrupt propagation
|
|
# spawn_requested now — the child may queue for seconds when the pool is
|
|
# saturated — then the subagent_start lifecycle hook.
|
|
_safe_progress(child_progress_cb, "subagent.spawn_requested", preview=goal)
|
|
with _quiet("subagent_start hook invocation failed", exc_info=True):
|
|
from hermes_cli.lifecycle import invoke_hook as _invoke_hook
|
|
_invoke_hook(
|
|
"subagent_start", parent_session_id=parent_sid,
|
|
parent_turn_id=getattr(parent_agent, "_current_turn_id", "") or "", parent_subagent_id=parent_subagent_id,
|
|
child_session_id=getattr(child, "session_id", None), child_subagent_id=subagent_id,
|
|
child_role=effective_role, child_goal=goal,
|
|
)
|
|
return child
|
|
|
|
def _run_single_child(
|
|
task_index: int, goal: str, child=None, parent_agent=None, *, owner_session_id: Optional[str] = None,
|
|
owner_transport: Any = None, owner_session_record: Any = None, **_kwargs,
|
|
) -> Dict[str, Any]:
|
|
"""Run a pre-built child agent (called from a worker thread) and return its result entry.
|
|
|
|
Contract, derived from the child's structured completion fields:
|
|
status ∈ {completed, interrupted, failed} — a structured failure
|
|
(failed=True / non-empty error) or an invalid terminal state
|
|
is "failed" even when a summary exists.
|
|
exit_reason ∈ {completed, max_iterations, interrupted, error} —
|
|
"max_iterations" only for genuine budget exhaustion
|
|
(completed=False with no failure fields), never for errors.
|
|
truncated == (exit_reason == "max_iterations").
|
|
|
|
* ``"completed"`` — normal finish. See #97655.
|
|
"""
|
|
child_progress_cb = getattr(child, "tool_progress_callback", None)
|
|
child_pool, leased_cred_id = _lease_child_credential(child)
|
|
# Heartbeat keeps the parent's _last_activity_ts moving so the gateway inactivity timeout doesn't fire while the
|
|
# child works; it stops itself once the child looks stale (see _HEARTBEAT_STALE_CYCLES_*).
|
|
heartbeat = _start_heartbeat(child, parent_agent, task_index)
|
|
# TUI/RPC registry entry (kill/pause/status by subagent_id); None for test
|
|
# doubles without a stable id. Unregistered in the finally block.
|
|
_subagent_id = _register_child(
|
|
child, parent_agent, goal, owner_session_id=owner_session_id, owner_transport=owner_transport,
|
|
owner_session_record=owner_session_record,
|
|
)
|
|
run = _ChildRun(child, parent_agent, task_index, goal, _subagent_id, child_progress_cb)
|
|
# Set when a timed-out Future still owns the child: closing it from this
|
|
# thread before the worker settles races the conversation's finally path.
|
|
_child_close_deferred = False
|
|
try:
|
|
heartbeat.start()
|
|
_safe_progress(child_progress_cb, "subagent.start", preview=goal)
|
|
run.seed_workspace()
|
|
result, failure_entry, _child_close_deferred = run.await_child()
|
|
if failure_entry is not None:
|
|
return failure_entry
|
|
|
|
schema = _validate_child_output_schema(child, result, task_index, run.child_task_id, run.relay_text)
|
|
_merge_late_steer(result, _subagent_id, child)
|
|
# Flush any remaining batched progress to gateway
|
|
if child_progress_cb and hasattr(child_progress_cb, "_flush"):
|
|
with _quiet("Progress callback flush failed: %s"):
|
|
child_progress_cb._flush()
|
|
|
|
duration = run.elapsed()
|
|
entry = _build_result_entry(child, result, task_index, duration, schema)
|
|
run.append_sibling_write_reminder(entry)
|
|
run.emit_complete(result, entry, duration)
|
|
return run.attach_worktree(entry)
|
|
except Exception as exc:
|
|
# Close steer acceptance before any completion callback (see _merge_late_steer).
|
|
_late_pending_steer = run.close_steering()
|
|
logging.exception(f"[subagent-{task_index}] failed")
|
|
# Entry status "error" (contract), progress event status "failed" (UI vocabulary).
|
|
return run.finish_failed(
|
|
_fabricated_entry(task_index, "error", str(exc), child, run.elapsed()), _late_pending_steer,
|
|
preview=str(exc), summary=str(exc), status="failed",
|
|
)
|
|
finally:
|
|
run.cleanup(heartbeat=heartbeat, child_pool=child_pool, leased_cred_id=leased_cred_id, close_deferred=_child_close_deferred)
|
|
|
|
|
|
def _build_children(
|
|
task_list: List[Dict[str, Any]], task_schemas: List[Optional[Dict[str, Any]]], creds: Dict[str, Any], *,
|
|
top_role: str, max_iterations: int, parent_agent, live_deleg_id: Optional[str], live_writers: list,
|
|
) -> tuple[List[tuple], Optional[str]]:
|
|
"""Build every child on the main thread (construction is not thread-safe);
|
|
``(children, None)`` or ``([], error)`` on an explicit-pin preflight failure."""
|
|
from tools.delegation_live_log import wrap_progress_callback
|
|
from tools.delegation_output_schema import append_output_contract
|
|
overrides = {
|
|
"override_provider": creds["provider"], "override_base_url": creds["base_url"],
|
|
"override_api_key": creds["api_key"], "override_api_mode": creds["api_mode"],
|
|
"override_request_overrides": creds.get("request_overrides"),
|
|
"override_max_tokens": creds.get("max_output_tokens"), "override_acp_command": creds.get("command"),
|
|
"override_acp_args": creds.get("args"),
|
|
}
|
|
children = []
|
|
for i, t in enumerate(task_list):
|
|
_task_schema = task_schemas[i] if i < len(task_schemas) else None
|
|
_child_context = t.get("context")
|
|
if _task_schema is not None:
|
|
_child_context = append_output_contract(_child_context, _task_schema)
|
|
try:
|
|
child = _build_child_preserving_parent_tools(
|
|
task_index=i, goal=t["goal"], context=_child_context,
|
|
toolsets=None, # always inherit the parent's toolsets
|
|
model=creds["model"], max_iterations=max_iterations, task_count=len(task_list),
|
|
parent_agent=parent_agent, role=_normalize_role(t.get("role") or top_role), **overrides,
|
|
)
|
|
except ValueError as exc:
|
|
return [], str(exc)
|
|
if _task_schema is not None:
|
|
with _quiet("Could not attach output schema to child %d", i):
|
|
child._delegate_output_schema = _task_schema
|
|
# Tee progress events into the live transcript (wrapper keeps the
|
|
# _flush contract and swallows writer failures).
|
|
_writer = live_writers[i] if i < len(live_writers) else None
|
|
if _writer is not None:
|
|
child.tool_progress_callback = wrap_progress_callback(getattr(child, "tool_progress_callback", None), _writer)
|
|
child._live_transcript_path = str(_writer.path)
|
|
if live_deleg_id:
|
|
setattr(child, "_delegation_id", live_deleg_id)
|
|
_ident_ref = getattr(child, "_progress_identity_ref", None)
|
|
if isinstance(_ident_ref, dict):
|
|
_ident_ref["delegation_id"] = live_deleg_id
|
|
children.append((i, t, child))
|
|
return children, None
|
|
|
|
|
|
def delegate_task(
|
|
goal: Optional[str] = None, context: Optional[str] = None, tasks: Optional[List[Dict[str, Any]]] = None,
|
|
max_iterations: Optional[int] = None, role: Optional[str] = None, background: Optional[bool] = None,
|
|
output_schema: Optional[Dict[str, Any]] = None, action: Optional[str] = None, subagent_id: Optional[str] = None,
|
|
message: Optional[str] = None, parent_agent=None, credentials_cfg: Optional[Dict[str, Any]] = None,
|
|
) -> str:
|
|
"""Spawn child agents (single ``goal`` or ``tasks=[...]`` batch) or control running ones. ``action``
|
|
list/steer/stop run synchronously and bypass the pause gate, depth limit and async dispatch. ``role`` is legacy
|
|
(per-task beats top-level; capability is depth-derived). Returns JSON with one results entry per task, or a
|
|
dispatch handle when running in the background."""
|
|
if parent_agent is None:
|
|
return tool_error("delegate_task requires a parent agent context.")
|
|
|
|
normalized_action = (action or "").strip().lower()
|
|
if normalized_action in _CONTROL_ACTIONS:
|
|
return _handle_control_action(normalized_action, subagent_id, message, parent_agent)
|
|
if normalized_action and normalized_action != "spawn":
|
|
return tool_error(f"Unknown action '{action}'. Use spawn (default), list, steer, or stop.")
|
|
|
|
# Operator kill switch (TUI / delegation.pause RPC): blocks NEW spawns only.
|
|
if is_spawn_paused():
|
|
return tool_error(
|
|
"Delegation spawning is paused. Clear the pause via the TUI "
|
|
"(`p` in /agents) or the `delegation.pause` RPC before retrying."
|
|
)
|
|
|
|
top_role = _normalize_role(role)
|
|
# background applies to single tasks AND batches: a batch is ONE async unit
|
|
# that joins on every child and re-enters as a single consolidated message.
|
|
background = is_truthy_value(background, default=False) if background is not None else False
|
|
|
|
depth = getattr(parent_agent, "_delegate_depth", 0)
|
|
max_spawn = _get_max_spawn_depth()
|
|
if depth >= max_spawn:
|
|
return tool_error(
|
|
f"Delegation depth limit reached (depth={depth}, max_spawn_depth={max_spawn}). Raise "
|
|
f"delegation.max_spawn_depth in config.yaml if deeper nesting is required (no hard ceiling, but each level "
|
|
f"multiplies API cost)."
|
|
)
|
|
|
|
cfg = _load_config()
|
|
default_max_iter = cfg.get("max_iterations", DEFAULT_MAX_ITERATIONS)
|
|
# Caller-supplied max_iterations is ignored: the config value is authoritative
|
|
# so budgets stay predictable (kwarg kept for internal callers/tests).
|
|
if max_iterations is not None and max_iterations != default_max_iter:
|
|
logger.debug(
|
|
"delegate_task: ignoring caller-supplied max_iterations=%s; using delegation.max_iterations=%s from config",
|
|
max_iterations, default_max_iter,
|
|
)
|
|
# credentials_cfg (internal callers only, e.g. /review → auxiliary.review) is
|
|
# a per-call override shaped like the delegation config section.
|
|
try:
|
|
creds = _resolve_delegation_credentials(credentials_cfg if credentials_cfg else cfg, parent_agent)
|
|
except ValueError as exc:
|
|
# Explicit-pin preflight failures (e.g. pinned delegation.command missing from PATH) refuse the
|
|
# spawn loudly (#80450).
|
|
return tool_error(str(exc))
|
|
max_children = _get_max_concurrent_children()
|
|
task_list, err = _normalize_task_list(goal, context, tasks, output_schema, top_role, max_children)
|
|
if not err:
|
|
task_schemas, err = _coerce_task_schemas(task_list, output_schema)
|
|
if err:
|
|
return tool_error(err)
|
|
|
|
overall_start = time.monotonic()
|
|
# Live transcripts: cache/delegation/live/<id>/task-<n>.log per task, a side channel with zero effect on message
|
|
# content or prompt caching. Best-effort: on failure live_paths is empty and delegation proceeds.
|
|
from tools.delegation_live_log import create_live_transcripts
|
|
live_deleg_id, live_writers, live_paths = create_live_transcripts(
|
|
task_list, context, model=creds.get("model"), provider=creds.get("provider")
|
|
)
|
|
_announce_batch(parent_agent, len(task_list), live_deleg_id)
|
|
origin = _capture_origin()
|
|
|
|
children, err = _build_children(
|
|
task_list, task_schemas, creds, top_role=top_role, max_iterations=default_max_iter, parent_agent=parent_agent,
|
|
live_deleg_id=live_deleg_id, live_writers=live_writers,
|
|
)
|
|
if err:
|
|
return tool_error(err)
|
|
batch = _Batch(
|
|
task_list, children, parent_agent, creds, context, top_role, max_children,
|
|
live_deleg_id, live_writers, live_paths, *origin, overall_start,
|
|
)
|
|
return _run_batch(batch, background)
|
|
|
|
|
|
# ── OpenAI function-calling schema ──────────────────────────────────────────
|
|
|
|
def _build_top_level_description() -> str:
|
|
"""delegate_task description: ONLY guidance stated nowhere else in the schema
|
|
(limits live in the 'tasks' parameter description, rebuilt per get_definitions())."""
|
|
try:
|
|
orchestration_available = _get_max_spawn_depth() >= 2 and _get_orchestrator_enabled()
|
|
except Exception:
|
|
orchestration_available = False
|
|
# Mention recursion only where it's actually available. send_message is deliberately not named (gateway-internal
|
|
# vocabulary); model_tools session-filters the list to tools the session has.
|
|
if orchestration_available:
|
|
restrictions_rule = (
|
|
"- Children cannot call clarify, memory, or cronjob.\n"
|
|
f"- Children can themselves delegate while depth remains (max_spawn_depth={_get_max_spawn_depth()}); the "
|
|
"runtime derives this from depth automatically.\n"
|
|
)
|
|
else:
|
|
restrictions_rule = "- Children cannot call delegate_task, clarify, memory, or cronjob.\n"
|
|
return _DESCRIPTION_HEAD + restrictions_rule + _DESCRIPTION_TAIL
|
|
|
|
_DESCRIPTION_HEAD = (
|
|
"Spawn subagents in isolated contexts; each gets its own conversation, terminal session, and toolset, and only its "
|
|
"final summary returns to you. Pass every task in `tasks` — one entry spawns one subagent, several run in parallel "
|
|
"(limit in the tasks description).\n\n"
|
|
"Runs in the background: dispatch returns immediately with live transcript paths, and the completed result (one "
|
|
"consolidated message, results in task order) re-enters the conversation on its own. Do NOT wait or poll; continue "
|
|
"other work. While children run, `action` (list/steer/stop) controls them live — steer when a transcript shows a "
|
|
"child drifting.\n\n"
|
|
"USE FOR: reasoning-heavy subtasks, work that would flood your context with intermediate data, or independent "
|
|
"parallel workstreams.\n"
|
|
"DO NOT USE FOR (use these instead):\n"
|
|
"- Mechanical multi-step work with no reasoning needed -> execute_code\n"
|
|
"- A single tool call -> call the tool directly\n"
|
|
"- Tasks needing user interaction -> subagents cannot ask questions\n"
|
|
"- Durable work that must survive this session -> cronjob or terminal(background=True, notify=True); /stop, /new, "
|
|
"or process exit discards running subagents.\n\n"
|
|
"RULES:\n"
|
|
"- Children know nothing of this conversation: pass everything needed via 'context', including any required "
|
|
"output language, tone, or style (e.g. \"respond in Chinese\").\n"
|
|
"- Child summaries are SELF-REPORTS, not verified facts: a child claiming \"uploaded successfully\" or "
|
|
"\"file written\" may be wrong. For external side effects (uploads, remote writes, publishing), require a "
|
|
"verifiable handle (URL, ID, absolute path) and verify it yourself before telling the user the operation "
|
|
"succeeded.\n"
|
|
)
|
|
_DESCRIPTION_TAIL = (
|
|
"- Children inherit the parent model unless pinned via delegation.provider / delegation.model in config.yaml."
|
|
)
|
|
|
|
def _build_tasks_param_description() -> str:
|
|
"""Compose the 'tasks' parameter description with current concurrency limit."""
|
|
try:
|
|
max_children = _get_max_concurrent_children()
|
|
except Exception:
|
|
max_children = _DEFAULT_MAX_CONCURRENT_CHILDREN
|
|
return (
|
|
f"The task(s), up to {max_children} in parallel for this user (set "
|
|
"via delegation.max_concurrent_children). Each entry spawns one "
|
|
"subagent with isolated context and terminal session; a single task "
|
|
"is a one-entry array. Required when spawning."
|
|
)
|
|
|
|
def _build_dynamic_schema_overrides() -> dict:
|
|
"""Per-call schema overrides (ToolEntry.dynamic_schema_overrides): every
|
|
get_definitions() pass rewrites the descriptions to the user's actual limits."""
|
|
overrides_params = {**DELEGATE_TASK_SCHEMA["parameters"]}
|
|
# Copy properties so the static schema dict is never mutated.
|
|
overrides_params["properties"] = {k: dict(v) for k, v in DELEGATE_TASK_SCHEMA["parameters"]["properties"].items()}
|
|
overrides_params["properties"]["tasks"]["description"] = _build_tasks_param_description()
|
|
|
|
return {"description": _build_top_level_description(), "parameters": overrides_params}
|
|
|
|
def _p(type_: str, description: str, **extra) -> dict:
|
|
return {"type": type_, **extra, "description": description}
|
|
|
|
DELEGATE_TASK_SCHEMA = {
|
|
"name": "delegate_task",
|
|
# description / tasks.description are placeholders: the real text is built per get_definitions() call by
|
|
# _build_dynamic_schema_overrides() so the model sees the user's actual max_concurrent_children / max_spawn_depth.
|
|
# Lazy (not at import) so cli.CLI_CONFIG isn't forced to load before the test conftest redirects HERMES_HOME.
|
|
"description": (
|
|
"Spawn one or more subagents in isolated contexts. "
|
|
"Description is rebuilt at every get_definitions() call to reflect the user's current delegation limits."
|
|
),
|
|
"parameters": {
|
|
"type": "object",
|
|
"properties": {
|
|
# The handler also accepts the legacy single-goal shape (top-level `goal`/`context`/`output_schema`),
|
|
# wrapped into a one-entry batch at dispatch, and a per-task `role` (legacy, ignored: capability is
|
|
# depth-derived). Both unadvertised on purpose (old transcripts only); do not re-add. No maxItems — the
|
|
# runtime limit (delegation.max_concurrent_children) is enforced with a clear error in delegate_task().
|
|
"tasks": {
|
|
"type": "array",
|
|
"minItems": 1,
|
|
"items": {
|
|
"type": "object",
|
|
"properties": {
|
|
"goal": _p(
|
|
"string",
|
|
"What this subagent should accomplish. Be specific and self-contained — it knows "
|
|
"nothing about your conversation history.",
|
|
),
|
|
"context": _p(
|
|
"string",
|
|
"Background THIS child needs: file paths, error messages, constraints. Each child "
|
|
"sees only its own context — repeat shared background in every task that needs it.",
|
|
),
|
|
"output_schema": _p(
|
|
"object",
|
|
"Optional JSON Schema this child's final answer must validate against (told to the "
|
|
"child up front; parent validates with one bounded correction retry; result gains "
|
|
"schema_valid, plus schema_errors on failure). Keep it forgiving — require only "
|
|
"fields you will read.",
|
|
),
|
|
},
|
|
"required": ["goal"],
|
|
},
|
|
"description": "(rebuilt at get_definitions() time)",
|
|
},
|
|
# `background` (bool) is also accepted — DEPRECATED, ignored: top-level
|
|
# delegations always run in the background. Unadvertised; do not re-add.
|
|
"action": _p(
|
|
"string",
|
|
"Default 'spawn'. Live control of running children: "
|
|
"'list' = ids/goals/status/transcripts; 'steer' = queue "
|
|
"course-correction text into one child (subagent_id + "
|
|
"message) without stopping it; 'stop' = end one child "
|
|
"early (subagent_id; partial result still returns). "
|
|
"Control actions return immediately; goal/tasks are ignored unless spawning.",
|
|
enum=["spawn", "list", "steer", "stop"],
|
|
),
|
|
"subagent_id": _p("string", "Target for action='steer'/'stop' (ids from the spawn response or action='list')."),
|
|
"message": _p(
|
|
"string",
|
|
"For action='steer': the course correction, appended to "
|
|
"the child's next tool result mid-run. Be directive and specific.",
|
|
),
|
|
},
|
|
"required": [],
|
|
},
|
|
}
|
|
|
|
|
|
# --- Registry ---
|
|
from tools.registry import registry, tool_error
|
|
|
|
def _model_background_value(args: dict, parent_agent=None) -> bool:
|
|
"""Background flag for the MODEL-facing dispatch path (registry fallback). Top-level delegations always run in the
|
|
background — the model does not choose — for single tasks and fan-out batches alike (one async unit, one
|
|
consolidated result); an orchestrator subagent (depth > 0) is the exception since it needs its workers' results
|
|
within its own turn. The live path is ``run_agent._dispatch_delegate_task``; this mirrors it for the rare case
|
|
the intercept is bypassed. Direct Python callers keep the synchronous default."""
|
|
return not getattr(parent_agent, "_delegate_depth", 0) > 0
|
|
|
|
_MODEL_HIDDEN_TASK_FIELDS = {"acp_command", "acp_args"}
|
|
|
|
def _strip_model_hidden_task_fields(tasks: Any) -> Any:
|
|
"""Drop trusted-config-only task fields from model-supplied tasks (same list object back when nothing changed)."""
|
|
if not isinstance(tasks, list) or not any(isinstance(t, dict) and _MODEL_HIDDEN_TASK_FIELDS & t.keys() for t in tasks):
|
|
return tasks
|
|
return [{k: v for k, v in t.items() if k not in _MODEL_HIDDEN_TASK_FIELDS} if isinstance(t, dict) else t for t in tasks]
|
|
|
|
|
|
registry.register(
|
|
name="delegate_task",
|
|
toolset="delegation",
|
|
schema=DELEGATE_TASK_SCHEMA,
|
|
handler=lambda args, **kw: delegate_task(
|
|
goal=args.get("goal"), context=args.get("context"), tasks=_strip_model_hidden_task_fields(args.get("tasks")),
|
|
max_iterations=args.get("max_iterations"), role=args.get("role"),
|
|
background=_model_background_value(args, kw.get("parent_agent")), output_schema=args.get("output_schema"),
|
|
action=args.get("action"), subagent_id=args.get("subagent_id"), message=args.get("message"),
|
|
parent_agent=kw.get("parent_agent"),
|
|
),
|
|
check_fn=check_delegate_requirements,
|
|
emoji="🔀",
|
|
dynamic_schema_overrides=_build_dynamic_schema_overrides,
|
|
)
|
|
|
|
|
|
# ---- BEGIN PLUGIN-COMPAT (revert-scheduled; see COMPAT_MANIFEST.md) ----
|
|
# Names external plugins imported from this module before the Sep 2026 decomposition.
|
|
# Internal code MUST NOT use these (scripts/check_compat_pointers.py fails CI if it does).
|
|
# The whole block is removed by reverting the commit that added it.
|
|
from concurrent.futures import TimeoutError as FuturesTimeoutError # noqa: F401,E402
|
|
import contextvars # noqa: F401,E402
|
|
import enum # noqa: F401,E402
|
|
import json # noqa: F401,E402
|
|
import os # noqa: F401,E402
|
|
import re # noqa: F401,E402
|
|
import threading # noqa: F401,E402
|
|
from urllib.parse import urlsplit # noqa: F401,E402
|
|
from urllib.parse import urlunsplit # noqa: F401,E402
|
|
|
|
|
|
_PLUGIN_COMPAT_LAZY = {
|
|
'DEFAULT_CHILD_TIMEOUT': ('tools.delegate_tool_config', 'DEFAULT_CHILD_TIMEOUT'),
|
|
'DEFAULT_MAX_SUMMARY_CHARS': ('tools.delegate_tool_results', 'DEFAULT_MAX_SUMMARY_CHARS'),
|
|
'DEFAULT_TOOLSETS': ('tools.delegate_tool_toolsets', 'DEFAULT_TOOLSETS'),
|
|
'MAX_DEPTH': ('tools.delegate_tool_config', 'MAX_DEPTH'),
|
|
'TOOLSETS': ('toolsets', 'TOOLSETS'),
|
|
'base_url_hostname': ('utils', 'base_url_hostname'),
|
|
'file_state': ('tools', 'file_state'),
|
|
'request_hard_interrupt': ('agent.interrupt_compat', 'request_hard_interrupt'),
|
|
}
|
|
|
|
|
|
def __getattr__(name): # PEP 562 — lazy so no import cycles
|
|
target = _PLUGIN_COMPAT_LAZY.get(name)
|
|
if target is None:
|
|
raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
|
|
import importlib
|
|
return getattr(importlib.import_module(target[0]), target[1])
|
|
# ---- END PLUGIN-COMPAT ----
|