""" spawn_agent — delegates a focused sub-task to an isolated agent instance. The sub-agent runs its own tool-calling loop with a clean context window. It cannot spawn further sub-agents (recursion is blocked via exclude_tools). The result is returned as a plain text summary. """ import structlog from navi.exceptions import ProfileNotFound from navi.profiles.base import admin_only_blocked from ._internal.base import Tool, ToolContext, ToolResult, current_session_id, current_user_role log = structlog.get_logger() def _visible_profile(profile, role: str | None) -> bool: """A profile this role may be told exists.""" return not getattr(profile, "is_hidden", False) and not admin_only_blocked(profile, role) # The parent synthesises the sub-agent's result into its own answer, so a long # run must not flood the parent's context: Anthropic's multi-agent write-up puts # a useful sub-agent digest at ~1–2k tokens, and this is the boundary that # enforces it. Head *and* tail are kept — the head carries the status line and # the early turns, the tail carries the sub-agent's final answer. _MAX_RESULT_CHARS = 8_000 # Before this, every non-ok run reached the parent as "hit iteration limit". # A timeout reported as an iteration limit is a wrong diagnosis the parent then # repeats to the user, since it synthesises from this header. _STATUS_LABELS = { "timeout": "Sub-agent timed out", "max_iterations": "Sub-agent hit the iteration limit", "thinking_stall": "Sub-agent stalled while thinking", "user_stop": "Sub-agent was stopped by the user", "context_overflow": "Sub-agent ran out of context", } def _truncate_result(text: str) -> str: """Bound the result at ``_MAX_RESULT_CHARS``, keeping both ends.""" if len(text) <= _MAX_RESULT_CHARS: return text half = _MAX_RESULT_CHARS // 2 dropped = len(text) - 2 * half return ( f"{text[:half]}\n\n" f"[... {dropped} characters truncated from the middle ...]\n\n" f"{text[-half:]}" ) def _render_result(text: str, status: str) -> str: """Wrap the sub-agent's report in a header naming how the run actually ended.""" if status == "ok": header = ( "[Sub-agent completed — USER CANNOT SEE THIS. " "Synthesise the findings into your own response.]" ) else: label = _STATUS_LABELS.get(status, "Sub-agent stopped") header = ( f"[{label} — result may be incomplete. USER CANNOT SEE THIS. " "Synthesise what was found and note what is missing.]" ) return f"{header}\n\n{_truncate_result(text)}" class SpawnAgentTool(Tool): name = "spawn_agent" description = ( "Delegate EXACTLY ONE step of your plan to an isolated sub-agent with its own " "context and tool loop.\n\n" "Use it when a step needs 3+ tool calls as one logical unit — never for a single " "call, and one call per plan step, never bundling several steps into one " "sub-agent.\n\n" "Synchronous by default (blocks until it finishes); background=true detaches and " "returns a task_id. The user cannot see sub-agent output — synthesise findings " "into your own response.\n\n" "Omit profile_id unless the step needs another profile's specialisation.\n\n" 'Details, examples and the profile list: tool_manual("spawn_agent").' ) parameters = { "type": "object", "properties": { "task": { "type": "string", "description": ( "What the sub-agent must accomplish: exact goal, success criteria, " "expected output format. " "End with: 'Complete ALL assigned work before responding. Your output is final.'" ), }, "briefing": { "type": "string", "description": ( "Static context injected as system-level instruction: " "IPs, credentials, file paths, constraints, step-by-step instructions." ), }, "profile_id": { "type": "string", "description": ( "Defaults to the current session's profile — the right choice for most " "work. Set it only to specialise: 'server_admin' (remote ops), " "'secretary' (research/writing), 'developer' (code, incl. Navi MCP " "servers)." ), }, "system_prompt": { "type": "string", "description": ( "Optional role definition, injected on top of the profile default. " "e.g. 'You are a security auditor. Report findings by severity.'" ), }, "max_iterations": { "type": "integer", "description": "Maximum tool-call iterations for the sub-agent (default: 40).", }, "background": { "type": "boolean", "description": ( "Detach: returns a task_id immediately; collect via tasks, or wait for the " "completion note. Use above ~45-60s. Default false (synchronous)." ), }, "inherit_system_prompt": { "type": "boolean", "description": ( "Start from the parent profile's full system prompt as a base, then overlay " "this sub-agent's own specialisation. Default false — the sub-agent uses only " "its own subagent prompt." ), }, }, "required": ["task"], } def __init__( self, profile_registry, tool_registry, backend_registry, session_store, memory_store=None, mcp_manager=None, ) -> None: self._profile_registry = profile_registry self._tool_registry = tool_registry self._backend_registry = backend_registry self._session_store = session_store self._memory_store = memory_store self._mcp_manager = mcp_manager async def execute(self, params: dict, ctx: ToolContext | None = None) -> ToolResult: # Import here to avoid module-level circular import from navi.core.agent import Agent from navi.tools.scratchpad import get_section # Everything that can fail lives inside the try: a bad-but-schema-plausible # argument used to raise out of the tool and reach the parent as a generic # tool_exception warning instead of a clean error result the model can read. try: task = (params.get("task") or "").strip() role = ctx.user_role if ctx else current_user_role.get() if not task: return ToolResult( success=False, output="spawn_agent requires a non-empty 'task'.", error="missing_task", ) briefing = (params.get("briefing") or "").strip() or None custom_system_prompt = (params.get("system_prompt") or "").strip() or None try: max_iterations = int(params.get("max_iterations") or 40) except (TypeError, ValueError): return ToolResult( success=False, output=( "max_iterations must be an integer, got " f"{params.get('max_iterations')!r}." ), error="invalid_max_iterations", ) inherit_system_prompt = bool(params.get("inherit_system_prompt", False)) # task → user message (what to do) # briefing → system prompt (context, credentials, instructions) user_message = task # Resolve profile: explicit override → parent session's profile → first available # Support both 'profile_id' and 'profile' as parameter names since models sometimes # use 'profile' when the description says "Profile to use". profile_id = (params.get("profile_id") or params.get("profile") or "").strip() if not profile_id: profile_id = await self._resolve_parent_profile(ctx) try: selected_profile = self._profile_registry.get(profile_id) except ProfileNotFound: available = ", ".join( p.id for p in self._profile_registry.all() if _visible_profile(p, role) ) return ToolResult( success=False, output=( f"Unknown sub-agent profile_id: {profile_id!r}. " f"Available profiles: {available or '(none)'}." ), error=f"unknown_profile:{profile_id}", ) if admin_only_blocked(selected_profile, role): return ToolResult( success=False, output=f"Sub-agent profile '{profile_id}' requires admin access.", error="admin_only", ) # Read parent scratchpad context_transfer section and pass it to the sub-agent. parent_sid = ctx.session_id if ctx else current_session_id.get() context_transfer = (await get_section(parent_sid, "context_transfer")) if parent_sid else "" scope = selected_profile.get_subagent_tools() log.info("spawn_agent.start", profile_id=profile_id, max_iterations=max_iterations, task_preview=task[:80], has_briefing=bool(briefing), has_context_transfer=bool(context_transfer), has_system_prompt=bool(custom_system_prompt), subagent_tools=len(scope.native) + sum(len(v) for v in scope.mcp.values())) agent = Agent( session_store=None, # ephemeral — no DB access profile_registry=self._profile_registry, tool_registry=self._tool_registry, backend_registry=self._backend_registry, workers=[], # no post-response workers for sub-agents memory_store=self._memory_store, mcp_manager=self._mcp_manager, ) outcome = await agent.run_ephemeral( user_message=user_message, profile_id=profile_id, max_iterations=max_iterations, exclude_tools=["spawn_agent"], # prevent recursion briefing=briefing, custom_system_prompt=custom_system_prompt, inherit_system_prompt=inherit_system_prompt, context_transfer=context_transfer or None, parent_session_id=parent_sid, timeout_seconds=300.0, ) log.info("spawn_agent.done", profile_id=profile_id, status=outcome.status, result_len=len(outcome.text)) return ToolResult( success=outcome.completed, output=_render_result(outcome.text, outcome.status), metadata={ "completed": outcome.completed, "status": outcome.status, "result_len": len(outcome.text), }, ) except Exception as e: log.error("spawn_agent.error", error=str(e), exc_info=True) return ToolResult(success=False, output=f"Sub-agent failed: {e}", error=str(e)) async def _resolve_parent_profile(self, ctx: ToolContext | None = None) -> str: """Return the profile of the current parent session, or fallback to first profile.""" session_id = ctx.session_id if ctx else current_session_id.get() if session_id and self._session_store: try: session = await self._session_store.get(session_id) if session: return session.profile_id except Exception: pass # Fallback: the first profile this role may actually run — never an # admin-only one, which would hand a non-admin a wider tool surface. role = ctx.user_role if ctx else current_user_role.get() allowed = [p for p in self._profile_registry.all() if _visible_profile(p, role)] return allowed[0].id if allowed else "secretary"