diff --git a/acp_adapter/tools.py b/acp_adapter/tools.py index 8756aa92967e..ca742aa5ef6f 100644 --- a/acp_adapter/tools.py +++ b/acp_adapter/tools.py @@ -43,6 +43,12 @@ "browser_get_images": "read", # Agent internals "delegate_task": "execute", + "spawn_background_subagent": "execute", + "list_background_subagents": "read", + "send_background_subagent": "execute", + "poll_background_subagent": "read", + "get_background_subagent_status": "read", + "stop_background_subagent": "execute", "vision_analyze": "read", "image_generate": "execute", "text_to_speech": "execute", diff --git a/agent/async_delegate_tasks.py b/agent/async_delegate_tasks.py new file mode 100644 index 000000000000..d7d11ab0e272 --- /dev/null +++ b/agent/async_delegate_tasks.py @@ -0,0 +1,461 @@ +"""Background async delegate_task runtime.""" + +from __future__ import annotations + +import json +import logging +import os +import threading +import time +import uuid +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any, Optional + +logger = logging.getLogger(__name__) + +_DEFAULT_ASYNC_DELEGATE_CONFIG: dict[str, Any] = { + "enabled": True, + "max_per_session": 2, + "max_global": 4, + "idle_timeout_seconds": 900, + "max_duration_seconds": 1800, + "output_dir": ".hermes-async-delegates", +} + +_ACTIVE_STATUSES = {"starting", "running", "idle"} +_TERMINAL_STATUSES = {"completed", "timed_out", "error"} + + +def _now() -> float: + return time.time() + + +def _deep_merge(base: dict[str, Any], override: dict[str, Any]) -> dict[str, Any]: + merged = dict(base) + for key, value in (override or {}).items(): + if isinstance(value, dict) and isinstance(merged.get(key), dict): + merged[key] = _deep_merge(merged[key], value) + else: + merged[key] = value + return merged + + +def _load_async_delegate_config() -> dict[str, Any]: + runtime_cfg: dict[str, Any] = {} + try: + from cli import CLI_CONFIG + + runtime_cfg = dict(((CLI_CONFIG or {}).get("delegation", {}) or {}).get("async_subagents", {}) or {}) + except Exception: + runtime_cfg = {} + + persistent_cfg: dict[str, Any] = {} + try: + from hermes_cli.config import load_config + + persistent_cfg = dict(((load_config() or {}).get("delegation", {}) or {}).get("async_subagents", {}) or {}) + except Exception: + persistent_cfg = {} + + return _deep_merge(_deep_merge(_DEFAULT_ASYNC_DELEGATE_CONFIG, persistent_cfg), runtime_cfg) + + +def _truncate(text: str, limit: int = 120) -> str: + clean = " ".join(str(text or "").split()) + if len(clean) <= limit: + return clean + return clean[: limit - 1].rstrip() + "…" + + +def _display_name(delegate_id: str) -> str: + short = delegate_id.replace("async-delegate-", "")[:8] + return f"delegate-{short}" + + +def _resolve_workspace_root(parent_agent) -> Path: + candidates = [ + getattr(parent_agent, "_current_workspace", None), + os.getenv("TERMINAL_CWD"), + os.getcwd(), + ] + for item in candidates: + if not item: + continue + try: + path = Path(str(item)).expanduser() + if path.is_dir(): + return path.resolve() + except Exception: + continue + return Path(os.getcwd()).resolve() + + +@dataclass +class AsyncDelegateRecord: + id: str + name: str + owner_session_id: str + goal: str + context: str + output_file: str + status: str + started_at: float + last_activity_at: float + idle_timeout_at: float + max_duration_at: float + child_session_id: str = "" + profile: str = "" + toolsets: list[str] = field(default_factory=list) + final_result: dict[str, Any] | None = None + last_error: str = "" + nudged_completion: bool = False + nudged_timeout: bool = False + thread: threading.Thread | None = None + child: Any = None + activity_log: list[str] = field(default_factory=list) + lock: threading.RLock = field(default_factory=threading.RLock, repr=False) + + def is_alive(self) -> bool: + return bool(self.thread and self.thread.is_alive()) + + def summary(self, now: float | None = None, idle_timeout_seconds: int = 900) -> dict[str, Any]: + now_ts = now or _now() + effective_status = self.status + if self.status == "running" and now_ts >= self.last_activity_at + idle_timeout_seconds: + effective_status = "idle" + return { + "id": self.id, + "name": self.name, + "status": effective_status, + "goal": self.goal, + "output_file": self.output_file, + "started_at": self.started_at, + "last_activity_at": self.last_activity_at, + "max_duration_at": self.max_duration_at, + "child_session_id": self.child_session_id, + "profile": self.profile, + "toolsets": list(self.toolsets), + } + + +class AsyncDelegateManager: + def __init__(self): + self._lock = threading.RLock() + self._records: dict[str, AsyncDelegateRecord] = {} + self._pending_nudges: dict[str, list[str]] = {} + self._monitor_thread: threading.Thread | None = None + self._monitor_stop = threading.Event() + + def reset_for_tests(self) -> None: + with self._lock: + records = list(self._records.values()) + self._records.clear() + self._pending_nudges.clear() + for record in records: + child = record.child + if child and hasattr(child, "interrupt"): + try: + child.interrupt() + except Exception: + pass + + def spawn( + self, + *, + owner_session_id: str, + parent_agent, + goal: str, + context: str = "", + toolsets: list[str] | None = None, + profile: str | None = None, + max_iterations: int | None = None, + creds: dict[str, Any] | None = None, + ) -> dict[str, Any]: + cfg = _load_async_delegate_config() + if not cfg.get("enabled", True): + return {"success": False, "error": "Async delegate subagents are disabled in config."} + + goal = str(goal or "").strip() + if not goal: + return {"success": False, "error": "goal is required."} + if getattr(parent_agent, "_delegate_depth", 0) >= 1: + return {"success": False, "error": "Async delegate_task is only supported from the top-level agent."} + + with self._lock: + session_active = [ + record for record in self._records.values() + if record.owner_session_id == owner_session_id and record.status in _ACTIVE_STATUSES and record.is_alive() + ] + global_active = [ + record for record in self._records.values() + if record.status in _ACTIVE_STATUSES and record.is_alive() + ] + if len(session_active) >= int(cfg.get("max_per_session", 2)): + return { + "success": False, + "error": "Async delegate limit reached for this session.", + "active_delegates": [record.summary(idle_timeout_seconds=int(cfg.get("idle_timeout_seconds", 900))) for record in session_active], + } + if len(global_active) >= int(cfg.get("max_global", 4)): + return { + "success": False, + "error": "Global async delegate limit reached.", + "active_delegates": [record.summary(idle_timeout_seconds=int(cfg.get("idle_timeout_seconds", 900))) for record in session_active], + } + + workspace_root = _resolve_workspace_root(parent_agent) + output_dir = workspace_root / str(cfg.get("output_dir") or ".hermes-async-delegates") + output_dir.mkdir(parents=True, exist_ok=True) + delegate_id = f"async-delegate-{uuid.uuid4().hex[:10]}" + output_file = output_dir / f"{delegate_id}.md" + + from tools.delegate_tool import ( + _build_child_agent, + _load_config as _load_delegate_cfg, + _resolve_delegation_profile, + _run_single_child, + ) + import model_tools as _model_tools + + delegate_cfg = _load_delegate_cfg() + try: + resolved_profile = _resolve_delegation_profile(delegate_cfg, profile) + except ValueError as exc: + return {"success": False, "error": str(exc)} + parent_tool_names = list(getattr(_model_tools, "_last_resolved_tool_names", [])) + try: + child = _build_child_agent( + task_index=0, + goal=goal, + context=context, + toolsets=toolsets, + profile=resolved_profile, + model=(creds or {}).get("model"), + max_iterations=max_iterations or delegate_cfg.get("max_iterations", 50), + parent_agent=parent_agent, + override_provider=(creds or {}).get("provider"), + override_base_url=(creds or {}).get("base_url"), + override_api_key=(creds or {}).get("api_key"), + override_api_mode=(creds or {}).get("api_mode"), + ) + finally: + _model_tools._last_resolved_tool_names = parent_tool_names + child._delegate_saved_tool_names = parent_tool_names + + # Async delegates should share the current workspace inside docker sandboxes. + terminal_overrides = dict(getattr(child, "_delegate_terminal_overrides", None) or {}) + if terminal_overrides.get("env_type") == "docker": + terminal_overrides.setdefault("docker_mount_cwd_to_workspace", True) + terminal_overrides.setdefault("cwd", "/workspace") + child._delegate_terminal_overrides = terminal_overrides + + record = AsyncDelegateRecord( + id=delegate_id, + name=_display_name(delegate_id), + owner_session_id=owner_session_id, + goal=goal, + context=context, + output_file=str(output_file), + status="running", + started_at=_now(), + last_activity_at=_now(), + idle_timeout_at=_now() + int(cfg.get("idle_timeout_seconds", 900)), + max_duration_at=_now() + int(cfg.get("max_duration_seconds", 1800)), + profile=str(profile or ""), + toolsets=list(toolsets or []), + child=child, + ) + + child.tool_progress_callback = self._build_progress_callback(record) + self._write_record_file(record) + + thread = threading.Thread( + target=self._run_delegate_thread, + args=(record, parent_agent, _run_single_child), + daemon=True, + name=f"async-delegate-{delegate_id}", + ) + record.thread = thread + + with self._lock: + self._records[record.id] = record + self._ensure_monitor_thread() + + thread.start() + return { + "success": True, + "mode": "async", + "id": record.id, + "name": record.name, + "status": "running", + "goal": record.goal, + "output_file": record.output_file, + "workspace_root": str(workspace_root), + "max_duration_at": record.max_duration_at, + } + + def render_turn_context(self, owner_session_id: str) -> str: + cfg = _load_async_delegate_config() + idle_timeout_seconds = int(cfg.get("idle_timeout_seconds", 900)) + with self._lock: + nudges = list(self._pending_nudges.get(owner_session_id, [])) + if nudges: + self._pending_nudges[owner_session_id].clear() + active = [ + record for record in self._records.values() + if record.owner_session_id == owner_session_id and record.status in _ACTIVE_STATUSES and record.is_alive() + ] + + if not nudges and not active: + return "" + + parts: list[str] = [] + if nudges: + parts.append( + "Async delegated subagent updates since your last turn:\n" + + "\n".join(f"- {item}" for item in nudges) + ) + if active: + lines = [] + now_ts = _now() + for record in sorted(active, key=lambda item: item.started_at): + summary = record.summary(now=now_ts, idle_timeout_seconds=idle_timeout_seconds) + lines.append( + f"- {summary['name']} ({summary['id']}) [{summary['status']}] " + f"goal={_truncate(summary['goal'], 120)} output_file={summary['output_file']}" + ) + parts.append( + "Open async delegate_task subagents still running in the background:\n" + + "\n".join(lines) + ) + return "\n\n".join(parts) + + def _build_progress_callback(self, record: AsyncDelegateRecord): + def _callback(tool_name: str, preview: str = None, args: dict | None = None): + text = tool_name + if preview: + text = f"{tool_name}: {_truncate(preview, 100)}" + with record.lock: + record.last_activity_at = _now() + record.idle_timeout_at = record.last_activity_at + int(_load_async_delegate_config().get("idle_timeout_seconds", 900)) + if record.status not in _TERMINAL_STATUSES: + record.status = "running" + record.activity_log.append(text) + self._write_record_file_locked(record) + return _callback + + def _run_delegate_thread(self, record: AsyncDelegateRecord, parent_agent, run_single_child) -> None: + result = run_single_child( + task_index=0, + goal=record.goal, + child=record.child, + parent_agent=parent_agent, + ) + with record.lock: + record.final_result = result + record.child_session_id = getattr(record.child, "session_id", "") or "" + if record.status == "timed_out": + record.activity_log.append("Subagent exceeded maximum runtime and was interrupted.") + elif result.get("status") == "completed": + record.status = "completed" + self._queue_nudge(record.owner_session_id, f"{record.name} ({record.id}) completed. Read {record.output_file} for the summary.") + else: + record.status = "error" + record.last_error = str(result.get("error") or "Subagent failed.") + self._queue_nudge(record.owner_session_id, f"{record.name} ({record.id}) failed: {record.last_error}") + if result.get("summary"): + record.activity_log.append("Final summary written below.") + self._write_record_file_locked(record) + + def _ensure_monitor_thread(self) -> None: + if self._monitor_thread and self._monitor_thread.is_alive(): + return + self._monitor_stop.clear() + self._monitor_thread = threading.Thread( + target=self._monitor_loop, + daemon=True, + name="async-delegate-monitor", + ) + self._monitor_thread.start() + + def _monitor_loop(self) -> None: + while not self._monitor_stop.wait(1.0): + self._sweep_once() + with self._lock: + if not any(record.status in _ACTIVE_STATUSES and record.is_alive() for record in self._records.values()): + self._monitor_stop.set() + + def _sweep_once(self) -> None: + now_ts = _now() + with self._lock: + records = list(self._records.values()) + for record in records: + if record.status in _TERMINAL_STATUSES or not record.is_alive(): + continue + with record.lock: + if now_ts >= record.max_duration_at and not record.nudged_timeout: + record.status = "timed_out" + record.nudged_timeout = True + record.activity_log.append("Maximum runtime reached. Interrupt requested.") + child = record.child + if child and hasattr(child, "interrupt"): + try: + child.interrupt() + except Exception: + logger.debug("Failed to interrupt async delegate child", exc_info=True) + self._queue_nudge(record.owner_session_id, f"{record.name} ({record.id}) hit its maximum runtime.") + self._write_record_file_locked(record) + + def _queue_nudge(self, owner_session_id: str, text: str) -> None: + with self._lock: + self._pending_nudges.setdefault(owner_session_id, []).append(text) + + def _write_record_file(self, record: AsyncDelegateRecord) -> None: + with record.lock: + self._write_record_file_locked(record) + + def _write_record_file_locked(self, record: AsyncDelegateRecord) -> None: + path = Path(record.output_file) + path.parent.mkdir(parents=True, exist_ok=True) + body_lines = [ + f"# Async Delegate {record.name}", + "", + f"- ID: `{record.id}`", + f"- Status: `{record.status}`", + f"- Goal: {record.goal}", + f"- Started At: `{record.started_at}`", + f"- Last Activity At: `{record.last_activity_at}`", + f"- Max Duration At: `{record.max_duration_at}`", + ] + if record.child_session_id: + body_lines.append(f"- Child Session ID: `{record.child_session_id}`") + if record.profile: + body_lines.append(f"- Profile: `{record.profile}`") + if record.toolsets: + body_lines.append(f"- Toolsets: `{', '.join(record.toolsets)}`") + + body_lines.extend(["", "## Progress", ""]) + if record.activity_log: + body_lines.extend(f"- {line}" for line in record.activity_log[-50:]) + else: + body_lines.append("- Subagent started.") + + body_lines.extend(["", "## Result", ""]) + if record.final_result and record.final_result.get("summary"): + body_lines.append(str(record.final_result["summary"])) + elif record.last_error: + body_lines.append(f"Error: {record.last_error}") + else: + body_lines.append("Still running.") + + tmp_path = path.with_suffix(path.suffix + ".tmp") + tmp_path.write_text("\n".join(body_lines) + "\n", encoding="utf-8") + tmp_path.replace(path) + + +_ASYNC_DELEGATE_MANAGER = AsyncDelegateManager() + + +def get_async_delegate_manager() -> AsyncDelegateManager: + return _ASYNC_DELEGATE_MANAGER diff --git a/agent/background_subagents.py b/agent/background_subagents.py new file mode 100644 index 000000000000..75b05b0f9b30 --- /dev/null +++ b/agent/background_subagents.py @@ -0,0 +1,1054 @@ +"""Persistent ACP background subagents for long-lived sandboxed work.""" + +from __future__ import annotations + +import json +import logging +import os +import threading +import time +import uuid +from collections import defaultdict, deque +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any, Callable, Deque, Optional + +logger = logging.getLogger(__name__) + +try: + import acp + + _ACP_PROTOCOL_VERSION = acp.PROTOCOL_VERSION +except Exception: + _ACP_PROTOCOL_VERSION = 1 + + +_DEFAULT_BACKGROUND_CONFIG: dict[str, Any] = { + "enabled": True, + "max_per_session": 3, + "max_global": 8, + "idle_timeout_seconds": 900, + "max_lifetime_seconds": 7200, + "default_agent_kind": "opencode", + "agents": { + "opencode": { + "command": "opencode", + "args": ["acp"], + "cwd_mode": "session", + }, + }, +} + +_ACTIVE_STATUSES = {"starting", "running", "idle"} +_TERMINAL_STATUSES = {"completed", "stopped", "timed_out", "error"} + +_NAME_LEFT = [ + "amber", + "aster", + "cedar", + "cinder", + "ember", + "harbor", + "hollow", + "lumen", + "marble", + "meadow", + "morrow", + "river", + "sable", + "sierra", + "silver", + "solar", +] +_NAME_RIGHT = [ + "badger", + "falcon", + "finch", + "fox", + "heron", + "ibis", + "lynx", + "otter", + "owl", + "panda", + "raven", + "robin", + "starling", + "stoat", + "swift", + "wolf", +] + + +def _now() -> float: + return time.time() + + +def _deep_merge(base: dict[str, Any], override: dict[str, Any]) -> dict[str, Any]: + merged = dict(base) + for key, value in (override or {}).items(): + if isinstance(value, dict) and isinstance(merged.get(key), dict): + merged[key] = _deep_merge(merged[key], value) + else: + merged[key] = value + return merged + + +def _load_background_subagent_config() -> dict[str, Any]: + runtime_cfg: dict[str, Any] = {} + try: + from cli import CLI_CONFIG + + runtime_cfg = dict(((CLI_CONFIG or {}).get("delegation", {}) or {}).get("background_subagents", {}) or {}) + except Exception: + runtime_cfg = {} + + persistent_cfg: dict[str, Any] = {} + try: + from hermes_cli.config import load_config + + persistent_cfg = dict(((load_config() or {}).get("delegation", {}) or {}).get("background_subagents", {}) or {}) + except Exception: + persistent_cfg = {} + + return _deep_merge(_deep_merge(_DEFAULT_BACKGROUND_CONFIG, persistent_cfg), runtime_cfg) + + +def _generate_display_name(subagent_id: str) -> str: + sanitized = "".join(ch for ch in subagent_id if ch in "0123456789abcdefABCDEF") + seed = int((sanitized or "0")[:8], 16) + left = _NAME_LEFT[seed % len(_NAME_LEFT)] + right = _NAME_RIGHT[(seed // len(_NAME_LEFT)) % len(_NAME_RIGHT)] + return f"{left}-{right}" + + +def _truncate(text: str, limit: int = 120) -> str: + clean = " ".join(str(text or "").split()) + if len(clean) <= limit: + return clean + return clean[: limit - 1].rstrip() + "…" + + +def _jsonrpc_error(message_id: Any, code: int, message: str) -> dict[str, Any]: + return { + "jsonrpc": "2.0", + "id": message_id, + "error": { + "code": code, + "message": message, + }, + } + + +@dataclass +class BufferedEvent: + seq: int + kind: str + text: str + timestamp: float + data: dict[str, Any] = field(default_factory=dict) + + def to_dict(self) -> dict[str, Any]: + payload = { + "seq": self.seq, + "kind": self.kind, + "text": self.text, + "timestamp": self.timestamp, + } + if self.data: + payload["data"] = self.data + return payload + + +@dataclass +class PendingPrompt: + prompt_text: str + queued_at: float + meta: dict[str, Any] = field(default_factory=dict) + + +@dataclass +class PendingRequest: + request_id: int + method: str + event: threading.Event = field(default_factory=threading.Event) + result: Any = None + error: dict[str, Any] | None = None + callback: Optional[Callable[[Any, dict[str, Any] | None], None]] = None + + +@dataclass +class BackgroundSubagentRecord: + id: str + name: str + owner_session_id: str + purpose: str + cwd: str + agent_kind: str + status: str + started_at: float + last_activity_at: float + idle_timeout_at: float + max_lifetime_at: float + acp_session_id: str = "" + container_id: str = "" + acp_process_pid: int | None = None + current_task: str = "" + active_prompt_request_id: int | None = None + pending_prompts: Deque[PendingPrompt] = field(default_factory=deque) + events: list[BufferedEvent] = field(default_factory=list) + last_polled_seq: int = 0 + last_seq: int = 0 + last_error: str = "" + terminal_reason: str = "" + environment: Any = None + connection: Any = None + + @property + def unread_count(self) -> int: + return max(self.last_seq - self.last_polled_seq, 0) + + def summary(self) -> dict[str, Any]: + return { + "id": self.id, + "name": self.name, + "status": self.status, + "purpose": self.purpose, + "cwd": self.cwd, + "agent_kind": self.agent_kind, + "unread_count": self.unread_count, + "current_task": self.current_task, + "started_at": self.started_at, + "last_activity_at": self.last_activity_at, + "idle_timeout_at": self.idle_timeout_at, + "max_lifetime_at": self.max_lifetime_at, + "container_id": self.container_id, + "acp_process_pid": self.acp_process_pid, + "acp_session_id": self.acp_session_id, + "terminal_reason": self.terminal_reason, + } + + +class ACPPeerConnection: + """Persistent ACP stdio client over a long-lived exec session.""" + + def __init__( + self, + exec_session: Any, + *, + cwd: str, + on_update: Callable[[dict[str, Any]], None], + on_stderr: Callable[[str], None], + on_exit: Callable[[int | None], None], + on_protocol_error: Callable[[str], None], + on_activity: Callable[[], None], + inbound_handler: Optional[Callable[[str, dict[str, Any]], tuple[Any, dict[str, Any] | None]]] = None, + ): + self._exec_session = exec_session + self._cwd = cwd + self._on_update = on_update + self._on_stderr = on_stderr + self._on_exit = on_exit + self._on_protocol_error = on_protocol_error + self._on_activity = on_activity + self._inbound_handler = inbound_handler or self._default_inbound_handler + self._lock = threading.Lock() + self._next_id = 0 + self._pending: dict[int, PendingRequest] = {} + self._closed = False + self._exec_session.read_loop( + stdout_handler=self._handle_stdout_line, + stderr_handler=self._handle_stderr_line, + exit_handler=self._handle_exit, + ) + + def initialize(self) -> None: + self.request( + "initialize", + { + "protocolVersion": _ACP_PROTOCOL_VERSION, + "clientCapabilities": { + "fs": { + "readTextFile": False, + "writeTextFile": False, + } + }, + "clientInfo": { + "name": "hermes-agent", + "title": "Hermes Agent", + "version": "0.0.0", + }, + }, + timeout=20.0, + ) + + def open_session(self) -> str: + result = self.request( + "session/new", + { + "cwd": self._cwd, + "mcpServers": [], + }, + timeout=20.0, + ) or {} + session_id = str(result.get("sessionId") or "").strip() + if not session_id: + raise RuntimeError("ACP peer did not return a sessionId.") + return session_id + + def send_prompt_async( + self, + *, + session_id: str, + prompt_text: str, + meta: dict[str, Any] | None = None, + on_response: Optional[Callable[[Any, dict[str, Any] | None], None]] = None, + ) -> int: + return self.request( + "session/prompt", + { + "sessionId": session_id, + "prompt": [{"type": "text", "text": prompt_text}], + "_meta": dict(meta or {}), + }, + wait=False, + callback=on_response, + ) + + def cancel(self, session_id: str) -> None: + try: + self.request( + "session/cancel", + {"sessionId": session_id}, + wait=False, + ) + except Exception: + logger.debug("ACP cancel failed", exc_info=True) + + def close(self) -> None: + self._closed = True + try: + self._exec_session.terminate_session() + except Exception: + pass + + def is_alive(self) -> bool: + return not self._closed and bool(self._exec_session.is_session_alive()) + + @property + def pid(self) -> int | None: + return getattr(self._exec_session, "pid", None) + + def request( + self, + method: str, + params: dict[str, Any], + *, + timeout: float = 60.0, + wait: bool = True, + callback: Optional[Callable[[Any, dict[str, Any] | None], None]] = None, + ) -> Any: + with self._lock: + self._next_id += 1 + request_id = self._next_id + pending = PendingRequest( + request_id=request_id, + method=method, + callback=callback, + ) + self._pending[request_id] = pending + payload = { + "jsonrpc": "2.0", + "id": request_id, + "method": method, + "params": params, + } + try: + self._exec_session.write_line(json.dumps(payload, ensure_ascii=True)) + except Exception as exc: + with self._lock: + self._pending.pop(request_id, None) + raise RuntimeError(f"Failed to write ACP request {method}: {exc}") from exc + + self._on_activity() + if not wait: + return request_id + + if not pending.event.wait(timeout): + with self._lock: + self._pending.pop(request_id, None) + raise TimeoutError(f"Timed out waiting for ACP response to {method}.") + + if pending.error: + raise RuntimeError( + f"ACP {method} failed: {pending.error.get('message') or pending.error}" + ) + return pending.result + + def _handle_stdout_line(self, line: str) -> None: + if self._closed: + return + try: + msg = json.loads(line) + except Exception: + self._on_protocol_error(f"Malformed ACP stdout line: {line!r}") + return + + self._on_activity() + method = msg.get("method") + if isinstance(method, str): + self._handle_inbound_message(msg) + return + + message_id = msg.get("id") + if not isinstance(message_id, int): + self._on_protocol_error(f"ACP message missing integer id: {msg!r}") + return + + with self._lock: + pending = self._pending.pop(message_id, None) + if pending is None: + logger.debug("Ignoring ACP response for unknown request id %s", message_id) + return + + pending.result = msg.get("result") + pending.error = msg.get("error") + pending.event.set() + if pending.callback: + try: + pending.callback(pending.result, pending.error) + except Exception: + logger.warning("ACP pending callback failed", exc_info=True) + + def _handle_stderr_line(self, line: str) -> None: + if line.strip(): + self._on_stderr(line.rstrip("\n")) + + def _handle_exit(self, returncode: int | None) -> None: + self._closed = True + with self._lock: + pending = list(self._pending.values()) + self._pending.clear() + for item in pending: + item.error = {"message": f"ACP process exited before {item.method} completed."} + item.event.set() + if item.callback: + try: + item.callback(None, item.error) + except Exception: + logger.warning("ACP exit callback failed", exc_info=True) + self._on_exit(returncode) + + def _handle_inbound_message(self, msg: dict[str, Any]) -> None: + method = str(msg.get("method") or "") + if method == "session/update": + self._on_update(msg) + return + + message_id = msg.get("id") + params = msg.get("params") or {} + result, error = self._inbound_handler(method, params) + if message_id is None: + return + if error: + response = _jsonrpc_error(message_id, error.get("code", -32601), error.get("message", "Unsupported ACP method")) + else: + response = { + "jsonrpc": "2.0", + "id": message_id, + "result": result, + } + try: + self._exec_session.write_line(json.dumps(response, ensure_ascii=True)) + except Exception as exc: + self._on_protocol_error(f"Failed to write ACP inbound response: {exc}") + + @staticmethod + def _default_inbound_handler(method: str, params: dict[str, Any]) -> tuple[Any, dict[str, Any] | None]: + if method == "session/request_permission": + return { + "outcome": { + "outcome": "allow_once", + } + }, None + return None, {"code": -32601, "message": f"ACP client method '{method}' is not supported by Hermes yet."} + + +class BackgroundSubagentManager: + """Tracks persistent ACP background subagents across Hermes turns.""" + + def __init__(self): + self._lock = threading.RLock() + self._records: dict[str, BackgroundSubagentRecord] = {} + self._pending_nudges: dict[str, list[str]] = defaultdict(list) + self._monitor_thread: threading.Thread | None = None + self._monitor_stop = threading.Event() + + def check_requirements(self) -> bool: + cfg = _load_background_subagent_config() + return bool(cfg.get("enabled", True)) + + def reset_for_tests(self) -> None: + with self._lock: + records = list(self._records.values()) + self._records.clear() + self._pending_nudges.clear() + for record in records: + self._shutdown_record(record, "stopped", "reset") + + def spawn_subagent( + self, + *, + owner_session_id: str, + purpose: str, + initial_task: str, + cwd: str, + agent_kind: str | None = None, + ) -> dict[str, Any]: + cfg = _load_background_subagent_config() + if not cfg.get("enabled", True): + return {"success": False, "error": "Background ACP subagents are disabled in config."} + + purpose = str(purpose or "").strip() + initial_task = str(initial_task or "").strip() + cwd = str(cwd or "").strip() + if not purpose: + return {"success": False, "error": "purpose is required."} + if not initial_task: + return {"success": False, "error": "initial_task is required."} + if not cwd: + return {"success": False, "error": "cwd is required."} + + resolved_agent_kind = str(agent_kind or cfg.get("default_agent_kind") or "opencode").strip() or "opencode" + agent_cfg = self._resolve_agent_config(cfg, resolved_agent_kind) + if "error" in agent_cfg: + return {"success": False, "error": agent_cfg["error"]} + + with self._lock: + session_active = [ + record for record in self._records.values() + if record.owner_session_id == owner_session_id and record.status in _ACTIVE_STATUSES + ] + global_active = [record for record in self._records.values() if record.status in _ACTIVE_STATUSES] + if len(session_active) >= int(cfg.get("max_per_session", 3)): + return { + "success": False, + "error": "Background subagent limit reached for this session.", + "stoppable_subagents": [record.summary() for record in session_active], + } + if len(global_active) >= int(cfg.get("max_global", 8)): + return { + "success": False, + "error": "Global background subagent limit reached.", + "stoppable_subagents": [record.summary() for record in session_active], + } + + subagent_id = f"bg-{uuid.uuid4().hex[:10]}" + record = BackgroundSubagentRecord( + id=subagent_id, + name=_generate_display_name(subagent_id), + owner_session_id=owner_session_id, + purpose=purpose, + cwd=cwd, + agent_kind=resolved_agent_kind, + status="starting", + started_at=_now(), + last_activity_at=_now(), + idle_timeout_at=_now() + int(cfg.get("idle_timeout_seconds", 900)), + max_lifetime_at=_now() + int(cfg.get("max_lifetime_seconds", 7200)), + ) + + try: + environment = self._create_environment(task_id=subagent_id, cwd=cwd) + command = [agent_cfg["command"], *list(agent_cfg.get("args") or [])] + exec_session = environment.start_persistent_exec( + cwd=cwd, + command=command, + env=dict(agent_cfg.get("env") or {}), + ) + connection = ACPPeerConnection( + exec_session, + cwd=cwd, + on_update=lambda msg, record_id=subagent_id: self._handle_update(record_id, msg), + on_stderr=lambda line, record_id=subagent_id: self._handle_stderr(record_id, line), + on_exit=lambda returncode, record_id=subagent_id: self._handle_exit(record_id, returncode), + on_protocol_error=lambda message, record_id=subagent_id: self._handle_protocol_error(record_id, message), + on_activity=lambda record_id=subagent_id: self._touch(record_id), + ) + connection.initialize() + session_id = connection.open_session() + except Exception as exc: + try: + environment.cleanup() # type: ignore[name-defined] + except Exception: + pass + return { + "success": False, + "error": f"Failed to start background ACP subagent: {exc}", + } + + record.environment = environment + record.connection = connection + record.acp_session_id = session_id + record.container_id = str(getattr(environment, "_container_id", "") or "") + record.acp_process_pid = connection.pid + record.status = "idle" + + with self._lock: + self._records[record.id] = record + self._ensure_monitor_thread() + + send_result = self.send_message( + owner_session_id=owner_session_id, + subagent_id=record.id, + message=initial_task, + source="spawn", + ) + if not send_result.get("success"): + return send_result + + return { + "success": True, + "id": record.id, + "name": record.name, + "status": record.status, + "purpose": record.purpose, + "cwd": record.cwd, + "agent_kind": record.agent_kind, + "container_id": record.container_id, + "acp_session_id": record.acp_session_id, + "initial_dispatch": send_result, + } + + def list_subagents(self, *, owner_session_id: str) -> dict[str, Any]: + with self._lock: + subagents = [ + record.summary() + for record in self._records.values() + if record.owner_session_id == owner_session_id and record.status in _ACTIVE_STATUSES + ] + subagents.sort(key=lambda item: item["started_at"]) + return {"success": True, "subagents": subagents} + + def get_status(self, *, owner_session_id: str, subagent_id: str) -> dict[str, Any]: + record = self._get_owned_record(owner_session_id, subagent_id) + if record is None: + return {"success": False, "error": f"Unknown background subagent: {subagent_id}"} + alive = bool(record.connection and record.connection.is_alive()) + payload = record.summary() + payload.update( + { + "success": True, + "transport_alive": alive, + "queued_messages": len(record.pending_prompts), + } + ) + if record.last_error: + payload["last_error"] = record.last_error + return payload + + def send_message( + self, + *, + owner_session_id: str, + subagent_id: str, + message: str, + source: str = "tool", + ) -> dict[str, Any]: + record = self._get_owned_record(owner_session_id, subagent_id) + if record is None: + return {"success": False, "error": f"Unknown background subagent: {subagent_id}"} + text = str(message or "").strip() + if not text: + return {"success": False, "error": "message is required."} + if record.status in _TERMINAL_STATUSES: + return {"success": False, "error": f"Background subagent {record.name} is no longer running."} + + prompt = PendingPrompt( + prompt_text=text, + queued_at=_now(), + meta={"source": source, "subagentId": record.id}, + ) + shutdown_record: BackgroundSubagentRecord | None = None + with self._lock: + if record.active_prompt_request_id is not None: + record.pending_prompts.append(prompt) + record.status = "running" + queued = True + else: + queued = False + try: + self._dispatch_prompt_locked(record, prompt) + except Exception as exc: + record.last_error = str(exc) + self._append_event_locked(record, "error", record.last_error, data={"error": str(exc)}) + shutdown_record = record + else: + shutdown_record = None + if shutdown_record is not None: + self._shutdown_record(shutdown_record, "error", "dispatch_failed") + return { + "success": False, + "error": f"Failed to dispatch prompt to {record.name}: {record.last_error}", + "id": record.id, + "name": record.name, + } + return { + "success": True, + "id": record.id, + "name": record.name, + "status": record.status, + "queued": queued, + "queued_messages": len(record.pending_prompts), + } + + def poll_subagent( + self, + *, + owner_session_id: str, + subagent_id: str, + since_seq: int | None = None, + ) -> dict[str, Any]: + record = self._get_owned_record(owner_session_id, subagent_id) + if record is None: + return {"success": False, "error": f"Unknown background subagent: {subagent_id}"} + + with self._lock: + start_seq = int(since_seq or record.last_polled_seq) + events = [event.to_dict() for event in record.events if event.seq > start_seq] + if since_seq is None: + record.last_polled_seq = record.last_seq + + return { + "success": True, + "id": record.id, + "name": record.name, + "status": record.status, + "events": events, + "unread_count": record.unread_count, + "transport_alive": bool(record.connection and record.connection.is_alive()), + } + + def stop_subagent( + self, + *, + owner_session_id: str, + subagent_id: str, + reason: str = "", + ) -> dict[str, Any]: + record = self._get_owned_record(owner_session_id, subagent_id) + if record is None: + return {"success": False, "error": f"Unknown background subagent: {subagent_id}"} + self._shutdown_record(record, "stopped", reason or "stopped_by_parent") + return { + "success": True, + "id": record.id, + "name": record.name, + "status": record.status, + "reason": record.terminal_reason, + } + + def render_turn_context(self, owner_session_id: str) -> str: + with self._lock: + nudges = list(self._pending_nudges.get(owner_session_id, [])) + if nudges: + self._pending_nudges[owner_session_id].clear() + active = [ + record for record in self._records.values() + if record.owner_session_id == owner_session_id and record.status in _ACTIVE_STATUSES + ] + + if not nudges and not active: + return "" + + parts: list[str] = [] + if nudges: + parts.append( + "Background subagent updates since your last turn:\n" + + "\n".join(f"- {item}" for item in nudges) + ) + if active: + roster_lines = [] + for record in sorted(active, key=lambda item: item.started_at): + detail = ( + f"- {record.name} ({record.id}) [{record.status}] " + f"agent={record.agent_kind} cwd={record.cwd} unread={record.unread_count} " + f"purpose={_truncate(record.purpose, 140)}" + ) + if record.current_task: + detail += f" current_task={_truncate(record.current_task, 100)}" + roster_lines.append(detail) + parts.append( + "Open background ACP subagents you can manage with " + "spawn_background_subagent/list_background_subagents/" + "send_background_subagent/poll_background_subagent/" + "get_background_subagent_status/stop_background_subagent:\n" + + "\n".join(roster_lines) + ) + return "\n\n".join(parts) + + def _resolve_agent_config(self, cfg: dict[str, Any], agent_kind: str) -> dict[str, Any]: + agents = dict(cfg.get("agents") or {}) + agent_cfg = dict(agents.get(agent_kind) or {}) + if not agent_cfg: + return {"error": f"Unknown background_subagents agent kind: {agent_kind}"} + command = str(agent_cfg.get("command") or "").strip() + if not command: + return {"error": f"background_subagents agent '{agent_kind}' is missing a command."} + args = agent_cfg.get("args") or [] + if not isinstance(args, list) or not all(isinstance(item, str) for item in args): + return {"error": f"background_subagents agent '{agent_kind}' args must be a list of strings."} + return { + "command": command, + "args": list(args), + "cwd_mode": str(agent_cfg.get("cwd_mode") or "session"), + "env": dict(agent_cfg.get("env") or {}), + } + + def _create_environment(self, *, task_id: str, cwd: str) -> Any: + from tools.terminal_tool import _create_environment, _get_env_config, _resolve_task_environment_settings + + settings = _resolve_task_environment_settings(task_id, _get_env_config()) + env_type = str(settings.get("env_type") or "") + if env_type != "docker": + raise RuntimeError( + f"Background ACP subagents currently require terminal backend 'docker' (got {env_type!r})." + ) + return _create_environment( + env_type=env_type, + image=settings.get("image", ""), + cwd=cwd or settings.get("cwd", "/root"), + timeout=max(int(settings.get("timeout", 180)), 180), + ssh_config=settings.get("ssh_config"), + container_config=settings.get("container_config"), + local_config=settings.get("local_config"), + task_id=task_id, + host_cwd=settings.get("host_cwd"), + ) + + def _dispatch_prompt_locked(self, record: BackgroundSubagentRecord, prompt: PendingPrompt) -> None: + if not record.connection or not record.acp_session_id: + raise RuntimeError("Background subagent connection is not ready.") + request_id = record.connection.send_prompt_async( + session_id=record.acp_session_id, + prompt_text=prompt.prompt_text, + meta={ + "purpose": record.purpose, + "backgroundSubagentId": record.id, + **dict(prompt.meta or {}), + }, + on_response=lambda result, error, record_id=record.id: self._handle_prompt_response(record_id, result, error), + ) + record.active_prompt_request_id = request_id + record.current_task = prompt.prompt_text + record.status = "running" + self._touch_locked(record) + self._append_event_locked( + record, + kind="task_dispatched", + text=_truncate(prompt.prompt_text, 300), + data={"source": prompt.meta.get("source", "tool")}, + ) + + def _handle_prompt_response(self, record_id: str, result: Any, error: dict[str, Any] | None) -> None: + shutdown_record: BackgroundSubagentRecord | None = None + with self._lock: + record = self._records.get(record_id) + if not record: + return + record.active_prompt_request_id = None + record.current_task = "" + if error: + record.last_error = str(error.get("message") or error) + self._append_event_locked(record, "error", record.last_error, data={"error": error}) + shutdown_record = record + else: + stop_reason = "" + if isinstance(result, dict): + stop_reason = str(result.get("stopReason") or result.get("stop_reason") or "").strip() + if stop_reason: + text = f"Latest task finished ({stop_reason})." + else: + text = "Latest task finished." + record.status = "idle" + self._append_event_locked(record, "task_complete", text, data={"result": result or {}}) + self._queue_hidden_nudge_locked( + record.owner_session_id, + f"{record.name} ({record.id}) finished its latest task.", + ) + if record.pending_prompts: + prompt = record.pending_prompts.popleft() + self._dispatch_prompt_locked(record, prompt) + else: + self._touch_locked(record) + + if shutdown_record is not None: + self._shutdown_record(shutdown_record, "error", "prompt_error") + + def _handle_update(self, record_id: str, msg: dict[str, Any]) -> None: + with self._lock: + record = self._records.get(record_id) + if not record: + return + params = msg.get("params") or {} + update = params.get("update") or {} + kind = str(update.get("sessionUpdate") or update.get("type") or "update").strip() or "update" + text = self._extract_update_text(update) + self._append_event_locked(record, kind, text, data={"raw": update}) + if record.status not in _TERMINAL_STATUSES: + record.status = "running" + self._touch_locked(record) + + def _handle_stderr(self, record_id: str, line: str) -> None: + with self._lock: + record = self._records.get(record_id) + if not record: + return + self._append_event_locked(record, "stderr", line, data={}) + self._touch_locked(record) + + def _handle_protocol_error(self, record_id: str, message: str) -> None: + with self._lock: + record = self._records.get(record_id) + if not record: + return + record.last_error = message + self._append_event_locked(record, "protocol_error", message, data={}) + self._shutdown_record(record, "error", "protocol_error") + + def _handle_exit(self, record_id: str, returncode: int | None) -> None: + with self._lock: + record = self._records.get(record_id) + if not record: + return + if record.status in _TERMINAL_STATUSES: + return + status = "completed" if (returncode or 0) == 0 else "error" + reason = f"channel_closed:{returncode if returncode is not None else 'unknown'}" + self._shutdown_record(record, status, reason) + + def _append_event_locked( + self, + record: BackgroundSubagentRecord, + kind: str, + text: str, + *, + data: dict[str, Any] | None = None, + ) -> None: + record.last_seq += 1 + record.events.append( + BufferedEvent( + seq=record.last_seq, + kind=kind, + text=str(text or ""), + timestamp=_now(), + data=dict(data or {}), + ) + ) + + def _touch(self, record_id: str) -> None: + with self._lock: + record = self._records.get(record_id) + if record is None: + return + self._touch_locked(record) + + def _touch_locked(self, record: BackgroundSubagentRecord) -> None: + cfg = _load_background_subagent_config() + record.last_activity_at = _now() + record.idle_timeout_at = record.last_activity_at + int(cfg.get("idle_timeout_seconds", 900)) + + def _queue_hidden_nudge_locked(self, owner_session_id: str, text: str) -> None: + self._pending_nudges[owner_session_id].append(text) + + def _get_owned_record(self, owner_session_id: str, subagent_id: str) -> BackgroundSubagentRecord | None: + with self._lock: + record = self._records.get(subagent_id) + if record is None or record.owner_session_id != owner_session_id: + return None + return record + + def _ensure_monitor_thread(self) -> None: + if self._monitor_thread and self._monitor_thread.is_alive(): + return + self._monitor_stop.clear() + self._monitor_thread = threading.Thread( + target=self._monitor_loop, + daemon=True, + name="background-subagent-monitor", + ) + self._monitor_thread.start() + + def _monitor_loop(self) -> None: + while not self._monitor_stop.wait(1.0): + self._sweep_once() + with self._lock: + if not any(record.status in _ACTIVE_STATUSES for record in self._records.values()): + self._monitor_stop.set() + + def _sweep_once(self) -> None: + now = _now() + with self._lock: + records = [record for record in self._records.values() if record.status in _ACTIVE_STATUSES] + for record in records: + if now >= record.max_lifetime_at: + self._shutdown_record(record, "timed_out", "max_lifetime_exceeded") + continue + if now >= record.idle_timeout_at: + self._shutdown_record(record, "timed_out", "idle_timeout") + continue + if record.connection and not record.connection.is_alive(): + self._shutdown_record(record, "error", "transport_not_alive") + + def _shutdown_record(self, record: BackgroundSubagentRecord, final_status: str, reason: str) -> None: + with self._lock: + if record.status in _TERMINAL_STATUSES: + return + record.status = final_status + record.terminal_reason = reason + if reason: + self._append_event_locked(record, final_status, f"Session ended: {reason}", data={"reason": reason}) + self._queue_hidden_nudge_locked( + record.owner_session_id, + f"{record.name} ({record.id}) is now {final_status} ({reason}).", + ) + if record.connection and record.acp_session_id: + try: + record.connection.cancel(record.acp_session_id) + except Exception: + pass + if record.connection: + try: + record.connection.close() + except Exception: + pass + if record.environment: + try: + record.environment.cleanup() + except Exception: + logger.debug("Background subagent cleanup failed", exc_info=True) + + @staticmethod + def _extract_update_text(update: dict[str, Any]) -> str: + content = update.get("content") + if isinstance(content, dict): + if isinstance(content.get("text"), str): + return content["text"] + if isinstance(content, list): + parts: list[str] = [] + for item in content: + if isinstance(item, dict) and isinstance(item.get("text"), str): + parts.append(item["text"]) + if parts: + return "\n".join(parts) + message = update.get("message") + if isinstance(message, str): + return message + return "" + + +_BACKGROUND_SUBAGENT_MANAGER = BackgroundSubagentManager() + + +def get_background_subagent_manager() -> BackgroundSubagentManager: + return _BACKGROUND_SUBAGENT_MANAGER diff --git a/cli-config.yaml.example b/cli-config.yaml.example index 922807f17a2e..1ae1928dacef 100644 --- a/cli-config.yaml.example +++ b/cli-config.yaml.example @@ -681,7 +681,16 @@ code_execution: # Supports single tasks and batch mode (up to 3 parallel). delegation: max_iterations: 50 # Max tool-calling turns per child (default: 50) - default_toolsets: ["terminal", "file", "web"] # Default toolsets for subagents + default_toolsets: ["terminal", "file", "web"] # Legacy fallback when no delegation profile is selected + # default_profile: "friendly" # Built-in: restricted, friendly, privileged + # profiles: + # docs: + # toolsets: ["file", "web"] + # memory: "read" # none | read | write + # provider_tools: false + # terminal: + # backend: "docker" # docker | modal | local | ssh | singularity | daytona + # docker_image: "nikolaik/python-nodejs:python3.11-nodejs20" # model: "google/gemini-3-flash-preview" # Override model for subagents (empty = inherit parent) # provider: "openrouter" # Override provider for subagents (empty = inherit parent) # # Resolves full credentials (base_url, api_key) automatically. diff --git a/cli.py b/cli.py index 8e1054e8cba6..7dc7e33cfbb6 100644 --- a/cli.py +++ b/cli.py @@ -199,6 +199,39 @@ def load_cli_config() -> Dict[str, Any]: "hype": "YOOO LET'S GOOOO!!! I am SO PUMPED to help you today! Every question is AMAZING and we're gonna CRUSH IT together! This is gonna be LEGENDARY! ARE YOU READY?! LET'S DO THIS!", }, }, + "delegation": { + "max_iterations": 45, # Max tool-calling turns per child agent + "default_toolsets": ["terminal", "file", "web"], # Default toolsets for subagents + "model": "", # Subagent model override (empty = inherit parent model) + "provider": "", # Subagent provider override (empty = inherit parent provider) + "base_url": "", # Direct OpenAI-compatible endpoint for subagents + "api_key": "", # API key for delegation.base_url (falls back to OPENAI_API_KEY) + "default_profile": "", # Named child capability profile (empty = legacy/default toolsets) + "profiles": {}, # Optional custom delegation profiles + "async_subagents": { + "enabled": True, + "max_per_session": 2, + "max_global": 4, + "idle_timeout_seconds": 900, + "max_duration_seconds": 1800, + "output_dir": ".hermes-async-delegates", + }, + "background_subagents": { + "enabled": True, + "max_per_session": 3, + "max_global": 8, + "idle_timeout_seconds": 900, + "max_lifetime_seconds": 7200, + "default_agent_kind": "opencode", + "agents": { + "opencode": { + "command": "opencode", + "args": ["acp"], + "cwd_mode": "session", + }, + }, + }, + }, "display": { "compact": False, @@ -230,14 +263,6 @@ def load_cli_config() -> Dict[str, Any]: "api_key": "", }, }, - "delegation": { - "max_iterations": 45, # Max tool-calling turns per child agent - "default_toolsets": ["terminal", "file", "web"], # Default toolsets for subagents - "model": "", # Subagent model override (empty = inherit parent model) - "provider": "", # Subagent provider override (empty = inherit parent provider) - "base_url": "", # Direct OpenAI-compatible endpoint for subagents - "api_key": "", # API key for delegation.base_url (falls back to OPENAI_API_KEY) - }, } # Track whether the config file explicitly set terminal config. diff --git a/hermes_cli/config.py b/hermes_cli/config.py index da266eedac28..20cd8fef8e88 100644 --- a/hermes_cli/config.py +++ b/hermes_cli/config.py @@ -444,8 +444,33 @@ def ensure_hermes_home(): "provider": "", # e.g. "openrouter" (empty = inherit parent provider + credentials) "base_url": "", # direct OpenAI-compatible endpoint for subagents "api_key": "", # API key for delegation.base_url (falls back to OPENAI_API_KEY) + "default_profile": "", # e.g. "friendly" for built-in or custom delegation profiles + "profiles": {}, # optional named child capability profiles (toolsets, memory, terminal) "max_iterations": 50, # per-subagent iteration cap (each subagent gets its own budget, # independent of the parent's max_iterations) + "async_subagents": { + "enabled": True, + "max_per_session": 2, + "max_global": 4, + "idle_timeout_seconds": 900, + "max_duration_seconds": 1800, + "output_dir": ".hermes-async-delegates", + }, + "background_subagents": { + "enabled": True, + "max_per_session": 3, + "max_global": 8, + "idle_timeout_seconds": 900, + "max_lifetime_seconds": 7200, + "default_agent_kind": "opencode", + "agents": { + "opencode": { + "command": "opencode", + "args": ["acp"], + "cwd_mode": "session", + }, + }, + }, }, # Ephemeral prefill messages file — JSON list of {role, content} dicts diff --git a/model_tools.py b/model_tools.py index ec472ff99ea9..11c4e9f79b2e 100644 --- a/model_tools.py +++ b/model_tools.py @@ -154,6 +154,7 @@ def _discover_tools(): "tools.clarify_tool", "tools.code_execution_tool", "tools.delegate_tool", + "tools.background_subagent_tool", "tools.process_registry", "tools.send_message_tool", # "tools.honcho_tools", # Removed — Honcho is now a memory provider plugin @@ -361,7 +362,18 @@ def get_tool_definitions( # because they need agent-level state (TodoStore, MemoryStore, etc.). # The registry still holds their schemas; dispatch just returns a stub error # so if something slips through, the LLM sees a sensible message. -_AGENT_LOOP_TOOLS = {"todo", "memory", "session_search", "delegate_task"} +_AGENT_LOOP_TOOLS = { + "todo", + "memory", + "session_search", + "delegate_task", + "spawn_background_subagent", + "list_background_subagents", + "send_background_subagent", + "poll_background_subagent", + "get_background_subagent_status", + "stop_background_subagent", +} _READ_SEARCH_TOOLS = {"read_file", "search_files"} diff --git a/run_agent.py b/run_agent.py index ab1023233224..9763fd7b7804 100644 --- a/run_agent.py +++ b/run_agent.py @@ -466,6 +466,9 @@ def __init__( platform: str = None, skip_context_files: bool = False, skip_memory: bool = False, + memory_write_enabled: bool = True, + provider_tool_access: bool = True, + agent_context: str = "primary", session_db=None, iteration_budget: "IterationBudget" = None, fallback_model: Dict[str, Any] = None, @@ -534,6 +537,9 @@ def __init__( self._print_fn = None self.background_review_callback = None # Optional sync callback for gateway delivery self.skip_context_files = skip_context_files + self._memory_write_enabled = bool(memory_write_enabled) + self._provider_tool_access = bool(provider_tool_access) + self._agent_context = str(agent_context or "primary") self.pass_session_id = pass_session_id self.persist_session = persist_session self._credential_pool = credential_pool @@ -1070,7 +1076,7 @@ def __init__( "session_id": self.session_id, "platform": platform or "cli", "hermes_home": str(_ghh()), - "agent_context": "primary", + "agent_context": self._agent_context, } # Profile identity for per-profile provider scoping try: @@ -1090,7 +1096,7 @@ def __init__( self._memory_manager = None # Inject memory provider tool schemas into the tool surface - if self._memory_manager and self.tools is not None: + if self._memory_manager and self._provider_tool_access and self.tools is not None: for _schema in self._memory_manager.get_all_tool_schemas(): _wrapped = {"type": "function", "function": _schema} self.tools.append(_wrapped) @@ -5586,6 +5592,116 @@ def _execute_tool_calls(self, assistant_message, messages: list, effective_task_ finally: self._executing_tools = False + def _handle_builtin_memory_tool(self, function_args: dict) -> str: + """Handle the built-in memory tool with optional write protection.""" + if not self._memory_write_enabled: + return json.dumps({ + "success": False, + "error": "Memory writes are disabled for this agent.", + }) + + target = function_args.get("target", "memory") + from tools.memory_tool import memory_tool as _memory_tool + result = _memory_tool( + action=function_args.get("action"), + target=target, + content=function_args.get("content"), + old_text=function_args.get("old_text"), + store=self._memory_store, + ) + + if self._memory_manager and function_args.get("action") in ("add", "replace"): + try: + self._memory_manager.on_memory_write( + function_args.get("action", ""), + target, + function_args.get("content", ""), + ) + except Exception: + pass + + return result + + def _handle_provider_memory_tool(self, function_name: str, function_args: dict) -> str: + """Handle external memory-provider tools with optional access control.""" + if not self._provider_tool_access: + return json.dumps({ + "success": False, + "error": "Memory provider tools are disabled for this agent.", + }) + return self._memory_manager.handle_tool_call(function_name, function_args) + + def _handle_background_subagent_tool(self, function_name: str, function_args: dict) -> str: + from agent.background_subagents import get_background_subagent_manager + + manager = get_background_subagent_manager() + owner_session_id = str(self.session_id or "").strip() + if not owner_session_id: + return json.dumps({ + "success": False, + "error": "Background subagents require a parent session_id.", + }) + + if function_name == "spawn_background_subagent": + result = manager.spawn_subagent( + owner_session_id=owner_session_id, + purpose=function_args.get("purpose", ""), + initial_task=function_args.get("initial_task", ""), + cwd=function_args.get("cwd", ""), + agent_kind=function_args.get("agent_kind"), + ) + elif function_name == "list_background_subagents": + result = manager.list_subagents(owner_session_id=owner_session_id) + elif function_name == "send_background_subagent": + result = manager.send_message( + owner_session_id=owner_session_id, + subagent_id=function_args.get("id", ""), + message=function_args.get("message", ""), + ) + elif function_name == "poll_background_subagent": + result = manager.poll_subagent( + owner_session_id=owner_session_id, + subagent_id=function_args.get("id", ""), + since_seq=function_args.get("since_seq"), + ) + elif function_name == "get_background_subagent_status": + result = manager.get_status( + owner_session_id=owner_session_id, + subagent_id=function_args.get("id", ""), + ) + elif function_name == "stop_background_subagent": + result = manager.stop_subagent( + owner_session_id=owner_session_id, + subagent_id=function_args.get("id", ""), + reason=function_args.get("reason", ""), + ) + else: + result = { + "success": False, + "error": f"Unknown background subagent tool: {function_name}", + } + return json.dumps(result) + + def _build_background_subagent_context(self) -> str: + owner_session_id = str(self.session_id or "").strip() + if not owner_session_id: + return "" + try: + from agent.background_subagents import get_background_subagent_manager + from agent.async_delegate_tasks import get_async_delegate_manager + + parts = [] + subagent_context = get_background_subagent_manager().render_turn_context(owner_session_id) + if subagent_context: + parts.append(subagent_context) + async_delegate_context = get_async_delegate_manager().render_turn_context(owner_session_id) + if async_delegate_context: + parts.append(async_delegate_context) + return "\n\n".join(parts) + except Exception: + logger.debug("Failed to render background subagent context", exc_info=True) + return "" + def _invoke_tool(self, function_name: str, function_args: dict, effective_task_id: str) -> str: """Invoke a single tool and return the result string. No display logic. @@ -5612,28 +5728,9 @@ def _invoke_tool(self, function_name: str, function_args: dict, effective_task_i current_session_id=self.session_id, ) elif function_name == "memory": - target = function_args.get("target", "memory") - from tools.memory_tool import memory_tool as _memory_tool - result = _memory_tool( - action=function_args.get("action"), - target=target, - content=function_args.get("content"), - old_text=function_args.get("old_text"), - store=self._memory_store, - ) - # Bridge: notify external memory provider of built-in memory writes - if self._memory_manager and function_args.get("action") in ("add", "replace"): - try: - self._memory_manager.on_memory_write( - function_args.get("action", ""), - target, - function_args.get("content", ""), - ) - except Exception: - pass - return result + return self._handle_builtin_memory_tool(function_args) elif self._memory_manager and self._memory_manager.has_tool(function_name): - return self._memory_manager.handle_tool_call(function_name, function_args) + return self._handle_provider_memory_tool(function_name, function_args) elif function_name == "clarify": from tools.clarify_tool import clarify_tool as _clarify_tool return _clarify_tool( @@ -5647,10 +5744,21 @@ def _invoke_tool(self, function_name: str, function_args: dict, effective_task_i goal=function_args.get("goal"), context=function_args.get("context"), toolsets=function_args.get("toolsets"), + profile=function_args.get("profile"), + async_mode=bool(function_args.get("async", False)), tasks=function_args.get("tasks"), max_iterations=function_args.get("max_iterations"), parent_agent=self, ) + elif function_name in { + "spawn_background_subagent", + "list_background_subagents", + "send_background_subagent", + "poll_background_subagent", + "get_background_subagent_status", + "stop_background_subagent", + }: + return self._handle_background_subagent_tool(function_name, function_args) else: return handle_function_call( function_name, function_args, effective_task_id, @@ -5969,15 +6077,7 @@ def _execute_tool_calls_sequential(self, assistant_message, messages: list, effe if self.quiet_mode: self._vprint(f" {_get_cute_tool_message_impl('session_search', function_args, tool_duration, result=function_result)}") elif function_name == "memory": - target = function_args.get("target", "memory") - from tools.memory_tool import memory_tool as _memory_tool - function_result = _memory_tool( - action=function_args.get("action"), - target=target, - content=function_args.get("content"), - old_text=function_args.get("old_text"), - store=self._memory_store, - ) + function_result = self._handle_builtin_memory_tool(function_args) tool_duration = time.time() - tool_start_time if self.quiet_mode: self._vprint(f" {_get_cute_tool_message_impl('memory', function_args, tool_duration, result=function_result)}") @@ -6011,6 +6111,8 @@ def _execute_tool_calls_sequential(self, assistant_message, messages: list, effe goal=function_args.get("goal"), context=function_args.get("context"), toolsets=function_args.get("toolsets"), + profile=function_args.get("profile"), + async_mode=bool(function_args.get("async", False)), tasks=tasks_arg, max_iterations=function_args.get("max_iterations"), parent_agent=self, @@ -6024,6 +6126,18 @@ def _execute_tool_calls_sequential(self, assistant_message, messages: list, effe spinner.stop(cute_msg) elif self.quiet_mode: self._vprint(f" {cute_msg}") + elif function_name in { + "spawn_background_subagent", + "list_background_subagents", + "send_background_subagent", + "poll_background_subagent", + "get_background_subagent_status", + "stop_background_subagent", + }: + function_result = self._handle_background_subagent_tool(function_name, function_args) + tool_duration = time.time() - tool_start_time + if self.quiet_mode: + self._vprint(f" {_get_cute_tool_message_impl(function_name, function_args, tool_duration, result=function_result)}") elif self._memory_manager and self._memory_manager.has_tool(function_name): # Memory provider tools (hindsight_retain, honcho_search, etc.) # These are not in the tool registry — route through MemoryManager. @@ -6036,7 +6150,7 @@ def _execute_tool_calls_sequential(self, assistant_message, messages: list, effe spinner.start() _mem_result = None try: - function_result = self._memory_manager.handle_tool_call(function_name, function_args) + function_result = self._handle_provider_memory_tool(function_name, function_args) _mem_result = function_result except Exception as tool_error: function_result = json.dumps({"error": f"Memory tool '{function_name}' failed: {tool_error}"}) @@ -6259,6 +6373,9 @@ def _handle_max_iterations(self, messages: list, api_call_count: int) -> str: effective_system = self._cached_system_prompt or "" if self.ephemeral_system_prompt: effective_system = (effective_system + "\n\n" + self.ephemeral_system_prompt).strip() + background_context = self._build_background_subagent_context() + if background_context: + effective_system = (effective_system + "\n\n" + background_context).strip() if effective_system: api_messages = [{"role": "system", "content": effective_system}] + api_messages if self.prefill_messages: @@ -6771,6 +6888,9 @@ def run_conversation( effective_system = active_system_prompt or "" if self.ephemeral_system_prompt: effective_system = (effective_system + "\n\n" + self.ephemeral_system_prompt).strip() + background_context = self._build_background_subagent_context() + if background_context: + effective_system = (effective_system + "\n\n" + background_context).strip() # Plugin context from pre_llm_call hooks — ephemeral, not cached. if _plugin_turn_context: effective_system = (effective_system + "\n\n" + _plugin_turn_context).strip() diff --git a/tests/agent/test_async_delegate_tasks.py b/tests/agent/test_async_delegate_tasks.py new file mode 100644 index 000000000000..9a11ff63f6f9 --- /dev/null +++ b/tests/agent/test_async_delegate_tasks.py @@ -0,0 +1,338 @@ +import threading +import time +from pathlib import Path +from types import SimpleNamespace + +import pytest + +from agent import async_delegate_tasks as adt + + +class FakeChild: + def __init__(self): + self.session_id = "child-session-1" + self.tool_progress_callback = None + self._delegate_terminal_overrides = {"env_type": "docker"} + self.interrupted = False + + def interrupt(self): + self.interrupted = True + + +@pytest.fixture() +def manager(monkeypatch, tmp_path): + mgr = adt.get_async_delegate_manager() + mgr.reset_for_tests() + monkeypatch.setattr( + adt, + "_load_async_delegate_config", + lambda: { + "enabled": True, + "max_per_session": 2, + "max_global": 4, + "idle_timeout_seconds": 10, + "max_duration_seconds": 20, + "output_dir": ".hermes-async-delegates", + }, + ) + monkeypatch.setattr(adt, "_resolve_workspace_root", lambda parent: tmp_path) + yield mgr + mgr.reset_for_tests() + + +def test_async_delegate_writes_workspace_file_and_nudges_completion(manager, monkeypatch, tmp_path): + child = FakeChild() + run_gate = threading.Event() + parent_tool_names = ["terminal", "file", "web"] + child_tool_names = ["delegate_task", "terminal"] + + monkeypatch.setattr("tools.delegate_tool._load_config", lambda: {"max_iterations": 50}) + monkeypatch.setattr( + "tools.delegate_tool._resolve_delegation_profile", + lambda cfg, profile: {"name": profile or "", "toolsets": ["terminal", "file"], "memory": "none", "provider_tools": False, "terminal": {"backend": "docker"}}, + ) + monkeypatch.setattr(adt, "_resolve_workspace_root", lambda parent: tmp_path / "workspace") + + def _fake_build_child_agent(**kwargs): + import model_tools + + model_tools._last_resolved_tool_names = list(child_tool_names) + return child + + monkeypatch.setattr("tools.delegate_tool._build_child_agent", _fake_build_child_agent) + + def _fake_run_single_child(**kwargs): + run_gate.wait(timeout=2) + return { + "task_index": 0, + "status": "completed", + "summary": "Likely root cause is stale cache state.", + "api_calls": 3, + "duration_seconds": 1.0, + } + + monkeypatch.setattr("tools.delegate_tool._run_single_child", _fake_run_single_child) + + workspace_root = tmp_path / "workspace" + workspace_root.mkdir() + parent = SimpleNamespace(session_id="session-1", _delegate_depth=0, _current_workspace=str(workspace_root)) + + import model_tools + + original_tool_names = list(getattr(model_tools, "_last_resolved_tool_names", [])) + model_tools._last_resolved_tool_names = list(parent_tool_names) + spawned = manager.spawn( + owner_session_id="session-1", + parent_agent=parent, + goal="Investigate flaky tests", + context="Check cache state first", + toolsets=["terminal", "file"], + profile="friendly", + max_iterations=30, + creds={"model": None, "provider": None, "base_url": None, "api_key": None, "api_mode": None}, + ) + + assert spawned["success"] is True + output_file = Path(spawned["output_file"]) + assert output_file.exists() + assert str(workspace_root / ".hermes-async-delegates") in spawned["output_file"] + assert child._delegate_terminal_overrides["docker_mount_cwd_to_workspace"] is True + assert child._delegate_terminal_overrides["cwd"] == "/workspace" + assert child._delegate_saved_tool_names == parent_tool_names + assert model_tools._last_resolved_tool_names == parent_tool_names + + child.tool_progress_callback("terminal", "pytest -q") + run_gate.set() + + deadline = time.time() + 2 + while time.time() < deadline: + record = next(iter(manager._records.values())) + if record.status == "completed": + break + time.sleep(0.05) + + record = next(iter(manager._records.values())) + assert record.status == "completed" + contents = output_file.read_text(encoding="utf-8") + assert "pytest -q" in contents + assert "Likely root cause is stale cache state." in contents + + context = manager.render_turn_context("session-1") + assert "completed" in context + assert record.id in context + assert manager.render_turn_context("session-1") == "" + model_tools._last_resolved_tool_names = original_tool_names + + +def test_resolve_workspace_root_prefers_parent_workspace_over_process_cwd(tmp_path, monkeypatch): + process_cwd = tmp_path / "process-cwd" + process_cwd.mkdir() + workspace_root = tmp_path / "workspace" + workspace_root.mkdir() + monkeypatch.chdir(process_cwd) + + parent = SimpleNamespace(_current_workspace=str(workspace_root)) + assert adt._resolve_workspace_root(parent) == workspace_root.resolve() + + +def test_async_delegate_renders_idle_without_nudge(manager, monkeypatch): + child = FakeChild() + run_gate = threading.Event() + + monkeypatch.setattr("tools.delegate_tool._load_config", lambda: {"max_iterations": 50}) + monkeypatch.setattr( + "tools.delegate_tool._resolve_delegation_profile", + lambda cfg, profile: {"name": profile or "", "toolsets": ["terminal"], "memory": "none", "provider_tools": False, "terminal": {"backend": "docker"}}, + ) + monkeypatch.setattr("tools.delegate_tool._build_child_agent", lambda **kwargs: child) + monkeypatch.setattr( + "tools.delegate_tool._run_single_child", + lambda **kwargs: (run_gate.wait(timeout=2), {"task_index": 0, "status": "completed", "summary": "done", "api_calls": 1, "duration_seconds": 1.0})[1], + ) + + parent = SimpleNamespace(session_id="session-1", _delegate_depth=0) + spawned = manager.spawn( + owner_session_id="session-1", + parent_agent=parent, + goal="Long task", + context="", + toolsets=["terminal"], + profile=None, + max_iterations=20, + creds={}, + ) + record = manager._records[spawned["id"]] + record.last_activity_at = adt._now() - 30 + + context = manager.render_turn_context("session-1") + assert "[idle]" in context + assert "Async delegated subagent updates" not in context + + run_gate.set() + + +def test_async_delegate_max_duration_interrupts_and_nudges(manager, monkeypatch): + child = FakeChild() + run_gate = threading.Event() + + monkeypatch.setattr("tools.delegate_tool._load_config", lambda: {"max_iterations": 50}) + monkeypatch.setattr( + "tools.delegate_tool._resolve_delegation_profile", + lambda cfg, profile: {"name": profile or "", "toolsets": ["terminal"], "memory": "none", "provider_tools": False, "terminal": {"backend": "docker"}}, + ) + monkeypatch.setattr("tools.delegate_tool._build_child_agent", lambda **kwargs: child) + monkeypatch.setattr( + "tools.delegate_tool._run_single_child", + lambda **kwargs: (run_gate.wait(timeout=2), {"task_index": 0, "status": "completed", "summary": "partial", "api_calls": 1, "duration_seconds": 1.0})[1], + ) + + parent = SimpleNamespace(session_id="session-1", _delegate_depth=0) + spawned = manager.spawn( + owner_session_id="session-1", + parent_agent=parent, + goal="Time bounded task", + context="", + toolsets=["terminal"], + profile=None, + max_iterations=20, + creds={}, + ) + record = manager._records[spawned["id"]] + record.max_duration_at = adt._now() - 1 + + manager._sweep_once() + assert child.interrupted is True + assert record.status == "timed_out" + context = manager.render_turn_context("session-1") + assert "maximum runtime" in context + + run_gate.set() + + +def test_async_delegate_limits_enforced(manager, monkeypatch): + monkeypatch.setattr("tools.delegate_tool._load_config", lambda: {"max_iterations": 50}) + monkeypatch.setattr( + "tools.delegate_tool._resolve_delegation_profile", + lambda cfg, profile: {"name": profile or "", "toolsets": ["terminal"], "memory": "none", "provider_tools": False, "terminal": {"backend": "docker"}}, + ) + + run_gate = threading.Event() + + def _make_child(): + child = FakeChild() + child._delegate_terminal_overrides = {"env_type": "docker"} + return child + + monkeypatch.setattr("tools.delegate_tool._build_child_agent", lambda **kwargs: _make_child()) + monkeypatch.setattr( + "tools.delegate_tool._run_single_child", + lambda **kwargs: (run_gate.wait(timeout=2), {"task_index": 0, "status": "completed", "summary": "done", "api_calls": 1, "duration_seconds": 1.0})[1], + ) + + parent = SimpleNamespace(session_id="session-1", _delegate_depth=0, _current_workspace="/tmp") + first = manager.spawn( + owner_session_id="session-1", + parent_agent=parent, + goal="One", + toolsets=["terminal"], + profile=None, + creds={}, + ) + second = manager.spawn( + owner_session_id="session-1", + parent_agent=parent, + goal="Two", + toolsets=["terminal"], + profile=None, + creds={}, + ) + assert first["success"] is True + assert second["success"] is True + + limited = manager.spawn( + owner_session_id="session-1", + parent_agent=parent, + goal="Three", + toolsets=["terminal"], + profile=None, + creds={}, + ) + assert limited["success"] is False + assert "limit reached" in limited["error"].lower() + assert len(limited["active_delegates"]) == 2 + run_gate.set() + + +def test_async_delegate_failure_sets_error_and_nudges(manager, monkeypatch, tmp_path): + child = FakeChild() + + monkeypatch.setattr("tools.delegate_tool._load_config", lambda: {"max_iterations": 50}) + monkeypatch.setattr( + "tools.delegate_tool._resolve_delegation_profile", + lambda cfg, profile: {"name": profile or "", "toolsets": ["terminal"], "memory": "none", "provider_tools": False, "terminal": {"backend": "docker"}}, + ) + monkeypatch.setattr("tools.delegate_tool._build_child_agent", lambda **kwargs: child) + monkeypatch.setattr( + "tools.delegate_tool._run_single_child", + lambda **kwargs: {"task_index": 0, "status": "error", "error": "boom", "api_calls": 1, "duration_seconds": 1.0}, + ) + + workspace_root = tmp_path / "workspace" + workspace_root.mkdir() + parent = SimpleNamespace(session_id="session-1", _delegate_depth=0, _current_workspace=str(workspace_root)) + + spawned = manager.spawn( + owner_session_id="session-1", + parent_agent=parent, + goal="Failing task", + context="", + toolsets=["terminal"], + profile=None, + creds={}, + ) + + deadline = time.time() + 2 + record = manager._records[spawned["id"]] + while time.time() < deadline and record.is_alive(): + time.sleep(0.05) + if record.thread: + record.thread.join(timeout=0.5) + + assert record.status == "error" + contents = Path(spawned["output_file"]).read_text(encoding="utf-8") + assert "Error: boom" in contents + context = manager.render_turn_context("session-1") + assert "failed: boom" in context + + +def test_async_delegate_non_docker_preserves_terminal_overrides(manager, monkeypatch): + child = FakeChild() + child._delegate_terminal_overrides = {"env_type": "local", "cwd": "/tmp/existing"} + run_gate = threading.Event() + + monkeypatch.setattr("tools.delegate_tool._load_config", lambda: {"max_iterations": 50}) + monkeypatch.setattr( + "tools.delegate_tool._resolve_delegation_profile", + lambda cfg, profile: {"name": profile or "", "toolsets": ["terminal"], "memory": "none", "provider_tools": False, "terminal": {"backend": "local"}}, + ) + monkeypatch.setattr("tools.delegate_tool._build_child_agent", lambda **kwargs: child) + monkeypatch.setattr( + "tools.delegate_tool._run_single_child", + lambda **kwargs: (run_gate.wait(timeout=2), {"task_index": 0, "status": "completed", "summary": "done", "api_calls": 1, "duration_seconds": 1.0})[1], + ) + + parent = SimpleNamespace(session_id="session-1", _delegate_depth=0, _current_workspace="/tmp") + spawned = manager.spawn( + owner_session_id="session-1", + parent_agent=parent, + goal="Local task", + toolsets=["terminal"], + profile=None, + creds={}, + ) + + assert spawned["success"] is True + assert child._delegate_terminal_overrides["env_type"] == "local" + assert child._delegate_terminal_overrides["cwd"] == "/tmp/existing" + assert "docker_mount_cwd_to_workspace" not in child._delegate_terminal_overrides + run_gate.set() diff --git a/tests/agent/test_background_subagents.py b/tests/agent/test_background_subagents.py new file mode 100644 index 000000000000..4f661f94f8c1 --- /dev/null +++ b/tests/agent/test_background_subagents.py @@ -0,0 +1,438 @@ +import json + +import pytest + +from agent import background_subagents as bg + + +class FakeExecSession: + def __init__(self): + self.stdout_handler = None + self.stderr_handler = None + self.exit_handler = None + self.alive = True + self.client_responses = [] + self.pending_prompts = [] + self.writes = [] + self.pid = 4321 + + def read_loop(self, *, stdout_handler=None, stderr_handler=None, exit_handler=None): + self.stdout_handler = stdout_handler + self.stderr_handler = stderr_handler + self.exit_handler = exit_handler + + def write_line(self, payload: str) -> None: + msg = json.loads(payload) + self.writes.append(msg) + method = msg.get("method") + if method == "initialize": + self._emit({"jsonrpc": "2.0", "id": msg["id"], "result": {}}) + elif method == "session/new": + self._emit({"jsonrpc": "2.0", "id": msg["id"], "result": {"sessionId": "acp-session-1"}}) + elif method == "session/prompt": + self.pending_prompts.append(msg) + self._emit( + { + "jsonrpc": "2.0", + "method": "session/update", + "params": { + "update": { + "sessionUpdate": "agent_message_chunk", + "content": {"text": f"working:{msg['params']['prompt'][0]['text']}"}, + } + }, + } + ) + elif method == "session/cancel": + self._emit({"jsonrpc": "2.0", "id": msg["id"], "result": {}}) + elif "result" in msg or "error" in msg: + self.client_responses.append(msg) + + def complete_prompt(self, *, stop_reason: str = "end_turn") -> None: + prompt = self.pending_prompts.pop(0) + self._emit( + { + "jsonrpc": "2.0", + "id": prompt["id"], + "result": {"stopReason": stop_reason}, + } + ) + + def emit_stderr(self, text: str) -> None: + if self.stderr_handler: + self.stderr_handler(text) + + def emit_raw_stdout(self, line: str) -> None: + if self.stdout_handler: + self.stdout_handler(line) + + def is_session_alive(self) -> bool: + return self.alive + + def terminate_session(self) -> None: + if not self.alive: + return + self.alive = False + if self.exit_handler: + self.exit_handler(0) + + def _emit(self, payload: dict) -> None: + if self.stdout_handler: + self.stdout_handler(json.dumps(payload)) + + +class FakeEnvironment: + def __init__(self, exec_session: FakeExecSession): + self.exec_session = exec_session + self.cleanup_called = False + self.start_args = None + self._container_id = "container-123" + + def start_persistent_exec(self, *, cwd="", command=None, env=None): + self.start_args = { + "cwd": cwd, + "command": list(command or []), + "env": dict(env or {}), + } + return self.exec_session + + def cleanup(self): + self.cleanup_called = True + + +@pytest.fixture() +def manager(monkeypatch): + mgr = bg.get_background_subagent_manager() + mgr.reset_for_tests() + monkeypatch.setattr( + bg, + "_load_background_subagent_config", + lambda: { + "enabled": True, + "max_per_session": 2, + "max_global": 4, + "idle_timeout_seconds": 900, + "max_lifetime_seconds": 7200, + "default_agent_kind": "opencode", + "agents": { + "opencode": { + "command": "opencode", + "args": ["acp"], + "cwd_mode": "session", + } + }, + }, + ) + yield mgr + mgr.reset_for_tests() + + +def test_background_subagent_lifecycle(manager, monkeypatch): + exec_session = FakeExecSession() + environment = FakeEnvironment(exec_session) + monkeypatch.setattr(manager, "_create_environment", lambda **kwargs: environment) + + spawned = manager.spawn_subagent( + owner_session_id="session-1", + purpose="Track a long-running code investigation", + initial_task="Inspect failing tests", + cwd="/workspace", + ) + + assert spawned["success"] is True + subagent_id = spawned["id"] + assert environment.start_args["command"] == ["opencode", "acp"] + + listed = manager.list_subagents(owner_session_id="session-1") + assert [item["id"] for item in listed["subagents"]] == [subagent_id] + assert listed["subagents"][0]["status"] == "running" + + polled = manager.poll_subagent(owner_session_id="session-1", subagent_id=subagent_id) + kinds = [event["kind"] for event in polled["events"]] + assert "task_dispatched" in kinds + assert "agent_message_chunk" in kinds + + exec_session.complete_prompt() + status = manager.get_status(owner_session_id="session-1", subagent_id=subagent_id) + assert status["status"] == "idle" + + context = manager.render_turn_context("session-1") + assert "finished its latest task" in context + assert subagent_id in context + assert "Track a long-running code investigation" in context + assert "Inspect failing tests" not in manager.render_turn_context("session-1") + + send = manager.send_message( + owner_session_id="session-1", + subagent_id=subagent_id, + message="Inspect the Docker environment next", + ) + assert send["success"] is True + assert send["queued"] is False + + queued = manager.send_message( + owner_session_id="session-1", + subagent_id=subagent_id, + message="Then summarize the likely root cause", + ) + assert queued["success"] is True + assert queued["queued"] is True + + exec_session.complete_prompt() + mid_status = manager.get_status(owner_session_id="session-1", subagent_id=subagent_id) + assert mid_status["status"] == "running" + exec_session.complete_prompt() + final_status = manager.get_status(owner_session_id="session-1", subagent_id=subagent_id) + assert final_status["status"] == "idle" + + stopped = manager.stop_subagent( + owner_session_id="session-1", + subagent_id=subagent_id, + reason="done", + ) + assert stopped["success"] is True + assert stopped["status"] == "stopped" + assert environment.cleanup_called is True + + +def test_background_subagent_limits(manager, monkeypatch): + exec_one = FakeExecSession() + env_one = FakeEnvironment(exec_one) + monkeypatch.setattr(manager, "_create_environment", lambda **kwargs: env_one) + + manager.spawn_subagent( + owner_session_id="session-1", + purpose="First", + initial_task="One", + cwd="/workspace", + ) + + monkeypatch.setattr( + bg, + "_load_background_subagent_config", + lambda: { + "enabled": True, + "max_per_session": 1, + "max_global": 1, + "idle_timeout_seconds": 900, + "max_lifetime_seconds": 7200, + "default_agent_kind": "opencode", + "agents": {"opencode": {"command": "opencode", "args": ["acp"], "cwd_mode": "session"}}, + }, + ) + + limited = manager.spawn_subagent( + owner_session_id="session-1", + purpose="Second", + initial_task="Two", + cwd="/workspace", + ) + assert limited["success"] is False + assert "limit reached" in limited["error"].lower() + assert len(limited["stoppable_subagents"]) == 1 + + +def test_background_subagent_idle_timeout(manager, monkeypatch): + exec_session = FakeExecSession() + environment = FakeEnvironment(exec_session) + monkeypatch.setattr(manager, "_create_environment", lambda **kwargs: environment) + + spawned = manager.spawn_subagent( + owner_session_id="session-1", + purpose="Timeout me", + initial_task="Start work", + cwd="/workspace", + ) + exec_session.complete_prompt() + record = manager._get_owned_record("session-1", spawned["id"]) + assert record is not None + record.idle_timeout_at = bg._now() - 1 + + manager._sweep_once() + + status = manager.get_status(owner_session_id="session-1", subagent_id=spawned["id"]) + assert status["status"] == "timed_out" + assert environment.cleanup_called is True + + +def test_background_subagent_protocol_error_stops_session(manager, monkeypatch): + exec_session = FakeExecSession() + environment = FakeEnvironment(exec_session) + monkeypatch.setattr(manager, "_create_environment", lambda **kwargs: environment) + + spawned = manager.spawn_subagent( + owner_session_id="session-1", + purpose="Break protocol", + initial_task="Start work", + cwd="/workspace", + ) + exec_session.emit_raw_stdout("not-json") + + status = manager.get_status(owner_session_id="session-1", subagent_id=spawned["id"]) + assert status["status"] == "error" + context = manager.render_turn_context("session-1") + assert "protocol_error" in context + + +def test_background_subagent_max_lifetime_times_out_and_nudges(manager, monkeypatch): + exec_session = FakeExecSession() + environment = FakeEnvironment(exec_session) + monkeypatch.setattr(manager, "_create_environment", lambda **kwargs: environment) + + spawned = manager.spawn_subagent( + owner_session_id="session-1", + purpose="Lifetime bounded", + initial_task="Start work", + cwd="/workspace", + ) + record = manager._get_owned_record("session-1", spawned["id"]) + assert record is not None + record.max_lifetime_at = bg._now() - 1 + + manager._sweep_once() + + status = manager.get_status(owner_session_id="session-1", subagent_id=spawned["id"]) + assert status["status"] == "timed_out" + assert status["terminal_reason"] == "max_lifetime_exceeded" + context = manager.render_turn_context("session-1") + assert "max_lifetime_exceeded" in context + assert environment.cleanup_called is True + + +def test_background_subagent_tools_reject_wrong_owner(manager, monkeypatch): + exec_session = FakeExecSession() + environment = FakeEnvironment(exec_session) + monkeypatch.setattr(manager, "_create_environment", lambda **kwargs: environment) + + spawned = manager.spawn_subagent( + owner_session_id="session-1", + purpose="Owner scoped", + initial_task="Start work", + cwd="/workspace", + ) + subagent_id = spawned["id"] + + for result in ( + manager.get_status(owner_session_id="session-2", subagent_id=subagent_id), + manager.send_message(owner_session_id="session-2", subagent_id=subagent_id, message="hello"), + manager.poll_subagent(owner_session_id="session-2", subagent_id=subagent_id), + manager.stop_subagent(owner_session_id="session-2", subagent_id=subagent_id), + ): + assert result["success"] is False + assert "Unknown background subagent" in result["error"] + + +def test_background_subagent_stop_sends_cancel_and_cleans_up(manager, monkeypatch): + exec_session = FakeExecSession() + environment = FakeEnvironment(exec_session) + monkeypatch.setattr(manager, "_create_environment", lambda **kwargs: environment) + + spawned = manager.spawn_subagent( + owner_session_id="session-1", + purpose="Stop me", + initial_task="Start work", + cwd="/workspace", + ) + + stopped = manager.stop_subagent( + owner_session_id="session-1", + subagent_id=spawned["id"], + reason="done", + ) + + assert stopped["success"] is True + assert stopped["status"] == "stopped" + assert environment.cleanup_called is True + cancel_methods = [msg["method"] for msg in exec_session.writes if msg.get("method") == "session/cancel"] + assert cancel_methods == ["session/cancel"] + + +def test_background_subagent_poll_since_seq_returns_incremental_events(manager, monkeypatch): + exec_session = FakeExecSession() + environment = FakeEnvironment(exec_session) + monkeypatch.setattr(manager, "_create_environment", lambda **kwargs: environment) + + spawned = manager.spawn_subagent( + owner_session_id="session-1", + purpose="Incremental polling", + initial_task="Inspect failing tests", + cwd="/workspace", + ) + subagent_id = spawned["id"] + + first_poll = manager.poll_subagent(owner_session_id="session-1", subagent_id=subagent_id) + assert first_poll["events"] + assert first_poll["unread_count"] == 0 + last_seq = first_poll["events"][-1]["seq"] + + exec_session.complete_prompt() + + manager.send_message( + owner_session_id="session-1", + subagent_id=subagent_id, + message="Inspect the Docker environment next", + ) + second_poll = manager.poll_subagent( + owner_session_id="session-1", + subagent_id=subagent_id, + since_seq=last_seq, + ) + assert second_poll["events"] + assert all(event["seq"] > last_seq for event in second_poll["events"]) + + +def test_background_subagent_inbound_permission_request_gets_response(manager, monkeypatch): + exec_session = FakeExecSession() + environment = FakeEnvironment(exec_session) + monkeypatch.setattr(manager, "_create_environment", lambda **kwargs: environment) + + manager.spawn_subagent( + owner_session_id="session-1", + purpose="Permission checks", + initial_task="Start work", + cwd="/workspace", + ) + + exec_session.emit_raw_stdout( + json.dumps( + { + "jsonrpc": "2.0", + "id": 99, + "method": "session/request_permission", + "params": {"tool": "terminal"}, + } + ) + ) + + assert exec_session.client_responses + response = exec_session.client_responses[-1] + assert response["id"] == 99 + assert response["result"]["outcome"]["outcome"] == "allow_once" + + +def test_background_subagent_unsupported_inbound_method_returns_jsonrpc_error(manager, monkeypatch): + exec_session = FakeExecSession() + environment = FakeEnvironment(exec_session) + monkeypatch.setattr(manager, "_create_environment", lambda **kwargs: environment) + + manager.spawn_subagent( + owner_session_id="session-1", + purpose="Unsupported inbound", + initial_task="Start work", + cwd="/workspace", + ) + + exec_session.emit_raw_stdout( + json.dumps( + { + "jsonrpc": "2.0", + "id": 101, + "method": "session/run_tool", + "params": {"tool": "terminal"}, + } + ) + ) + + response = exec_session.client_responses[-1] + assert response["id"] == 101 + assert response["error"]["code"] == -32601 diff --git a/tests/test_cli_init.py b/tests/test_cli_init.py index b926d55f535d..1c00ce5dc4de 100644 --- a/tests/test_cli_init.py +++ b/tests/test_cli_init.py @@ -331,6 +331,24 @@ def test_normalize_root_model_keys_does_not_override_existing(self): assert "provider" not in result # root key still cleaned up +class TestDelegationDefaults: + def test_load_cli_config_keeps_async_and_background_subagent_defaults(self, tmp_path, monkeypatch): + hermes_home = tmp_path / ".hermes" + hermes_home.mkdir() + monkeypatch.setenv("HERMES_HOME", str(hermes_home)) + + import cli + + monkeypatch.setattr(cli, "_hermes_home", hermes_home) + cfg = cli.load_cli_config() + + delegation = cfg["delegation"] + assert delegation["max_iterations"] == 45 + assert delegation["default_toolsets"] == ["terminal", "file", "web"] + assert delegation["async_subagents"]["enabled"] is True + assert delegation["background_subagents"]["default_agent_kind"] == "opencode" + + class TestProviderResolution: def test_api_key_is_string_or_none(self): cli = _make_cli() diff --git a/tests/test_model_tools.py b/tests/test_model_tools.py index 8c2f8e6f7843..35deb8174f2c 100644 --- a/tests/test_model_tools.py +++ b/tests/test_model_tools.py @@ -49,6 +49,12 @@ def test_expected_tools_in_set(self): assert "memory" in _AGENT_LOOP_TOOLS assert "session_search" in _AGENT_LOOP_TOOLS assert "delegate_task" in _AGENT_LOOP_TOOLS + assert "spawn_background_subagent" in _AGENT_LOOP_TOOLS + assert "list_background_subagents" in _AGENT_LOOP_TOOLS + assert "send_background_subagent" in _AGENT_LOOP_TOOLS + assert "poll_background_subagent" in _AGENT_LOOP_TOOLS + assert "get_background_subagent_status" in _AGENT_LOOP_TOOLS + assert "stop_background_subagent" in _AGENT_LOOP_TOOLS def test_no_regular_tools_in_set(self): assert "web_search" not in _AGENT_LOOP_TOOLS diff --git a/tests/test_run_agent.py b/tests/test_run_agent.py index 9217117e2570..527e0b620e94 100644 --- a/tests/test_run_agent.py +++ b/tests/test_run_agent.py @@ -1300,6 +1300,50 @@ def test_invoke_tool_handles_agent_level_tools(self, agent): mock_todo.assert_called_once() assert "ok" in result + def test_invoke_tool_blocks_memory_writes_when_disabled(self, agent_with_memory_tool): + agent = agent_with_memory_tool + agent._memory_write_enabled = False + agent._memory_store = MagicMock() + + result = json.loads( + agent._invoke_tool( + "memory", + {"action": "add", "target": "memory", "content": "blocked"}, + "task-1", + ) + ) + + assert result["success"] is False + assert "disabled" in result["error"].lower() + + def test_invoke_tool_blocks_provider_memory_tools_when_disabled(self, agent): + manager = MagicMock() + manager.has_tool.return_value = True + agent._memory_manager = manager + agent._provider_tool_access = False + + result = json.loads( + agent._invoke_tool("honcho_search", {"query": "test"}, "task-1") + ) + + assert result["success"] is False + assert "disabled" in result["error"].lower() + manager.handle_tool_call.assert_not_called() + + def test_invoke_tool_routes_provider_memory_tools_when_enabled(self, agent): + manager = MagicMock() + manager.has_tool.return_value = True + manager.handle_tool_call.return_value = '{"ok": true}' + agent._memory_manager = manager + agent._provider_tool_access = True + + result = agent._invoke_tool("honcho_search", {"query": "test"}, "task-1") + + assert json.loads(result)["ok"] is True + manager.handle_tool_call.assert_called_once_with( + "honcho_search", {"query": "test"} + ) + class TestPathsOverlap: """Unit tests for the _paths_overlap helper.""" @@ -1428,6 +1472,41 @@ def test_stop_finish_reason_returns_response(self, agent): assert result["final_response"] == "Final answer" assert result["completed"] is True + def test_background_subagent_context_is_injected_ephemerally(self, agent): + self._setup_agent(agent) + resp = _mock_response(content="Final answer", finish_reason="stop") + agent.client.chat.completions.create.return_value = resp + with ( + patch.object(agent, "_build_background_subagent_context", return_value="BG roster"), + patch.object(agent, "_persist_session"), + patch.object(agent, "_save_trajectory"), + patch.object(agent, "_cleanup_task_resources"), + ): + agent.run_conversation("hello") + + kwargs = agent.client.chat.completions.create.call_args.kwargs + messages = kwargs["messages"] + assert messages[0]["role"] == "system" + assert "BG roster" in messages[0]["content"] + + def test_background_context_combines_acp_and_async_delegate_sections(self, agent): + class _Mgr: + def __init__(self, text): + self.text = text + + def render_turn_context(self, session_id): + return self.text + + agent.session_id = "session-123" + with ( + patch("agent.background_subagents.get_background_subagent_manager", return_value=_Mgr("ACP roster")), + patch("agent.async_delegate_tasks.get_async_delegate_manager", return_value=_Mgr("Async delegate roster")), + ): + text = agent._build_background_subagent_context() + + assert "ACP roster" in text + assert "Async delegate roster" in text + def test_tool_calls_then_stop(self, agent): self._setup_agent(agent) tc = _mock_tool_call(name="web_search", arguments="{}", call_id="c1") diff --git a/tests/tools/test_delegate.py b/tests/tools/test_delegate.py index d86a8c4889d0..756845e986a3 100644 --- a/tests/tools/test_delegate.py +++ b/tests/tools/test_delegate.py @@ -25,6 +25,7 @@ delegate_task, _build_child_agent, _build_child_system_prompt, + _resolve_delegation_profile, _strip_blocked_tools, _resolve_delegation_credentials, ) @@ -44,6 +45,7 @@ def _make_mock_parent(depth=0): parent.providers_order = None parent.provider_sort = None parent._session_db = None + parent.session_id = "parent-session" parent._delegate_depth = depth parent._active_children = [] parent._active_children_lock = threading.Lock() @@ -61,7 +63,9 @@ def test_schema_valid(self): self.assertIn("tasks", props) self.assertIn("context", props) self.assertIn("toolsets", props) + self.assertIn("profile", props) self.assertIn("max_iterations", props) + self.assertIn("async", props) self.assertEqual(props["tasks"]["maxItems"], 3) @@ -119,11 +123,60 @@ def test_empty_goal(self): result = json.loads(delegate_task(goal=" ", parent_agent=parent)) self.assertIn("error", result) + @patch("tools.delegate_tool._load_config", return_value={"max_iterations": 45, "default_profile": "missing"}) + def test_unknown_profile_returns_json_error(self, _mock_cfg): + parent = _make_mock_parent() + result = json.loads(delegate_task(goal="test", parent_agent=parent)) + self.assertIn("error", result) + self.assertIn("Unknown delegation profile", result["error"]) + def test_task_missing_goal(self): parent = _make_mock_parent() result = json.loads(delegate_task(tasks=[{"context": "no goal here"}], parent_agent=parent)) self.assertIn("error", result) + def test_async_mode_rejects_batch(self): + parent = _make_mock_parent() + result = json.loads(delegate_task( + tasks=[{"goal": "one"}], + async_mode=True, + parent_agent=parent, + )) + self.assertIn("error", result) + self.assertIn("single-task", result["error"]) + + @patch("agent.async_delegate_tasks.get_async_delegate_manager") + def test_async_mode_dispatches_to_background_manager(self, mock_get_manager): + manager = MagicMock() + manager.spawn.return_value = { + "success": True, + "mode": "async", + "id": "async-delegate-123", + "output_file": "/tmp/out.md", + } + mock_get_manager.return_value = manager + parent = _make_mock_parent() + + result = json.loads(delegate_task( + goal="Investigate flaky tests", + context="Only report likely causes", + toolsets=["terminal", "file"], + profile="friendly", + async_mode=True, + parent_agent=parent, + )) + + self.assertTrue(result["success"]) + self.assertEqual(result["mode"], "async") + manager.spawn.assert_called_once() + + def test_async_mode_requires_parent_session_id(self): + parent = _make_mock_parent() + parent.session_id = "" + result = json.loads(delegate_task(goal="Investigate", async_mode=True, parent_agent=parent)) + self.assertIn("error", result) + self.assertIn("session_id", result["error"]) + @patch("tools.delegate_tool._run_single_child") def test_single_task_mode(self, mock_run): mock_run.return_value = { @@ -185,6 +238,60 @@ def test_batch_ignores_toplevel_goal(self, mock_run): call_args = mock_run.call_args self.assertEqual(call_args.kwargs.get("goal") or call_args[1].get("goal", call_args[0][1] if len(call_args[0]) > 1 else None), "Actual task") + @patch("tools.delegate_tool._load_config", return_value={"max_iterations": 45}) + @patch("tools.delegate_tool._resolve_delegation_credentials") + @patch("tools.delegate_tool._build_child_agent") + @patch("tools.delegate_tool._run_single_child") + def test_top_level_profile_reaches_child_builder(self, mock_run, mock_build, mock_creds, _mock_cfg): + mock_creds.return_value = { + "model": None, + "provider": None, + "base_url": None, + "api_key": None, + "api_mode": None, + } + mock_build.return_value = MagicMock() + mock_run.return_value = { + "task_index": 0, "status": "completed", + "summary": "Done", "api_calls": 1, "duration_seconds": 1.0 + } + parent = _make_mock_parent() + + delegate_task(goal="Profile test", profile="friendly", parent_agent=parent) + + profile_arg = mock_build.call_args.kwargs["profile"] + self.assertEqual(profile_arg["name"], "friendly") + self.assertEqual(profile_arg["memory"], "read") + + @patch("tools.delegate_tool._load_config", return_value={"max_iterations": 45, "default_profile": "restricted"}) + @patch("tools.delegate_tool._resolve_delegation_credentials") + @patch("tools.delegate_tool._build_child_agent") + @patch("tools.delegate_tool._run_single_child") + def test_task_profile_overrides_top_level_or_default_profile(self, mock_run, mock_build, mock_creds, _mock_cfg): + mock_creds.return_value = { + "model": None, + "provider": None, + "base_url": None, + "api_key": None, + "api_mode": None, + } + mock_build.return_value = MagicMock() + mock_run.return_value = { + "task_index": 0, "status": "completed", + "summary": "Done", "api_calls": 1, "duration_seconds": 1.0 + } + parent = _make_mock_parent() + + delegate_task( + profile="friendly", + tasks=[{"goal": "Use privileged child", "profile": "privileged"}], + parent_agent=parent, + ) + + profile_arg = mock_build.call_args.kwargs["profile"] + self.assertEqual(profile_arg["name"], "privileged") + self.assertEqual(profile_arg["memory"], "write") + @patch("tools.delegate_tool._run_single_child") def test_failed_child_included_in_results(self, mock_run): mock_run.return_value = { @@ -309,6 +416,7 @@ def test_build_child_agent_does_not_raise_name_error(self): goal="regression check", context=None, toolsets=None, + profile=_resolve_delegation_profile({}, None), model=None, max_iterations=10, parent_agent=parent, @@ -333,8 +441,9 @@ def test_saved_tool_names_set_on_child_before_run(self): with patch("run_agent.AIAgent") as MockAgent: mock_child = MagicMock() - def capture_and_return(user_message): + def capture_and_return(user_message, task_id=None): captured["saved"] = list(mock_child._delegate_saved_tool_names) + captured["task_id"] = task_id return {"final_response": "ok", "completed": True, "api_calls": 1} mock_child.run_conversation.side_effect = capture_and_return @@ -343,6 +452,57 @@ def capture_and_return(user_message): delegate_task(goal="capture test", parent_agent=parent) self.assertEqual(captured["saved"], expected_tools) + self.assertTrue(captured["task_id"].startswith("delegate-")) + + def test_friendly_profile_configures_read_only_memory(self): + parent = _make_mock_parent(depth=0) + profile = _resolve_delegation_profile({}, "friendly") + + with patch("run_agent.AIAgent") as MockAgent: + mock_child = MagicMock() + MockAgent.return_value = mock_child + + _build_child_agent( + task_index=0, + goal="friendly child", + context=None, + toolsets=None, + profile=profile, + model=None, + max_iterations=10, + parent_agent=parent, + ) + + _, kwargs = MockAgent.call_args + self.assertFalse(kwargs["skip_memory"]) + self.assertFalse(kwargs["memory_write_enabled"]) + self.assertFalse(kwargs["provider_tool_access"]) + self.assertEqual(kwargs["agent_context"], "subagent") + + def test_privileged_profile_configures_writable_memory(self): + parent = _make_mock_parent(depth=0) + profile = _resolve_delegation_profile({}, "privileged") + + with patch("run_agent.AIAgent") as MockAgent: + mock_child = MagicMock() + MockAgent.return_value = mock_child + + _build_child_agent( + task_index=0, + goal="privileged child", + context=None, + toolsets=None, + profile=profile, + model=None, + max_iterations=10, + parent_agent=parent, + ) + + _, kwargs = MockAgent.call_args + self.assertFalse(kwargs["skip_memory"]) + self.assertTrue(kwargs["memory_write_enabled"]) + self.assertTrue(kwargs["provider_tool_access"]) + self.assertEqual(kwargs["agent_context"], "subagent") class TestDelegateObservability(unittest.TestCase): diff --git a/tests/tools/test_delegate_toolset_scope.py b/tests/tools/test_delegate_toolset_scope.py index d853dbb042c5..283d0decd896 100644 --- a/tests/tools/test_delegate_toolset_scope.py +++ b/tests/tools/test_delegate_toolset_scope.py @@ -1,66 +1,68 @@ -"""Tests for delegate_tool toolset scoping. +"""Tests for delegate_tool toolset scoping and delegation profiles.""" -Verifies that subagents cannot gain tools that the parent does not have. -The LLM controls the `toolsets` parameter — without intersection with the -parent's enabled_toolsets, it can escalate privileges by requesting -arbitrary toolsets. -""" - -from unittest.mock import MagicMock, patch from types import SimpleNamespace -from tools.delegate_tool import _strip_blocked_tools +from toolsets import TOOLSETS, resolve_toolset +from tools.delegate_tool import ( + _build_child_blocked_tools, + _build_child_enabled_toolsets, + _resolve_delegation_profile, + _strip_blocked_tools, +) class TestToolsetIntersection: - """Subagent toolsets must be a subset of parent's enabled_toolsets.""" - - def test_requested_toolsets_intersected_with_parent(self): - """LLM requests toolsets parent doesn't have — extras are dropped.""" - parent = SimpleNamespace(enabled_toolsets=["terminal", "file"]) - - # Simulate the intersection logic from _build_child_agent - parent_toolsets = set(parent.enabled_toolsets) - requested = ["terminal", "file", "web", "browser", "rl"] - scoped = [t for t in requested if t in parent_toolsets] - - assert sorted(scoped) == ["file", "terminal"] - assert "web" not in scoped - assert "browser" not in scoped - assert "rl" not in scoped - - def test_all_requested_toolsets_available_on_parent(self): - """LLM requests subset of parent tools — all pass through.""" - parent = SimpleNamespace(enabled_toolsets=["terminal", "file", "web", "browser"]) - - parent_toolsets = set(parent.enabled_toolsets) - requested = ["terminal", "web"] - scoped = [t for t in requested if t in parent_toolsets] - - assert sorted(scoped) == ["terminal", "web"] - - def test_no_toolsets_requested_inherits_parent(self): - """When toolsets is None/empty, child inherits parent's set.""" - parent_toolsets = ["terminal", "file", "web"] - child = _strip_blocked_tools(parent_toolsets) - assert "terminal" in child - assert "file" in child - assert "web" in child + """Subagent toolsets must stay within parent + profile scope.""" def test_strip_blocked_removes_delegation(self): - """Blocked toolsets (delegation, clarify, etc.) are always removed.""" child = _strip_blocked_tools(["terminal", "delegation", "clarify", "memory"]) assert "delegation" not in child assert "clarify" not in child assert "memory" not in child assert "terminal" in child - def test_empty_intersection_yields_empty_toolsets(self): - """If parent has no overlap with requested, child gets nothing extra.""" - parent = SimpleNamespace(enabled_toolsets=["terminal"]) - - parent_toolsets = set(parent.enabled_toolsets) - requested = ["web", "browser"] - scoped = [t for t in requested if t in parent_toolsets] - - assert scoped == [] + def test_requested_toolsets_are_scoped_by_profile_and_blocklist(self): + parent = SimpleNamespace(enabled_toolsets=["hermes-cli"]) + blocked = _build_child_blocked_tools("none") + child_toolsets, temp_toolset = _build_child_enabled_toolsets( + task_index=0, + requested_toolsets=["terminal", "web", "memory"], + parent_agent=parent, + profile_toolsets=["terminal", "web", "memory"], + blocked_tools=blocked, + ) + try: + assert child_toolsets == [temp_toolset] + resolved = set(resolve_toolset(temp_toolset)) + assert "terminal" in resolved + assert "web_search" in resolved + assert "memory" not in resolved + assert "delegate_task" not in resolved + finally: + TOOLSETS.pop(temp_toolset, None) + + def test_requested_toolset_outside_profile_is_dropped(self): + parent = SimpleNamespace(enabled_toolsets=["terminal", "file", "web"]) + blocked = _build_child_blocked_tools("none") + _, temp_toolset = _build_child_enabled_toolsets( + task_index=1, + requested_toolsets=["web"], + parent_agent=parent, + profile_toolsets=["file"], + blocked_tools=blocked, + ) + try: + assert resolve_toolset(temp_toolset) == [] + finally: + TOOLSETS.pop(temp_toolset, None) + + def test_legacy_default_toolsets_apply_without_profile(self): + profile = _resolve_delegation_profile({"default_toolsets": ["file", "web"]}, None) + assert profile["toolsets"] == ["file", "web"] + assert profile["memory"] == "none" + + def test_privileged_profile_allows_memory_writes(self): + profile = _resolve_delegation_profile({}, "privileged") + blocked = _build_child_blocked_tools(profile["memory"]) + assert profile["memory"] == "write" + assert "memory" not in blocked diff --git a/tests/tools/test_docker_environment.py b/tests/tools/test_docker_environment.py index 002776ca3441..76a6504eb155 100644 --- a/tests/tools/test_docker_environment.py +++ b/tests/tools/test_docker_environment.py @@ -1,4 +1,5 @@ import logging +import threading from io import StringIO import subprocess import sys @@ -219,7 +220,9 @@ def test_non_persistent_cleanup_removes_container(monkeypatch): # Should have stop and rm calls via Popen stop_cmds = [c for c in popen_cmds if container_id in str(c) and "stop" in str(c)] + rm_cmds = [c for c in popen_cmds if container_id in str(c) and "rm -f" in str(c)] assert len(stop_cmds) >= 1, f"cleanup() should schedule docker stop for {container_id}" + assert len(rm_cmds) >= 1, f"cleanup() should schedule docker rm -f for {container_id}" class _FakePopen: @@ -264,6 +267,100 @@ def _fake_popen(cmd, **kwargs): assert "GITHUB_TOKEN=value_from_dotenv" in popen_calls[0] +def test_start_persistent_exec_builds_interactive_docker_exec(monkeypatch): + env = _make_execute_only_env(["OPENAI_API_KEY"]) + popen_calls = [] + + class _FakePersistentPopen(_FakePopen): + def __init__(self, cmd, **kwargs): + super().__init__(cmd, **kwargs) + self.stdin = StringIO() + self.stdout = StringIO("") + self.stderr = StringIO("") + self.pid = 1234 + self.returncode = None + + def wait(self, timeout=None): + self.returncode = 0 + return 0 + + def terminate(self): + self.returncode = 0 + + def kill(self): + self.returncode = -9 + + def _fake_popen(cmd, **kwargs): + popen_calls.append((cmd, kwargs)) + return _FakePersistentPopen(cmd, **kwargs) + + monkeypatch.setenv("OPENAI_API_KEY", "env-value") + monkeypatch.setattr(docker_env.subprocess, "Popen", _fake_popen) + + session = env.start_persistent_exec( + cwd="/workspace", + command=["opencode", "acp"], + env={"HERMES_FOO": "bar"}, + ) + + assert session.pid == 1234 + cmd, kwargs = popen_calls[0] + assert cmd[:5] == ["/usr/bin/docker", "exec", "-i", "-w", "/workspace"] + assert "-e" in cmd + assert "OPENAI_API_KEY=env-value" in cmd + assert "HERMES_FOO=bar" in cmd + assert cmd[-3:] == ["test-container", "opencode", "acp"] + assert kwargs["stdin"] == subprocess.PIPE + assert kwargs["stdout"] == subprocess.PIPE + assert kwargs["stderr"] == subprocess.PIPE + + +def test_persistent_exec_session_streams_callbacks_and_exit(): + stdout_lines = [] + stderr_lines = [] + exit_codes = [] + + class _FakeProcess: + def __init__(self): + self.stdin = StringIO() + self.stdout = StringIO("hello\nworld\n") + self.stderr = StringIO("warn\n") + self.pid = 4321 + self.returncode = None + self.terminated = False + + def poll(self): + return self.returncode + + def wait(self, timeout=None): + self.returncode = 0 + return 0 + + def terminate(self): + self.terminated = True + self.returncode = 0 + + def kill(self): + self.returncode = -9 + + process = _FakeProcess() + session = docker_env.PersistentDockerExecSession(process) + done = threading.Event() + + session.read_loop( + stdout_handler=stdout_lines.append, + stderr_handler=stderr_lines.append, + exit_handler=lambda code: (exit_codes.append(code), done.set()), + ) + assert done.wait(timeout=1.0) + + session.write_line("{\"jsonrpc\":\"2.0\"}") + assert process.stdin.getvalue().endswith("\n") + assert stdout_lines == ["hello", "world"] + assert stderr_lines == ["warn"] + assert exit_codes == [0] + assert session.is_session_alive() is False + def test_execute_prefers_shell_env_over_hermes_dotenv(monkeypatch): env = _make_execute_only_env(["GITHUB_TOKEN"]) popen_calls = [] diff --git a/tests/tools/test_parse_env_var.py b/tests/tools/test_parse_env_var.py index cffee7c9af0f..b2c7ebfae6c9 100644 --- a/tests/tools/test_parse_env_var.py +++ b/tests/tools/test_parse_env_var.py @@ -52,6 +52,20 @@ def test_create_environment_passes_docker_forward_env(self): assert result is fake_env assert mock_docker.call_args.kwargs["forward_env"] == ["GITHUB_TOKEN"] + def test_task_override_can_switch_backend_with_sandbox_default_cwd(self): + task_id = "delegate-friendly" + with patch.dict("os.environ", {"TERMINAL_ENV": "local"}, clear=False): + _tt_mod.register_task_env_overrides(task_id, {"env_type": "docker"}) + try: + resolved = _tt_mod._resolve_task_environment_settings(task_id) + finally: + _tt_mod.clear_task_env_overrides(task_id) + + assert resolved["env_type"] == "docker" + assert resolved["cwd"] == "/root" + assert resolved["host_cwd"] is None + assert resolved["container_config"] is not None + def test_falls_back_to_default(self): with patch.dict("os.environ", {}, clear=False): # Remove the var if it exists, rely on default diff --git a/tools/background_subagent_tool.py b/tools/background_subagent_tool.py new file mode 100644 index 000000000000..8e01424233fa --- /dev/null +++ b/tools/background_subagent_tool.py @@ -0,0 +1,184 @@ +"""Tool schemas for persistent ACP background subagents.""" + +from __future__ import annotations + +import json + +from agent.background_subagents import get_background_subagent_manager +from tools.registry import registry + + +def check_background_subagent_requirements() -> bool: + return get_background_subagent_manager().check_requirements() + + +SPAWN_BACKGROUND_SUBAGENT_SCHEMA = { + "name": "spawn_background_subagent", + "description": ( + "Start a persistent background ACP subagent inside a sandbox container. " + "Use this when work should continue across turns and you want to poll or " + "message the subagent later instead of waiting for a delegate_task result." + ), + "parameters": { + "type": "object", + "properties": { + "purpose": { + "type": "string", + "description": "Why this background subagent exists and what role it should keep over time.", + }, + "initial_task": { + "type": "string", + "description": "The first task or instruction to send to the subagent after it starts.", + }, + "cwd": { + "type": "string", + "description": "Working directory inside the sandbox container for this subagent.", + }, + "agent_kind": { + "type": "string", + "description": "ACP-capable subagent kind to launch. Default comes from config.", + }, + }, + "required": ["purpose", "initial_task", "cwd"], + }, +} + +LIST_BACKGROUND_SUBAGENTS_SCHEMA = { + "name": "list_background_subagents", + "description": "List the current session's active background ACP subagents.", + "parameters": { + "type": "object", + "properties": {}, + }, +} + +SEND_BACKGROUND_SUBAGENT_SCHEMA = { + "name": "send_background_subagent", + "description": "Send a follow-up instruction to an existing background ACP subagent.", + "parameters": { + "type": "object", + "properties": { + "id": { + "type": "string", + "description": "Background subagent identifier.", + }, + "message": { + "type": "string", + "description": "Instruction or follow-up task to send through ACP.", + }, + }, + "required": ["id", "message"], + }, +} + +POLL_BACKGROUND_SUBAGENT_SCHEMA = { + "name": "poll_background_subagent", + "description": "Fetch unread buffered updates from a background ACP subagent.", + "parameters": { + "type": "object", + "properties": { + "id": { + "type": "string", + "description": "Background subagent identifier.", + }, + "since_seq": { + "type": "integer", + "description": "Optional event cursor; when omitted, returns unread events and marks them read.", + }, + }, + "required": ["id"], + }, +} + +GET_BACKGROUND_SUBAGENT_STATUS_SCHEMA = { + "name": "get_background_subagent_status", + "description": "Check liveness, deadlines, and queue state for a background ACP subagent.", + "parameters": { + "type": "object", + "properties": { + "id": { + "type": "string", + "description": "Background subagent identifier.", + }, + }, + "required": ["id"], + }, +} + +STOP_BACKGROUND_SUBAGENT_SCHEMA = { + "name": "stop_background_subagent", + "description": "Stop a background ACP subagent and tear down its sandbox container.", + "parameters": { + "type": "object", + "properties": { + "id": { + "type": "string", + "description": "Background subagent identifier.", + }, + "reason": { + "type": "string", + "description": "Optional short reason for stopping it.", + }, + }, + "required": ["id"], + }, +} + + +def _must_be_agent_loop(name: str) -> str: + return json.dumps({"error": f"{name} must be handled by the agent loop"}) + + +registry.register( + name="spawn_background_subagent", + toolset="delegation", + schema=SPAWN_BACKGROUND_SUBAGENT_SCHEMA, + handler=lambda args, **kw: _must_be_agent_loop("spawn_background_subagent"), + check_fn=check_background_subagent_requirements, + emoji="🧭", +) + +registry.register( + name="list_background_subagents", + toolset="delegation", + schema=LIST_BACKGROUND_SUBAGENTS_SCHEMA, + handler=lambda args, **kw: _must_be_agent_loop("list_background_subagents"), + check_fn=check_background_subagent_requirements, + emoji="🧾", +) + +registry.register( + name="send_background_subagent", + toolset="delegation", + schema=SEND_BACKGROUND_SUBAGENT_SCHEMA, + handler=lambda args, **kw: _must_be_agent_loop("send_background_subagent"), + check_fn=check_background_subagent_requirements, + emoji="📨", +) + +registry.register( + name="poll_background_subagent", + toolset="delegation", + schema=POLL_BACKGROUND_SUBAGENT_SCHEMA, + handler=lambda args, **kw: _must_be_agent_loop("poll_background_subagent"), + check_fn=check_background_subagent_requirements, + emoji="📡", +) + +registry.register( + name="get_background_subagent_status", + toolset="delegation", + schema=GET_BACKGROUND_SUBAGENT_STATUS_SCHEMA, + handler=lambda args, **kw: _must_be_agent_loop("get_background_subagent_status"), + check_fn=check_background_subagent_requirements, + emoji="🔎", +) + +registry.register( + name="stop_background_subagent", + toolset="delegation", + schema=STOP_BACKGROUND_SUBAGENT_SCHEMA, + handler=lambda args, **kw: _must_be_agent_loop("stop_background_subagent"), + check_fn=check_background_subagent_requirements, + emoji="🛑", +) diff --git a/tools/delegate_tool.py b/tools/delegate_tool.py index 7b75838001df..6486e4ed2907 100644 --- a/tools/delegate_tool.py +++ b/tools/delegate_tool.py @@ -39,6 +39,36 @@ DEFAULT_MAX_ITERATIONS = 50 DEFAULT_TOOLSETS = ["terminal", "file", "web"] +BUILTIN_DELEGATION_PROFILES = { + "restricted": { + "description": "Minimal subagent profile with no memory access.", + "toolsets": ["terminal", "file"], + "memory": "none", + "provider_tools": False, + "terminal": { + "backend": "docker", + }, + }, + "friendly": { + "description": "Read-only memory profile for cooperative code and research tasks.", + "toolsets": ["terminal", "file", "web"], + "memory": "read", + "provider_tools": False, + "terminal": { + "backend": "docker", + }, + }, + "privileged": { + "description": "Write-capable memory profile for trusted subagents.", + "toolsets": ["terminal", "file", "web", "memory"], + "memory": "write", + "provider_tools": True, + "terminal": { + "backend": "docker", + }, + }, +} + def check_delegate_requirements() -> bool: """Delegation has no external requirements -- always available.""" @@ -75,6 +105,203 @@ def _strip_blocked_tools(toolsets: List[str]) -> List[str]: return [t for t in toolsets if t not in blocked_toolset_names] +def _resolve_toolset_tools(toolsets: List[str]) -> set[str]: + """Resolve a list of toolset names into concrete tool names.""" + from toolsets import resolve_toolset, validate_toolset + + resolved: set[str] = set() + for toolset_name in toolsets or []: + if validate_toolset(toolset_name): + resolved.update(resolve_toolset(toolset_name)) + return resolved + + +def _normalize_memory_access(value: object) -> str: + """Normalize a delegation profile memory mode.""" + normalized = str(value or "none").strip().lower() + if normalized not in {"none", "read", "write"}: + raise ValueError( + f"Invalid delegation memory mode '{value}'. " + "Expected one of: none, read, write." + ) + return normalized + + +def _parse_boolish(value: object, default: bool = False) -> bool: + """Parse profile bool-like config values safely.""" + if value is None: + return default + if isinstance(value, bool): + return value + if isinstance(value, str): + lowered = value.strip().lower() + if lowered in {"true", "1", "yes", "on"}: + return True + if lowered in {"false", "0", "no", "off"}: + return False + return default + + +def _get_delegation_profiles(cfg: dict) -> dict: + """Return built-in profiles plus any user-defined overrides.""" + profiles = { + name: { + "description": spec.get("description", ""), + "toolsets": list(spec.get("toolsets") or []), + "memory": spec.get("memory", "none"), + "provider_tools": spec.get("provider_tools", False), + "terminal": dict(spec.get("terminal") or {}), + } + for name, spec in BUILTIN_DELEGATION_PROFILES.items() + } + + custom = cfg.get("profiles") or {} + if not isinstance(custom, dict): + return profiles + + for name, spec in custom.items(): + if not isinstance(spec, dict): + continue + existing = profiles.get(name, {}) + merged = { + "description": spec.get("description", existing.get("description", "")), + "toolsets": list(spec.get("toolsets") or existing.get("toolsets") or []), + "memory": spec.get("memory", existing.get("memory", "none")), + "provider_tools": spec.get("provider_tools", existing.get("provider_tools", False)), + "terminal": { + **dict(existing.get("terminal") or {}), + **dict(spec.get("terminal") or {}), + }, + } + profiles[name] = merged + + return profiles + + +def _resolve_delegation_profile(cfg: dict, requested_profile: Optional[str]) -> dict: + """Resolve a delegation capability profile.""" + profile_name = (requested_profile or cfg.get("default_profile") or "").strip() + if not profile_name: + legacy_toolsets = cfg.get("default_toolsets") + if not isinstance(legacy_toolsets, list): + legacy_toolsets = [] + return { + "name": "", + "toolsets": list(legacy_toolsets), + "memory": "none", + "provider_tools": False, + "terminal": {}, + } + + profiles = _get_delegation_profiles(cfg) + spec = profiles.get(profile_name) + if not isinstance(spec, dict): + raise ValueError( + f"Unknown delegation profile '{profile_name}'. " + f"Available: {', '.join(sorted(profiles))}" + ) + + return { + "name": profile_name, + "toolsets": list(spec.get("toolsets") or []), + "memory": _normalize_memory_access(spec.get("memory")), + "provider_tools": _parse_boolish(spec.get("provider_tools"), default=False), + "terminal": dict(spec.get("terminal") or {}), + } + + +def _build_child_blocked_tools(memory_access: str) -> set[str]: + """Return the blocked tool names for a child profile.""" + blocked = set(DELEGATE_BLOCKED_TOOLS) + if memory_access == "write": + blocked.discard("memory") + return blocked + + +def _build_child_enabled_toolsets( + task_index: int, + requested_toolsets: Optional[List[str]], + parent_agent, + profile_toolsets: Optional[List[str]], + blocked_tools: set[str], +) -> tuple[List[str], str]: + """Create an exact child toolset derived from parent + profile constraints.""" + from toolsets import create_custom_toolset, resolve_toolset, validate_toolset + + parent_toolsets = list(getattr(parent_agent, "enabled_toolsets", None) or DEFAULT_TOOLSETS) + parent_allowed_tools = _resolve_toolset_tools(parent_toolsets) - blocked_tools + + profile_allowed_tools = set(parent_allowed_tools) + if profile_toolsets: + profile_allowed_tools = set() + for toolset_name in profile_toolsets: + if not validate_toolset(toolset_name): + logger.debug("Ignoring unknown toolset in delegation profile: %s", toolset_name) + continue + profile_allowed_tools.update(resolve_toolset(toolset_name)) + profile_allowed_tools &= parent_allowed_tools + + child_tools = set(profile_allowed_tools) + if requested_toolsets: + child_tools = set() + for toolset_name in requested_toolsets: + if not validate_toolset(toolset_name): + logger.debug("Dropping unknown delegated toolset request: %s", toolset_name) + continue + + requested_tools = set(resolve_toolset(toolset_name)) - blocked_tools + if not requested_tools: + logger.debug( + "Dropping delegated toolset '%s': contains no child-safe tools", + toolset_name, + ) + continue + + if not requested_tools.issubset(profile_allowed_tools): + logger.debug( + "Dropping delegated toolset '%s': requests tools outside parent/profile scope", + toolset_name, + ) + continue + + child_tools.update(requested_tools) + + toolset_name = f"_delegate_child_{os.getpid()}_{task_index}_{time.time_ns()}" + create_custom_toolset( + name=toolset_name, + description="Exact tool subset for delegated child agent", + tools=sorted(child_tools), + ) + return [toolset_name], toolset_name + + +def _build_terminal_overrides(profile: dict) -> Dict[str, Any]: + """Translate a delegation profile terminal block into per-task overrides.""" + terminal = profile.get("terminal") or {} + if not isinstance(terminal, dict): + return {} + + overrides: Dict[str, Any] = {} + backend = str(terminal.get("backend") or "").strip() + if backend: + overrides["env_type"] = backend + + for key in ( + "cwd", + "docker_image", + "modal_image", + "daytona_image", + "singularity_image", + "docker_mount_cwd_to_workspace", + "docker_forward_env", + "docker_volumes", + ): + if key in terminal: + overrides[key] = terminal[key] + + return overrides + + def _build_child_progress_callback(task_index: int, parent_agent, task_count: int = 1) -> Optional[callable]: """Build a callback that relays child agent tool calls to the parent display. @@ -152,6 +379,7 @@ def _build_child_agent( goal: str, context: Optional[str], toolsets: Optional[List[str]], + profile: dict, model: Optional[str], max_iterations: int, parent_agent, @@ -172,16 +400,14 @@ def _build_child_agent( """ from run_agent import AIAgent - # When no explicit toolsets given, inherit from parent's enabled toolsets - # so disabled tools (e.g. web) don't leak to subagents. - parent_toolsets = set(getattr(parent_agent, "enabled_toolsets", None) or DEFAULT_TOOLSETS) - if toolsets: - # Intersect with parent — subagent must not gain tools the parent lacks - child_toolsets = _strip_blocked_tools([t for t in toolsets if t in parent_toolsets]) - elif parent_agent and getattr(parent_agent, "enabled_toolsets", None): - child_toolsets = _strip_blocked_tools(parent_agent.enabled_toolsets) - else: - child_toolsets = _strip_blocked_tools(DEFAULT_TOOLSETS) + blocked_tools = _build_child_blocked_tools(profile.get("memory", "none")) + child_toolsets, delegate_toolset_name = _build_child_enabled_toolsets( + task_index=task_index, + requested_toolsets=toolsets, + parent_agent=parent_agent, + profile_toolsets=profile.get("toolsets"), + blocked_tools=blocked_tools, + ) child_prompt = _build_child_system_prompt(goal, context) # Extract parent's API key so subagents inherit auth (e.g. Nous Portal). @@ -224,7 +450,10 @@ def _build_child_agent( log_prefix=f"[subagent-{task_index}]", platform=parent_agent.platform, skip_context_files=True, - skip_memory=True, + skip_memory=profile.get("memory", "none") == "none", + memory_write_enabled=profile.get("memory", "none") == "write", + provider_tool_access=bool(profile.get("provider_tools", False)), + agent_context="subagent", clarify_callback=None, session_db=getattr(parent_agent, '_session_db', None), providers_allowed=parent_agent.providers_allowed, @@ -236,6 +465,9 @@ def _build_child_agent( ) # Set delegation depth so children can't spawn grandchildren child._delegate_depth = getattr(parent_agent, '_delegate_depth', 0) + 1 + child._delegate_temp_toolset = delegate_toolset_name + child._delegate_terminal_overrides = _build_terminal_overrides(profile) + child._delegate_task_id = f"delegate-{task_index}-{time.time_ns()}" # Register child for interrupt propagation if hasattr(parent_agent, '_active_children'): @@ -269,9 +501,15 @@ def _run_single_child( import model_tools _saved_tool_names = getattr(child, "_delegate_saved_tool_names", list(model_tools._last_resolved_tool_names)) + child_task_id = getattr(child, "_delegate_task_id", f"delegate-{task_index}-{time.time_ns()}") + terminal_overrides = getattr(child, "_delegate_terminal_overrides", None) or {} try: - result = child.run_conversation(user_message=goal) + if terminal_overrides: + from tools.terminal_tool import register_task_env_overrides + register_task_env_overrides(child_task_id, terminal_overrides) + + result = child.run_conversation(user_message=goal, task_id=child_task_id) # Flush any remaining batched progress to gateway if child_progress_cb and hasattr(child_progress_cb, '_flush'): @@ -383,11 +621,19 @@ def _run_single_child( # Restore the parent's tool names so the process-global is correct # for any subsequent execute_code calls or other consumers. import model_tools + from toolsets import TOOLSETS + from tools.terminal_tool import clear_task_env_overrides saved_tool_names = getattr(child, "_delegate_saved_tool_names", None) if isinstance(saved_tool_names, list): model_tools._last_resolved_tool_names = list(saved_tool_names) + temp_toolset = getattr(child, "_delegate_temp_toolset", None) + if temp_toolset: + TOOLSETS.pop(temp_toolset, None) + + clear_task_env_overrides(child_task_id) + # Unregister child from interrupt propagation if hasattr(parent_agent, '_active_children'): try: @@ -404,6 +650,8 @@ def delegate_task( goal: Optional[str] = None, context: Optional[str] = None, toolsets: Optional[List[str]] = None, + profile: Optional[str] = None, + async_mode: bool = False, tasks: Optional[List[Dict[str, Any]]] = None, max_iterations: Optional[int] = None, parent_agent=None, @@ -445,11 +693,36 @@ def delegate_task( except ValueError as exc: return json.dumps({"error": str(exc)}) + if async_mode: + if tasks and isinstance(tasks, list): + return json.dumps({ + "error": "delegate_task(async=true) currently supports single-task mode only.", + }) + if not goal or not isinstance(goal, str) or not goal.strip(): + return json.dumps({"error": "delegate_task(async=true) requires a non-empty goal."}) + from agent.async_delegate_tasks import get_async_delegate_manager + + owner_session_id = str(getattr(parent_agent, "session_id", "") or "").strip() + if not owner_session_id: + return json.dumps({"error": "delegate_task(async=true) requires a parent session_id."}) + + result = get_async_delegate_manager().spawn( + owner_session_id=owner_session_id, + parent_agent=parent_agent, + goal=goal, + context=context or "", + toolsets=list(toolsets or []), + profile=profile, + max_iterations=effective_max_iter, + creds=creds, + ) + return json.dumps(result, ensure_ascii=False) + # Normalize to task list if tasks and isinstance(tasks, list): task_list = tasks[:MAX_CONCURRENT_CHILDREN] elif goal and isinstance(goal, str) and goal.strip(): - task_list = [{"goal": goal, "context": context, "toolsets": toolsets}] + task_list = [{"goal": goal, "context": context, "toolsets": toolsets, "profile": profile}] else: return json.dumps({"error": "Provide either 'goal' (single task) or 'tasks' (batch)."}) @@ -480,8 +753,13 @@ def delegate_task( children = [] try: for i, t in enumerate(task_list): + try: + resolved_profile = _resolve_delegation_profile(cfg, t.get("profile") or profile) + except ValueError as exc: + return json.dumps({"error": str(exc)}) child = _build_child_agent( task_index=i, goal=t["goal"], context=t.get("context"), + profile=resolved_profile, toolsets=t.get("toolsets") or toolsets, model=creds["model"], max_iterations=effective_max_iter, parent_agent=parent_agent, override_provider=creds["provider"], override_base_url=creds["base_url"], @@ -706,7 +984,11 @@ def _load_config() -> dict: "TWO MODES (one of 'goal' or 'tasks' is required):\n" "1. Single task: provide 'goal' (+ optional context, toolsets)\n" "2. Batch (parallel): provide 'tasks' array with up to 3 items. " - "All run concurrently and results are returned together.\n\n" + "All run concurrently and results are returned together.\n" + "3. Async background: provide 'goal' with 'async'=true to launch a " + "single delegated subagent that keeps running after this tool call returns. " + "Its progress/summary are written to a workspace temp file and its status " + "appears in later turn context.\n\n" "WHEN TO USE delegate_task:\n" "- Reasoning-heavy subtasks (debugging, code review, research synthesis)\n" "- Tasks that would flood your context with intermediate data\n" @@ -716,10 +998,12 @@ def _load_config() -> dict: "- Single tool call -> just call the tool directly\n" "- Tasks needing user interaction -> subagents cannot use clarify\n\n" "IMPORTANT:\n" - "- Subagents have NO memory of your conversation. Pass all relevant " + "- Subagents do not automatically know your conversation history. Pass all relevant " "info (file paths, error messages, constraints) via the 'context' field.\n" - "- Subagents CANNOT call: delegate_task, clarify, memory, send_message, " - "execute_code.\n" + "- Profiles can change child capabilities, including memory access and " + "terminal backend selection.\n" + "- Subagents CANNOT call: delegate_task, clarify, send_message, " + "execute_code. Memory access depends on the selected profile.\n" "- Each subagent gets its own terminal session (separate working directory and state).\n" "- Results are always returned as an array, one entry per task." ), @@ -747,12 +1031,20 @@ def _load_config() -> dict: "items": {"type": "string"}, "description": ( "Toolsets to enable for this subagent. " - "Default: inherits your enabled toolsets. " + "Default: inherits the selected profile's child-safe tools. " "Common patterns: ['terminal', 'file'] for code work, " "['web'] for research, ['terminal', 'file', 'web'] for " "full-stack tasks." ), }, + "profile": { + "type": "string", + "description": ( + "Optional delegation capability profile. Built-ins: " + "'restricted', 'friendly', 'privileged'. Profiles can set " + "default toolsets, memory access, and terminal backend overrides." + ), + }, "tasks": { "type": "array", "items": { @@ -760,6 +1052,10 @@ def _load_config() -> dict: "properties": { "goal": {"type": "string", "description": "Task goal"}, "context": {"type": "string", "description": "Task-specific context"}, + "profile": { + "type": "string", + "description": "Delegation profile for this specific task", + }, "toolsets": { "type": "array", "items": {"type": "string"}, @@ -782,6 +1078,14 @@ def _load_config() -> dict: "Only set lower for simple tasks." ), }, + "async": { + "type": "boolean", + "description": ( + "When true, launch a single delegated subagent in the background " + "and return immediately with its tracking info and workspace output file. " + "Async mode currently supports only the single-task goal/context form." + ), + }, }, "required": [], }, @@ -799,6 +1103,8 @@ def _load_config() -> dict: goal=args.get("goal"), context=args.get("context"), toolsets=args.get("toolsets"), + profile=args.get("profile"), + async_mode=bool(args.get("async", False)), tasks=args.get("tasks"), max_iterations=args.get("max_iterations"), parent_agent=kw.get("parent_agent")), diff --git a/tools/environments/docker.py b/tools/environments/docker.py index 19889ea35ffe..189afca3809d 100644 --- a/tools/environments/docker.py +++ b/tools/environments/docker.py @@ -120,6 +120,85 @@ def find_docker() -> Optional[str]: _storage_opt_ok: Optional[bool] = None # cached result across instances +class PersistentDockerExecSession: + """Long-lived interactive `docker exec -i` session.""" + + def __init__(self, process: subprocess.Popen[str]): + self._process = process + self._reader_started = False + self._reader_lock = threading.Lock() + self._write_lock = threading.Lock() + + @property + def pid(self) -> int | None: + return getattr(self._process, "pid", None) + + def write_line(self, payload: str) -> None: + if self._process.stdin is None: + raise RuntimeError("Persistent docker exec stdin is not available.") + with self._write_lock: + self._process.stdin.write(payload + "\n") + self._process.stdin.flush() + + def read_loop( + self, + *, + stdout_handler=None, + stderr_handler=None, + exit_handler=None, + ) -> None: + with self._reader_lock: + if self._reader_started: + return + self._reader_started = True + + def _drain_stdout(): + if self._process.stdout is None: + return + try: + for line in self._process.stdout: + if stdout_handler: + stdout_handler(line.rstrip("\n")) + except Exception: + logger.debug("Persistent docker exec stdout reader crashed", exc_info=True) + + def _drain_stderr(): + if self._process.stderr is None: + return + try: + for line in self._process.stderr: + if stderr_handler: + stderr_handler(line.rstrip("\n")) + except Exception: + logger.debug("Persistent docker exec stderr reader crashed", exc_info=True) + + def _wait_for_exit(): + try: + self._process.wait() + finally: + if exit_handler: + exit_handler(self._process.returncode) + + threading.Thread(target=_drain_stdout, daemon=True).start() + threading.Thread(target=_drain_stderr, daemon=True).start() + threading.Thread(target=_wait_for_exit, daemon=True).start() + + def is_session_alive(self) -> bool: + return self._process.poll() is None + + def terminate_session(self) -> None: + if self._process.poll() is not None: + return + try: + self._process.terminate() + self._process.wait(timeout=2) + except Exception: + try: + self._process.kill() + except Exception: + pass + + def _ensure_docker_available() -> None: """Best-effort check that the docker CLI is available before use. @@ -524,6 +603,49 @@ def _drain(): except Exception as e: return {"output": f"Docker execution error: {e}", "returncode": 1} + def start_persistent_exec( + self, + *, + cwd: str = "", + command: list[str] | None = None, + env: dict[str, str] | None = None, + ) -> PersistentDockerExecSession: + """Start a long-lived interactive docker exec session.""" + assert self._container_id, "Container not started" + work_dir = cwd or self.cwd + + cmd = [self._docker_exe, "exec", "-i", "-w", work_dir] + forward_keys = set(self._forward_env) + try: + from tools.env_passthrough import get_all_passthrough + + forward_keys |= get_all_passthrough() + except Exception: + pass + hermes_env = _load_hermes_env_vars() if forward_keys else {} + for key in sorted(forward_keys): + value = os.getenv(key) + if value is None: + value = hermes_env.get(key) + if value is not None: + cmd.extend(["-e", f"{key}={value}"]) + for key, value in sorted((env or {}).items()): + if value is not None: + cmd.extend(["-e", f"{key}={value}"]) + + cmd.append(self._container_id) + cmd.extend(list(command or ["bash"])) + + process = subprocess.Popen( + cmd, + stdin=subprocess.PIPE, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + bufsize=1, + ) + return PersistentDockerExecSession(process) + def cleanup(self): """Stop and remove the container. Bind-mount dirs persist if persistent=True.""" if self._container_id: diff --git a/tools/file_tools.py b/tools/file_tools.py index 79a111cb7961..45dd48b130fb 100644 --- a/tools/file_tools.py +++ b/tools/file_tools.py @@ -158,7 +158,8 @@ def _get_file_ops(task_id: str = "default") -> ShellFileOperations: """ from tools.terminal_tool import ( _active_environments, _env_lock, _create_environment, - _get_env_config, _last_activity, _start_cleanup_thread, + _get_env_config, _last_activity, _resolve_task_environment_settings, + _start_cleanup_thread, _creation_locks, _creation_locks_lock, ) @@ -195,62 +196,23 @@ def _get_file_ops(task_id: str = "default") -> ShellFileOperations: terminal_env = None if terminal_env is None: - from tools.terminal_tool import _task_env_overrides - config = _get_env_config() - env_type = config["env_type"] - overrides = _task_env_overrides.get(task_id, {}) - - if env_type == "docker": - image = overrides.get("docker_image") or config["docker_image"] - elif env_type == "singularity": - image = overrides.get("singularity_image") or config["singularity_image"] - elif env_type == "modal": - image = overrides.get("modal_image") or config["modal_image"] - elif env_type == "daytona": - image = overrides.get("daytona_image") or config["daytona_image"] - else: - image = "" - - cwd = overrides.get("cwd") or config["cwd"] + resolved_env = _resolve_task_environment_settings(task_id, config) + env_type = resolved_env["env_type"] + image = resolved_env["image"] + cwd = resolved_env["cwd"] logger.info("Creating new %s environment for task %s...", env_type, task_id[:8]) - container_config = None - if env_type in ("docker", "singularity", "modal", "daytona"): - container_config = { - "container_cpu": config.get("container_cpu", 1), - "container_memory": config.get("container_memory", 5120), - "container_disk": config.get("container_disk", 51200), - "container_persistent": config.get("container_persistent", True), - "docker_volumes": config.get("docker_volumes", []), - } - - ssh_config = None - if env_type == "ssh": - ssh_config = { - "host": config.get("ssh_host", ""), - "user": config.get("ssh_user", ""), - "port": config.get("ssh_port", 22), - "key": config.get("ssh_key", ""), - "persistent": config.get("ssh_persistent", False), - } - - local_config = None - if env_type == "local": - local_config = { - "persistent": config.get("local_persistent", False), - } - terminal_env = _create_environment( env_type=env_type, image=image, cwd=cwd, - timeout=config["timeout"], - ssh_config=ssh_config, - container_config=container_config, - local_config=local_config, + timeout=resolved_env["timeout"], + ssh_config=resolved_env["ssh_config"], + container_config=resolved_env["container_config"], + local_config=resolved_env["local_config"], task_id=task_id, - host_cwd=config.get("host_cwd"), + host_cwd=resolved_env.get("host_cwd"), ) with _env_lock: diff --git a/tools/terminal_tool.py b/tools/terminal_tool.py index e11f9d43454b..bcbffdebd938 100644 --- a/tools/terminal_tool.py +++ b/tools/terminal_tool.py @@ -422,6 +422,15 @@ def replace_sudo(match): _task_env_overrides: Dict[str, Dict[str, Any]] = {} +def _default_cwd_for_env_type(env_type: str) -> str: + """Return the backend-appropriate default working directory.""" + if env_type == "local": + return os.getcwd() + if env_type == "ssh": + return "~" + return "/root" + + def register_task_env_overrides(task_id: str, overrides: Dict[str, Any]): """ Register environment overrides for a specific task/rollout. @@ -430,9 +439,15 @@ def register_task_env_overrides(task_id: str, overrides: Dict[str, Any]): per-task sandbox settings (e.g., a custom Dockerfile for the Modal image). Supported override keys: + - env_type: str -- Backend override ("docker", "modal", etc.) - modal_image: str -- Path to Dockerfile or Docker Hub image name - docker_image: str -- Docker image name + - singularity_image: str -- Singularity image reference + - daytona_image: str -- Daytona image name - cwd: str -- Working directory inside the sandbox + - docker_forward_env: list[str] -- Env vars to forward into Docker + - docker_volumes: list[str] -- Additional Docker bind mounts + - docker_mount_cwd_to_workspace: bool -- Opt-in host cwd bind mount Args: task_id: The rollout's unique task identifier @@ -475,15 +490,7 @@ def _get_env_config() -> Dict[str, Any]: mount_docker_cwd = os.getenv("TERMINAL_DOCKER_MOUNT_CWD_TO_WORKSPACE", "false").lower() in ("true", "1", "yes") - # Default cwd: local uses the host's current directory, everything - # else starts in the user's home (~ resolves to whatever account - # is running inside the container/remote). - if env_type == "local": - default_cwd = os.getcwd() - elif env_type == "ssh": - default_cwd = "~" - else: - default_cwd = "/root" + default_cwd = _default_cwd_for_env_type(env_type) # Read TERMINAL_CWD but sanity-check it for container backends. # If Docker cwd passthrough is explicitly enabled, remap the host path to @@ -546,6 +553,80 @@ def _get_env_config() -> Dict[str, Any]: } +def _resolve_task_environment_settings(task_id: str, config: Dict[str, Any] | None = None) -> Dict[str, Any]: + """Resolve per-task environment settings after applying registered overrides.""" + config = dict(config or _get_env_config()) + overrides = dict(_task_env_overrides.get(task_id, {}) or {}) + + env_type = str(overrides.get("env_type") or config["env_type"]) + base_env_type = str(config["env_type"]) + + if env_type == "docker": + image = overrides.get("docker_image") or config["docker_image"] + elif env_type == "singularity": + image = overrides.get("singularity_image") or config["singularity_image"] + elif env_type == "modal": + image = overrides.get("modal_image") or config["modal_image"] + elif env_type == "daytona": + image = overrides.get("daytona_image") or config["daytona_image"] + else: + image = "" + + cwd = overrides.get("cwd") + if not cwd: + cwd = config["cwd"] if env_type == base_env_type else _default_cwd_for_env_type(env_type) + + host_cwd = overrides.get("host_cwd") + if host_cwd is None: + host_cwd = config.get("host_cwd") if env_type == base_env_type else None + + container_config = None + if env_type in ("docker", "singularity", "modal", "daytona"): + container_config = { + "container_cpu": config.get("container_cpu", 1), + "container_memory": config.get("container_memory", 5120), + "container_disk": config.get("container_disk", 51200), + "container_persistent": config.get("container_persistent", True), + "modal_mode": config.get("modal_mode", "auto"), + "docker_volumes": overrides.get("docker_volumes", config.get("docker_volumes", [])), + "docker_mount_cwd_to_workspace": overrides.get( + "docker_mount_cwd_to_workspace", + config.get("docker_mount_cwd_to_workspace", False), + ), + "docker_forward_env": overrides.get( + "docker_forward_env", + config.get("docker_forward_env", []), + ), + } + + ssh_config = None + if env_type == "ssh": + ssh_config = { + "host": config.get("ssh_host", ""), + "user": config.get("ssh_user", ""), + "port": config.get("ssh_port", 22), + "key": config.get("ssh_key", ""), + "persistent": config.get("ssh_persistent", False), + } + + local_config = None + if env_type == "local": + local_config = { + "persistent": config.get("local_persistent", False), + } + + return { + "env_type": env_type, + "image": image, + "cwd": cwd, + "host_cwd": host_cwd, + "container_config": container_config, + "ssh_config": ssh_config, + "local_config": local_config, + "timeout": config["timeout"], + } + + def _get_modal_backend_state(modal_mode: object | None) -> Dict[str, Any]: """Resolve direct vs managed Modal backend selection.""" return resolve_modal_backend_state( @@ -941,29 +1022,13 @@ def terminal_tool( try: # Get configuration config = _get_env_config() - env_type = config["env_type"] - # Use task_id for environment isolation effective_task_id = task_id or "default" - - # Check per-task overrides (set by environments like TerminalBench2Env) - # before falling back to global env var config - overrides = _task_env_overrides.get(effective_task_id, {}) - - # Select image based on env type, with per-task override support - if env_type == "docker": - image = overrides.get("docker_image") or config["docker_image"] - elif env_type == "singularity": - image = overrides.get("singularity_image") or config["singularity_image"] - elif env_type == "modal": - image = overrides.get("modal_image") or config["modal_image"] - elif env_type == "daytona": - image = overrides.get("daytona_image") or config["daytona_image"] - else: - image = "" - - cwd = overrides.get("cwd") or config["cwd"] - default_timeout = config["timeout"] + resolved_env = _resolve_task_environment_settings(effective_task_id, config) + env_type = resolved_env["env_type"] + image = resolved_env["image"] + cwd = resolved_env["cwd"] + default_timeout = resolved_env["timeout"] effective_timeout = timeout or default_timeout # Start cleanup thread @@ -1002,32 +1067,9 @@ def terminal_tool( logger.info("Creating new %s environment for task %s...", env_type, effective_task_id[:8]) try: ssh_config = None - if env_type == "ssh": - ssh_config = { - "host": config.get("ssh_host", ""), - "user": config.get("ssh_user", ""), - "port": config.get("ssh_port", 22), - "key": config.get("ssh_key", ""), - "persistent": config.get("ssh_persistent", False), - } - - container_config = None - if env_type in ("docker", "singularity", "modal", "daytona"): - container_config = { - "container_cpu": config.get("container_cpu", 1), - "container_memory": config.get("container_memory", 5120), - "container_disk": config.get("container_disk", 51200), - "container_persistent": config.get("container_persistent", True), - "modal_mode": config.get("modal_mode", "auto"), - "docker_volumes": config.get("docker_volumes", []), - "docker_mount_cwd_to_workspace": config.get("docker_mount_cwd_to_workspace", False), - } - - local_config = None - if env_type == "local": - local_config = { - "persistent": config.get("local_persistent", False), - } + ssh_config = resolved_env["ssh_config"] + container_config = resolved_env["container_config"] + local_config = resolved_env["local_config"] new_env = _create_environment( env_type=env_type, @@ -1038,7 +1080,7 @@ def terminal_tool( container_config=container_config, local_config=local_config, task_id=effective_task_id, - host_cwd=config.get("host_cwd"), + host_cwd=resolved_env.get("host_cwd"), ) except ImportError as e: return json.dumps({ diff --git a/toolsets.py b/toolsets.py index 84c19637f940..b59031352e87 100644 --- a/toolsets.py +++ b/toolsets.py @@ -56,6 +56,9 @@ "clarify", # Code execution + delegation "execute_code", "delegate_task", + "spawn_background_subagent", "list_background_subagents", + "send_background_subagent", "poll_background_subagent", + "get_background_subagent_status", "stop_background_subagent", # Cronjob management "cronjob", # Cross-platform messaging (gated on gateway running via check_fn) @@ -190,7 +193,15 @@ "delegation": { "description": "Spawn subagents with isolated context for complex subtasks", - "tools": ["delegate_task"], + "tools": [ + "delegate_task", + "spawn_background_subagent", + "list_background_subagents", + "send_background_subagent", + "poll_background_subagent", + "get_background_subagent_status", + "stop_background_subagent", + ], "includes": [] }, @@ -240,6 +251,9 @@ "todo", "memory", "session_search", "execute_code", "delegate_task", + "spawn_background_subagent", "list_background_subagents", + "send_background_subagent", "poll_background_subagent", + "get_background_subagent_status", "stop_background_subagent", ], "includes": [] }, @@ -270,6 +284,9 @@ "session_search", # Code execution + delegation "execute_code", "delegate_task", + "spawn_background_subagent", "list_background_subagents", + "send_background_subagent", "poll_background_subagent", + "get_background_subagent_status", "stop_background_subagent", # Cronjob management "cronjob", # Home Assistant smart home control (gated on HASS_TOKEN via check_fn) diff --git a/website/docs/user-guide/features/delegation.md b/website/docs/user-guide/features/delegation.md index 80a5ad6230df..399f412d4796 100644 --- a/website/docs/user-guide/features/delegation.md +++ b/website/docs/user-guide/features/delegation.md @@ -206,10 +206,20 @@ Delegation has a **depth limit of 2** — a parent (depth 0) can spawn children # In ~/.hermes/config.yaml delegation: max_iterations: 50 # Max turns per child (default: 50) - default_toolsets: ["terminal", "file", "web"] # Default toolsets + default_toolsets: ["terminal", "file", "web"] # Legacy fallback when no profile is selected + default_profile: "friendly" # Built-in: restricted, friendly, privileged model: "google/gemini-3-flash-preview" # Optional provider/model override provider: "openrouter" # Optional built-in provider + # Optional custom child capability profiles + profiles: + docs: + toolsets: ["file", "web"] + memory: "read" + provider_tools: false + terminal: + backend: "docker" + # Or use a direct custom endpoint instead of provider: delegation: model: "qwen2.5-coder"