"""Runtime-managed native subagents.""" from __future__ import annotations import asyncio import inspect import time import uuid from dataclasses import dataclass, field from typing import Any, Callable, Awaitable from opc.core.config import OPCConfig, NativeSubagentProfileConfig from opc.core.models import OPCEvent, Task, TaskResult, TaskStatus from opc.layer2_organization.work_item_identity import projection_id_for_task, work_item_identity_payload_for_task from opc.layer3_agent.runtime_v2.worktree import cleanup_worktree, create_worktree ChildAgentFactory = Callable[..., Any] @dataclass class SubagentState: agent_id: str profile: str name: str = "" description: str = "" model: str = "" mode: str = "default" isolation: str = "shared" max_iterations: int = 24 status: str = "running" background: bool = False resident: bool = False fork_mode: bool = False created_at: float = field(default_factory=time.time) latest_result: str = "" pending_messages_count: int = 0 last_notification_kind: str = "" worktree: dict[str, Any] | None = None fork_system_prompt: str = "" fork_context_messages: list[dict[str, Any]] = field(default_factory=list) fork_allowed_tools: list[str] = field(default_factory=list) runtime_task: asyncio.Task[Any] | None = None inbox: asyncio.Queue[dict[str, Any]] = field(default_factory=asyncio.Queue) completion: asyncio.Event = field(default_factory=asyncio.Event) update_event: asyncio.Event = field(default_factory=asyncio.Event) task_result: TaskResult | None = None class SubagentManager: def __init__( self, *, parent_task: Task | None, config: OPCConfig | None, child_agent_factory: ChildAgentFactory | None, event_bus: Any = None, store: Any = None, runtime_session_id: str = "", ) -> None: self.parent_task = parent_task self.config = config or OPCConfig() self.child_agent_factory = child_agent_factory self.event_bus = event_bus self.store = store self.runtime_session_id = runtime_session_id self.states: dict[str, SubagentState] = {} self.agent_names: dict[str, str] = {} async def spawn( self, *, profile: str, prompt: str, background: bool | None = None, isolation: str | None = None, description: str = "", name: str = "", model: str = "", mode: str = "default", fork_context_messages: list[dict[str, Any]] | None = None, fork_system_prompt: str = "", fork_allowed_tools: list[str] | None = None, fork_mode: bool = False, resident: bool = False, ) -> dict[str, Any]: if self.child_agent_factory is None: return {"error": "Native subagent factory is not configured", "success": False} parent_depth = int(getattr(self.parent_task, "metadata", {}).get("_native_runtime_depth", 0) or 0) max_depth = int(self.config.system.native_runtime.subagent_max_depth or 3) if parent_depth >= max_depth: return { "error": f"Maximum native subagent depth ({max_depth}) reached", "success": False, } profile_cfg = self._profile_config(profile) effective_mode = str(mode or "default").strip() or "default" default_isolation = profile_cfg.default_isolation if not default_isolation: default_isolation = "shared" if profile in {"explore", "plan"} else "worktree" effective_isolation = str(isolation or default_isolation or "shared").strip().lower() or "shared" if effective_mode == "plan": effective_isolation = "shared" if effective_isolation not in {"shared", "worktree"}: effective_isolation = str(profile_cfg.default_isolation or "shared").strip().lower() or "shared" effective_background = profile_cfg.background if background is None else bool(background) effective_resident = bool(resident) and bool(effective_background) effective_model = str(model or profile_cfg.model or "").strip() effective_description = str(description or prompt or "").strip() effective_name = str(name or "").strip() if effective_name and effective_name in self.agent_names: existing_state = self.states.get(self.agent_names[effective_name]) if existing_state is not None and not existing_state.completion.is_set(): return {"error": f"Subagent name `{effective_name}` is already in use", "success": False} agent_id = f"na_{uuid.uuid4().hex[:10]}" worktree = None if effective_isolation == "worktree": base_path = str(getattr(self.parent_task, "metadata", {}).get("target_output_dir", "") or "").strip() worktree = await create_worktree(base_path or None, config=self.config) state = SubagentState( agent_id=agent_id, profile=profile, name=effective_name, description=effective_description, model=effective_model, mode=effective_mode, isolation=effective_isolation, max_iterations=max(1, int(profile_cfg.max_iterations or 24)), background=effective_background, resident=effective_resident, fork_mode=bool(fork_mode), worktree=worktree, fork_system_prompt=str(fork_system_prompt or ""), fork_context_messages=list(fork_context_messages or []), fork_allowed_tools=list(fork_allowed_tools or []), ) self.states[agent_id] = state if effective_name: self.agent_names[effective_name] = agent_id self._ensure_comms_endpoint(state) await self._save_state(state, "running") if state.worktree and self.store and hasattr(self.store, "save_runtime_worktree_session"): await self.store.save_runtime_worktree_session( worktree_session_id=f"wt_{agent_id}", runtime_session_id=self.runtime_session_id, task_id=self.parent_task.id if self.parent_task else None, path=str(state.worktree.get("path", "") or ""), status="active", metadata=dict(state.worktree or {}), ) await self._emit( "subagent_started", state, { "prompt": prompt, "description": effective_description, "name": effective_name, "isolation": effective_isolation, "mode": effective_mode, "model": effective_model, "fork_mode": state.fork_mode, "resident": state.resident, }, ) async def _execute_turn(turn_prompt: str) -> None: try: state.status = "running" state.update_event.set() await self._save_state(state, state.status) child = self._build_child_task(state, turn_prompt) child.metadata["_permission_bridge_runtime_session_id"] = self.runtime_session_id setattr(child, "_runtime_permission_bridge", self._build_permission_bridge(state)) child_agent = self._build_child_agent( profile=profile, allowed_tools=list(state.fork_allowed_tools) or self._resolve_allowed_tools(profile, mode=effective_mode), prompt_addendum=self._profile_prompt(profile, mode=effective_mode), state=state, ) setattr(child, "_runtime_inbox_queue", state.inbox) result = await child_agent.execute(child) state.task_result = result state.latest_result = result.content terminal_status = result.status.value state.last_notification_kind = self._resident_notification_kind(result) state.status = "idle" if state.resident else terminal_status await self._save_state(state, state.status, { "turn_status": terminal_status, "notification_kind": state.last_notification_kind, }) await self._emit( "subagent_completed", state, { "status": terminal_status, "resident": state.resident, "resident_status": state.status, "accepts_followups": bool(state.resident or not state.completion.is_set()), "pending_messages_count": state.pending_messages_count, "content_preview": result.content[:500], }, ) if state.resident: await self._emit_worker_notification( state, notification_kind=state.last_notification_kind or "idle", summary=result.content or f"{state.name or state.agent_id} is idle", ) except Exception as exc: # pragma: no cover - defensive state.latest_result = str(exc) state.last_notification_kind = "error" state.task_result = TaskResult(status=TaskStatus.FAILED, content=str(exc)) state.status = "idle" if state.resident else TaskStatus.FAILED.value await self._save_state(state, state.status, {"error": str(exc), "notification_kind": state.last_notification_kind}) await self._emit( "subagent_completed", state, { "status": TaskStatus.FAILED.value, "resident": state.resident, "resident_status": state.status, "accepts_followups": bool(state.resident), "pending_messages_count": state.pending_messages_count, "content_preview": str(exc)[:500], }, ) if state.resident: await self._emit_worker_notification( state, notification_kind="error", summary=str(exc), ) finally: state.update_event.set() async def _runner() -> None: current_prompt = prompt try: while True: await _execute_turn(current_prompt) if not state.resident: return state.status = "idle" state.update_event.set() await self._save_state(state, state.status, {"notification_kind": state.last_notification_kind or "idle"}) next_message = await state.inbox.get() state.pending_messages_count = max(0, state.pending_messages_count - 1) current_prompt = str(next_message.get("body", "") or "").strip() if not current_prompt: current_prompt = str(next_message.get("message", "") or "").strip() if not current_prompt: current_prompt = str(next_message) finally: state.completion.set() state.update_event.set() if state.worktree and self.store and hasattr(self.store, "save_runtime_worktree_session"): await self.store.save_runtime_worktree_session( worktree_session_id=f"wt_{agent_id}", runtime_session_id=self.runtime_session_id, task_id=self.parent_task.id if self.parent_task else None, path=str(state.worktree.get("path", "") or ""), status="closed", metadata=dict(state.worktree or {}), ) await cleanup_worktree(state.worktree) if effective_background: state.runtime_task = asyncio.create_task(_runner()) return { "success": True, "agent_id": agent_id, "name": effective_name, "status": "running", "background": True, "resident": state.resident, "resident_status": state.status, "accepts_followups": bool(state.resident), "worktree_path": (state.worktree or {}).get("path", ""), } await _runner() return self._result_payload(state) async def wait(self, agent_id: str, timeout_seconds: int = 300) -> dict[str, Any]: resolved_id = self._resolve_agent_id(agent_id) state = self.states.get(resolved_id) if state is None: return {"error": f"Unknown subagent: {agent_id}", "success": False} if state.resident and state.status != "running": return self._result_payload(state) deadline = time.time() + max(1, int(timeout_seconds or 1)) while True: remaining = deadline - time.time() if remaining <= 0: return self._result_payload(state) if state.completion.is_set(): return self._result_payload(state) if state.resident and state.status != "running": return self._result_payload(state) try: await asyncio.wait_for(state.update_event.wait(), timeout=remaining) except asyncio.TimeoutError: return self._result_payload(state) state.update_event.clear() return self._result_payload(state) async def send(self, agent_id: str, message: str) -> dict[str, Any]: resolved_id = self._resolve_agent_id(agent_id) state = self.states.get(resolved_id) if state is None: return {"error": f"Unknown subagent: {agent_id}", "success": False} if state.completion.is_set(): return {"error": f"Subagent {agent_id} has already completed", "success": False} rendered = str(message or "").strip() self._persist_follow_up_message(state, rendered) await state.inbox.put( { "body": rendered, "message_class": "chat", "actionable": True, "worker_id": state.agent_id, "origin_task_id": str(getattr(self.parent_task, "id", "") or "").strip(), "origin_session_id": str(getattr(self.parent_task, "session_id", "") or "").strip(), } ) state.pending_messages_count += 1 state.update_event.set() await self._save_state(state, state.status, {"queued_message": rendered[:500]}) await self._emit( "subagent_updated", state, { "message": rendered[:500], "resident": state.resident, "resident_status": state.status, "pending_messages_count": state.pending_messages_count, }, ) return {"success": True, "agent_id": state.agent_id, "name": state.name, "status": state.status} def list_agents(self) -> dict[str, Any]: return { "success": True, "agents": [self._result_payload(state) for state in self.states.values()], } def _build_child_task(self, state: SubagentState, prompt: str) -> Task: parent = self.parent_task or Task() metadata = dict(parent.metadata or {}) metadata.pop("_fork_allowed_tools", None) metadata["_native_runtime_depth"] = int(metadata.get("_native_runtime_depth", 0) or 0) + 1 metadata["subagent_profile"] = state.profile metadata["_subagent_name"] = state.name metadata["_subagent_description"] = state.description metadata["_subagent_model"] = state.model metadata["_subagent_mode"] = state.mode metadata["_subagent_max_iterations"] = state.max_iterations metadata["_fork_mode"] = state.fork_mode if state.profile == "verify": metadata[self.config.system.native_runtime.verification_policy.skip_metadata_key] = True metadata["work_item_verification_required"] = False if state.worktree and state.worktree.get("path"): metadata["target_output_dir"] = state.worktree["path"] execution_context = dict((state.worktree or {}).get("execution_context", {}) or {}) if execution_context: metadata["_execution_context"] = execution_context if state.fork_system_prompt: metadata["_runtime_system_prompt_override"] = state.fork_system_prompt if state.fork_context_messages: metadata["_fork_context_messages"] = list(state.fork_context_messages) if state.fork_allowed_tools: metadata["_fork_allowed_tools"] = list(state.fork_allowed_tools) metadata["_comms_endpoint_id"] = state.agent_id metadata["_comms_parent_endpoint_id"] = self._parent_endpoint_id() metadata["_subagent_resident"] = state.resident return Task( title=state.name or state.description or f"{state.profile} subagent", description=prompt, assigned_to=parent.assigned_to, project_id=parent.project_id, session_id=f"{parent.session_id or 'session'}:{state.agent_id}", parent_session_id=parent.session_id, parent_id=parent.id, tags=list(parent.tags), metadata=metadata, ) def _resolve_allowed_tools(self, profile: str, mode: str = "default") -> list[str]: profiles = self.config.agents.native_subagents or {} profile_cfg: NativeSubagentProfileConfig = profiles.get(profile) or profiles.get("general") or NativeSubagentProfileConfig() if profile_cfg.allowed_tools: return self._apply_mode_tool_filter(list(profile_cfg.allowed_tools), mode) read_only = [ "file_read", "file_search", "list_dir", "web_search", "web_fetch", "todo_read", "todo_write", "request_user_input", "agent_spawn", "agent_wait", "agent_send", "agent_list", ] implement = [ "shell_exec", "file_read", "file_write", "file_edit", "file_search", "list_dir", "web_search", "web_fetch", "python_exec", "todo_read", "todo_write", "request_user_input", "agent_spawn", "agent_wait", "agent_send", "agent_list", ] verify = [ "shell_exec", "file_read", "file_search", "list_dir", "web_search", "web_fetch", "python_exec", "browser_navigate", "browser_snapshot", "browser_click", "browser_type", "browser_wait_for", "browser_scroll", "browser_take_screenshot", "browser_close", "todo_read", "todo_write", "request_user_input", "agent_spawn", "agent_wait", "agent_send", "agent_list", ] mapping = { "general": implement, "explore": read_only, "plan": read_only, "implement": implement, "verify": verify, } return self._apply_mode_tool_filter(mapping.get(profile, implement), mode) def _profile_prompt(self, profile: str, mode: str = "default") -> str: prompts = { "general": "Complete the task directly and report only the essential outcome.", "explore": "Read-only exploration only. Do not modify files or system state.", "plan": "Read-only planning only. Produce a concise implementation plan with critical files.", "implement": "Implement directly in the assigned workspace, then verify your changes with commands.", "verify": ( "Try to break the implementation. Prefer executable checks over code reading. " "For every check you actually ran, emit `Check:`, `Command:`, `Observed Output:`, and `Result:` lines. " "End with exactly one line `VERDICT: PASS`, `VERDICT: FAIL`, or `VERDICT: PARTIAL`. " "You may start the final summary with `VERIFIED:` if the work is acceptable, or `ISSUES:` if blocking problems remain." ), } base = prompts.get(profile, prompts["general"]) normalized_mode = str(mode or "default").strip().lower() if normalized_mode == "plan": return base + " Operate in plan mode: do not make filesystem or shell changes." if normalized_mode in {"accept_edits", "bypass_permissions", "dont_ask", "acceptedits", "bypasspermissions", "dontask"}: return base + f" Runtime spawn mode hint: {mode}." return base def _profile_config(self, profile: str) -> NativeSubagentProfileConfig: profiles = self.config.agents.native_subagents or {} return profiles.get(profile) or profiles.get("general") or NativeSubagentProfileConfig() def _apply_mode_tool_filter(self, tools: list[str], mode: str) -> list[str]: if str(mode or "default").strip().lower() != "plan": return list(tools) plan_safe = { "file_read", "file_search", "list_dir", "web_search", "web_fetch", "todo_read", "todo_write", "request_user_input", "agent_spawn", "agent_wait", "agent_send", "agent_list", } return [tool for tool in tools if tool in plan_safe] def _build_child_agent( self, *, profile: str, allowed_tools: list[str], prompt_addendum: str, state: SubagentState, ) -> Any: overrides = { "name": state.name, "description": state.description, "model": state.model, "mode": state.mode, "max_iterations": state.max_iterations, } signature = inspect.signature(self.child_agent_factory) accepts_overrides = len(signature.parameters) >= 4 or any( parameter.kind == inspect.Parameter.VAR_POSITIONAL for parameter in signature.parameters.values() ) if accepts_overrides: return self.child_agent_factory(profile, allowed_tools, prompt_addendum, overrides) return self.child_agent_factory(profile, allowed_tools, prompt_addendum) def _build_permission_bridge(self, state: SubagentState) -> Callable[..., Awaitable[tuple[bool, Any]]]: async def _bridge( *, tool: Any, arguments: dict[str, Any], approval_engine: Any, on_progress: Any = None, ) -> tuple[bool, Any]: parent_task = self.parent_task or Task(project_id="default") metadata = { "category": getattr(tool, "category", "general"), "requires_confirmation": getattr(tool, "requires_confirmation", False), "description": getattr(tool, "description", ""), "subagent_id": state.agent_id, "subagent_profile": state.profile, "subagent_name": state.name, "subagent_mode": state.mode, "bridged_runtime_session_id": self.runtime_session_id, } return await approval_engine.authorize_tool_call( task=parent_task, tool_name=getattr(tool, "name", ""), arguments=arguments, metadata=metadata, on_progress=on_progress, ) return _bridge def _resolve_agent_id(self, agent_id: str) -> str: raw = str(agent_id or "").strip() if raw in self.states: return raw return self.agent_names.get(raw, raw) def _parent_endpoint_id(self) -> str: if self.parent_task is None: return "runtime-parent" session_id = str(getattr(self.parent_task, "session_id", "") or "").strip() task_id = str(getattr(self.parent_task, "id", "") or "").strip() return f"task::{session_id or task_id or 'runtime-parent'}" def _comms_layout(self): if self.parent_task is None: return None workspace_root = ( str(getattr(self.parent_task, "metadata", {}).get("comms_workspace_root", "") or "").strip() or str(getattr(self.parent_task, "metadata", {}).get("workspace_root", "") or "").strip() or str(getattr(self.parent_task, "metadata", {}).get("target_output_dir", "") or "").strip() ) if not workspace_root: return None try: from opc.layer2_organization import comms as _comms return _comms.resolve_layout( workspace_root, str(getattr(self.parent_task, "project_id", "") or "default").strip() or "default", str(getattr(self.parent_task, "parent_session_id", "") or getattr(self.parent_task, "session_id", "") or "default").strip() or "default", ) except Exception: return None def _ensure_comms_endpoint(self, state: SubagentState) -> None: layout = self._comms_layout() if layout is None: return try: from opc.layer2_organization import comms as _comms _comms.ensure_layout(layout, [self._parent_endpoint_id(), state.agent_id]) except Exception: return def _persist_follow_up_message(self, state: SubagentState, message: str) -> None: if not message: return layout = self._comms_layout() if layout is None: return try: from opc.layer2_organization import comms as _comms _comms.send_message( layout, from_role=self._parent_endpoint_id(), to_role=state.agent_id, subject=f"Follow-up for {state.name or state.agent_id}", body=message, priority="normal", extra_frontmatter={ "transport_kind": "system", "semantic_type": "work_update", "message_class": "chat", "actionable": True, "worker_id": state.agent_id, "origin_task_id": str(getattr(self.parent_task, "id", "") or "").strip(), "origin_session_id": str(getattr(self.parent_task, "session_id", "") or "").strip(), "comms_state": "open", "from_endpoint_type": "native_subagent", "to_endpoint_type": "native_subagent", "refs": { "task_id": str(getattr(self.parent_task, "id", "") or "").strip(), "runtime_session_id": self.runtime_session_id, }, }, ) except Exception: return async def _emit(self, event_type: str, state: SubagentState, payload: dict[str, Any]) -> None: if not self.event_bus: return await self.event_bus.publish(OPCEvent( event_type="runtime_event", payload={ "type": event_type, "agent_id": state.agent_id, "profile": state.profile, "task_id": str(getattr(self.parent_task, "id", "") or "").strip(), "session_id": str(getattr(self.parent_task, "session_id", "") or "").strip(), "resident": state.resident, "resident_status": state.status, "pending_messages_count": state.pending_messages_count, **payload, }, )) async def _save_state(self, state: SubagentState, status: str, metadata: dict[str, Any] | None = None) -> None: if not self.store or not hasattr(self.store, "save_runtime_subagent_run"): return merged_metadata = { "name": state.name, "description": state.description, "model": state.model, "mode": state.mode, "isolation": state.isolation, "max_iterations": state.max_iterations, "fork_mode": state.fork_mode, "fork_context_messages": len(state.fork_context_messages), "resident": state.resident, "resident_status": state.status, "accepts_followups": bool(state.resident and not state.completion.is_set()), "pending_messages_count": state.pending_messages_count, "last_notification_kind": state.last_notification_kind, } merged_metadata.update(metadata or {}) await self.store.save_runtime_subagent_run( subagent_run_id=state.agent_id, runtime_session_id=self.runtime_session_id, task_id=self.parent_task.id if self.parent_task else None, agent_id=state.agent_id, profile=state.profile, status=status, worktree_path=str((state.worktree or {}).get("path", "") or ""), metadata=merged_metadata, ) async def _emit_worker_notification( self, state: SubagentState, *, notification_kind: str, summary: str, actionable: bool = False, ) -> None: payload = { "worker_id": state.agent_id, "worker_type": "native_subagent", "notification_kind": str(notification_kind or "idle").strip() or "idle", "summary": str(summary or "").strip(), "task_id": str(getattr(self.parent_task, "id", "") or "").strip(), "session_id": str(getattr(self.parent_task, "session_id", "") or "").strip(), **work_item_identity_payload_for_task(self.parent_task), "projection_id": projection_id_for_task(self.parent_task) if self.parent_task is not None else "", "details_ref": state.agent_id, "actionable": bool(actionable), "resident_status": state.status, "pending_messages_count": state.pending_messages_count, "name": state.name, } await self._emit("worker_notification", state, payload) @staticmethod def _resident_notification_kind(result: TaskResult) -> str: if result.status in {TaskStatus.AWAITING_HUMAN, TaskStatus.AWAITING_REVIEW}: return "permission_needed" if result.status == TaskStatus.AWAITING_PEER: return "blocked" if result.status == TaskStatus.FAILED: return "error" if result.status == TaskStatus.DONE: return "task_complete" return "idle" def _result_payload(self, state: SubagentState) -> dict[str, Any]: payload = { "success": state.task_result is not None and state.task_result.status == TaskStatus.DONE, "agent_id": state.agent_id, "profile": state.profile, "name": state.name, "description": state.description, "model": state.model, "mode": state.mode, "isolation": state.isolation, "max_iterations": state.max_iterations, "status": state.status, "background": state.background, "fork_mode": state.fork_mode, "resident": state.resident, "resident_status": state.status, "accepts_followups": bool(state.resident and not state.completion.is_set()), "pending_messages_count": state.pending_messages_count, "last_notification_kind": state.last_notification_kind, "result": state.latest_result, "worktree_path": (state.worktree or {}).get("path", ""), "venv_path": (state.worktree or {}).get("venv_path", ""), "python_executable": (state.worktree or {}).get("python_executable", ""), } if state.task_result and state.task_result.status in {TaskStatus.AWAITING_HUMAN, TaskStatus.AWAITING_REVIEW}: artifacts = dict(state.task_result.artifacts or {}) payload.update( { "requires_user_input": True, "reason": state.task_result.content, "approval": dict(artifacts.get("approval", {}) or {}), "permission_requests": list(artifacts.get("permission_requests", []) or []), } ) if state.task_result and state.task_result.status == TaskStatus.AWAITING_PEER: artifacts = dict(state.task_result.artifacts or {}) payload.update( { "requires_peer_wait": True, "reason": state.task_result.content, "permission_requests": list(artifacts.get("permission_requests", []) or []), } ) return payload