hermes-admin/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 28 Packages Projects Releases Wiki Activity

refactor(delegate): fold _construct_child_agent/_announce_child_spawn into _build_child_agent; _quiet best-effort sites

Browse Source
This commit is contained in:
Teknium
2026-09-02 18:04:43 -07:00
parent b18228f377
commit 378692be4f
2 changed files with 78 additions and 142 deletions
Download Patch File Download Diff File Expand all files Collapse all files

213
tools/delegate_tool.py
View File

@@ -26,7 +26,7 @@ logger = logging.getLogger(__name__)
# moved name is re-imported so ``tools.delegate_tool.<name>`` keeps resolving for
# callers and patching tests. Mutable flag globals live only in their owning module.
from tools.delegate_tool_child_run import ( # noqa: F401
_WorktreeReporter, _append_missed_steer, _append_sibling_write_reminder, _attach_child,
_WorktreeReporter, _append_missed_steer, _append_sibling_write_reminder, _attach_child, _finish_failed_entry,
_await_child, _build_result_entry, _cleanup_child_run, _dump_subagent_timeout_diagnostic,
_emit_child_complete, _fabricated_entry, _lease_child_credential, _make_text_relay,
_merge_late_steer, _register_child, _seed_child_workspace, _start_heartbeat,
@@ -46,7 +46,7 @@ from tools.delegate_tool_dispatch import ( # noqa: F401
)
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, _resolve_workspace_hint,
_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
@@ -105,83 +105,11 @@ def _open_child_session_db(parent_agent) -> Any:
parent_session_db = getattr(parent_agent, "_session_db", None)
if parent_session_db is None:
return None
try:
with _quiet("subagent: failed to open dedicated SessionDB; child persistence disabled", exc_info=True):
from hermes_state import get_shared_session_db
_parent_db_path = getattr(parent_session_db, "db_path", None)
return get_shared_session_db(_parent_db_path) if _parent_db_path is not None else get_shared_session_db()
except Exception:
logger.debug("subagent: failed to open dedicated SessionDB; child persistence disabled", exc_info=True)
return None
def _construct_child_agent(
rt: Dict[str, Any], *, task_index: int, max_iterations: int, parent_agent, child_toolsets: List[str],
child_disabled_toolsets: List[str], child_prompt: str, child_progress_cb: Any, child_session_db: Any,
override_provider: Optional[str], override_request_overrides: Optional[Dict[str, Any]],
):
"""Instantiate the child AIAgent; releases the dedicated SessionDB handle on
a construction failure (no child close() will ever run)."""
from run_agent import AIAgent
from agent.delegation_context import delegated_child_context
child_thinking_cb = None
if child_progress_cb:
def _child_thinking(text: str) -> None:
if text:
_safe_progress(child_progress_cb, "_thinking", text)
child_thinking_cb = _child_thinking
with delegated_child_context():
try:
return 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=child_thinking_cb,
session_db=child_session_db,
parent_session_id=getattr(parent_agent, "session_id", None),
request_overrides=(
# honored whenever set, incl. the inherit branch where
# _resolve_delegation_credentials already merged OVER the parent's
dict(override_request_overrides)
if override_request_overrides is not None
else ({} if override_provider else dict(getattr(parent_agent, "request_overrides", {}) or {}))
),
tool_progress_callback=child_progress_cb,
iteration_budget=None, # fresh budget per subagent
)
except BaseException:
if child_session_db is not None:
try:
from hermes_state import release_or_close
release_or_close(child_session_db)
except Exception:
pass
raise
def _announce_child_spawn(child, parent_agent, child_progress_cb, *, goal, subagent_id, parent_subagent_id, role) -> None:
"""spawn_requested event (now — the child may queue for seconds when the
pool is saturated) plus the subagent_start lifecycle hook."""
_safe_progress(child_progress_cb, "subagent.spawn_requested", preview=goal)
try:
from hermes_cli.lifecycle import invoke_hook as _invoke_hook
_invoke_hook(
"subagent_start", parent_session_id=getattr(parent_agent, "session_id", None),
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=role,
child_goal=goal,
)
except Exception:
logger.debug("subagent_start hook invocation failed", exc_info=True)
return None
def _build_child_agent(
task_index: int,
@@ -211,6 +139,8 @@ def _build_child_agent(
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
@@ -232,18 +162,13 @@ def _build_child_agent(
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,
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,
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,
@@ -251,18 +176,48 @@ def _build_child_agent(
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 {})
child_session_db = _open_child_session_db(parent_agent)
child = _construct_child_agent(
rt, task_index=task_index, max_iterations=max_iterations, parent_agent=parent_agent,
child_toolsets=child_toolsets, child_disabled_toolsets=child_disabled_toolsets, child_prompt=child_prompt,
child_progress_cb=child_progress_cb, child_session_db=child_session_db, override_provider=override_provider,
override_request_overrides=override_request_overrides,
)
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=getattr(parent_agent, "session_id", None),
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 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
# Shared ref: session_id now, delegation_id once delegate_task stamps it —
# both ride on every relayed event (first emit is spawn_requested below).
child_session_ref["session_id"] = getattr(child, "session_id", "") or ""
child._progress_identity_ref = child_session_ref
child._delegate_depth = child_depth
@@ -287,10 +242,17 @@ def _build_child_agent(
child._credential_pool = child_pool
_attach_child(parent_agent, child) # interrupt propagation
_announce_child_spawn(
child, parent_agent, child_progress_cb,
goal=goal, subagent_id=subagent_id, parent_subagent_id=parent_subagent_id, role=effective_role,
)
# 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(
@@ -347,10 +309,8 @@ def _run_single_child(
# Flush any remaining batched progress to gateway
if child_progress_cb and hasattr(child_progress_cb, "_flush"):
try:
with _quiet("Progress callback flush failed: %s"):
child_progress_cb._flush()
except Exception as e:
logger.debug("Progress callback flush failed: %s", e)
duration = round(time.monotonic() - child_start, 2)
entry = _build_result_entry(child, result, task_index, duration, schema)
@@ -364,13 +324,10 @@ def _run_single_child(
_late_pending_steer = (_close_subagent_steering(_subagent_id, child) if _subagent_id else None)
duration = round(time.monotonic() - child_start, 2)
logging.exception(f"[subagent-{task_index}] failed")
_safe_progress(
child_progress_cb, "subagent.complete", preview=str(exc), status="failed", duration_seconds=duration, summary=str(exc),
return _finish_failed_entry(
_fabricated_entry(task_index, "error", str(exc), child, duration), _late_pending_steer, child_progress_cb,
worktree, preview=str(exc), summary=str(exc), status="failed",
)
_error_entry = _fabricated_entry(task_index, "error", str(exc), child, duration)
_append_missed_steer(_error_entry, _late_pending_steer)
worktree.attach(_error_entry) # no-op when isolation never engaged
return _error_entry
finally:
_cleanup_child_run(
@@ -388,7 +345,6 @@ def _build_children(
from tools.delegation_live_log import wrap_progress_callback
children = []
for i, t in enumerate(task_list):
effective_role = _normalize_role(t.get("role") or top_role)
_task_schema = task_schemas[i] if i < len(task_schemas) else None
_child_context = t.get("context")
if _task_schema is not None:
@@ -396,38 +352,25 @@ def _build_children(
_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,
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,
override_provider=creds["provider"],
override_base_url=creds["base_url"],
override_api_key=creds["api_key"],
override_api_mode=creds["api_mode"],
model=creds["model"], max_iterations=max_iterations, task_count=len(task_list),
parent_agent=parent_agent, 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"),
role=effective_role,
override_max_tokens=creds.get("max_output_tokens"), override_acp_command=creds.get("command"),
override_acp_args=creds.get("args"), role=_normalize_role(t.get("role") or top_role),
)
except ValueError as exc:
return [], str(exc)
if _task_schema is not None:
try:
with _quiet("Could not attach output schema to child %d", i):
child._delegate_output_schema = _task_schema
except Exception:
logger.debug("Could not attach output schema to child %d", i)
# 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.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)
@@ -713,18 +656,10 @@ def _model_background_value(args: dict, parent_agent=None) -> bool:
_MODEL_HIDDEN_TASK_FIELDS = {"acp_command", "acp_args"}
def _strip_model_hidden_task_fields(tasks: Any) -> Any:
if not isinstance(tasks, list):
"""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
stripped_tasks = []
changed = False
for task in tasks:
if not isinstance(task, dict):
stripped_tasks.append(task)
continue
stripped = {key: value for key, value in task.items() if key not in _MODEL_HIDDEN_TASK_FIELDS}
changed = changed or len(stripped) != len(task)
stripped_tasks.append(stripped)
return stripped_tasks if changed else 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(

7
tools/delegate_tool_child_run.py
View File

@@ -523,13 +523,14 @@ def _merge_late_steer(result: Dict[str, Any], subagent_id: Optional[str], child:
result["pending_steer"] = f"{existing}\n{late}" if isinstance(existing, str) and existing else late
def _finish_failed_entry(
entry: Dict[str, Any], late_steer: Optional[str], child_progress_cb: Any, worktree: _WorktreeReporter, *, preview: str,
entry: Dict[str, Any], late_steer: Optional[str], child_progress_cb: Any, worktree: _WorktreeReporter, *,
preview: str, summary: str = "", status: Optional[str] = None,
) -> Dict[str, Any]:
"""Shared tail of every child failure path: emit ``subagent.complete``, note
the steer text that won the race with the failure, report the worktree."""
_safe_progress(
child_progress_cb, "subagent.complete", preview=preview, status=entry["status"],
duration_seconds=entry["duration_seconds"], summary=entry["summary"] or "",
child_progress_cb, "subagent.complete", preview=preview, status=status or entry["status"],
duration_seconds=entry["duration_seconds"], summary=summary,
)
_append_missed_steer(entry, late_steer)
worktree.attach(entry) # no-op when isolation never engaged