"""Sub-agent execution — standalone loop without a persistent session.""" from __future__ import annotations import time import uuid from dataclasses import dataclass from datetime import datetime, timezone from typing import TYPE_CHECKING import structlog from navi.config import settings from navi.exceptions import ContextTooLargeError from navi.llm.base import LLMBackend, Message, ToolCallRequest from navi.tools._internal.base import ToolContext, current_event_sink, current_stop_event from navi.tools._internal.redact import redact_args from .events import AIHelperTokensUsed, SubagentComplete, ToolEvent, ToolStarted, TurnThinking from .stream_guard import _iter_stream_guarded from .tool_utils import build_tool_list if TYPE_CHECKING: from .compressor import ContextCompressor from .context_builder import ContextBuilder from .planning import PlanningEngine from .registry import BackendRegistry, ProfileRegistry, ToolRegistry from .session import SessionStore from .tool_executor import ToolExecutor log = structlog.get_logger() # Sub-agents are execution workers. If a sub-agent produces only thinking for a # long time without text or tool calls, local models can degenerate into endless # internal-token loops and keep the GPU busy with no user-visible progress. _SUBAGENT_THINKING_STALL_SECONDS = 60.0 _SUBAGENT_THINKING_STALL_CHARS = 12_000 # Always injected, whatever profile or custom prompt the caller supplied. A # sub-agent is handed exactly one step and sees a tool set drawn from its # profile's MCP servers — including, before this anchor, a task board, a secret # store and a chat client. Given those, a sub-agent asked to echo one word went # off to pick a project off the board and research it for forty iterations # instead. The rule belongs to the harness rather than to any one persona: the # two profiles without a `subagent_system_prompt.txt` fall back to their main # prompt, so a per-file edit would reach only some sub-agents and drift. _SCOPE_ANCHOR = ( "[Scope]\n" "Your task is exactly the user message below — nothing else. Do not go " "looking for additional work in task boards, memory, chat history or any " "other source, and do not widen the task. If the task looks trivial, just " "do it and report back." ) @dataclass(frozen=True) class SubAgentOutcome: """What one sub-agent run produced, and how it ended. ``status`` is the reason the loop stopped — ``ok``, ``timeout``, ``max_iterations``, ``thinking_stall``, ``user_stop`` or ``context_overflow``. The caller renders its result header from it, so a timeout no longer reaches the parent labelled as an iteration limit. ``text`` always carries the progress report, so a run that stopped early still hands back what it managed to do. """ text: str status: str @property def completed(self) -> bool: """True only when the sub-agent produced a final answer of its own.""" return self.status == "ok" class SubAgentRunner: """Runs a tool-calling sub-agent loop with timeout and thinking-stall guards.""" def __init__( self, profile_registry: ProfileRegistry, tool_registry: ToolRegistry, backend_registry: BackendRegistry, ctx_builder: ContextBuilder, compressor: ContextCompressor, planning: PlanningEngine, tool_executor: ToolExecutor, session_store: SessionStore | None, mcp_manager=None, ) -> None: self._profiles = profile_registry self._tools = tool_registry self._backends = backend_registry self._ctx_builder = ctx_builder self._compressor = compressor self._planning = planning self._tool_executor = tool_executor self._sessions = session_store self._mcp_manager = mcp_manager def set_profile(self, profile) -> None: """Propagate active profile to compressor so subagent uses profile-specific compression settings.""" self._compressor.set_profile(profile) async def run( self, user_message: str, profile_id: str, max_iterations: int = 40, exclude_tools: list[str] | None = None, briefing: str | None = None, custom_system_prompt: str | None = None, inherit_system_prompt: bool = False, context_transfer: str | None = None, parent_session_id: str | None = None, timeout_seconds: float = 300.0, ) -> SubAgentOutcome: """ Run a sub-agent loop without a persistent session. Returns a :class:`SubAgentOutcome`: the progress report plus the final answer, wrapped with the ``status`` the loop ended on. Callers decide what to do with a partial run — the text is always populated. """ from navi.tools._internal.base import ( current_session_id as _sid_var, current_model as _model_var, current_profile_id as _profile_var, current_todo_session_id as _todo_sid_var, current_user_id as _uid_var, current_user_role as _role_var, current_user_info as _uinfo_var, ) _prev_sid = _sid_var.get(None) _prev_model = _model_var.get(None) _prev_profile = _profile_var.get(None) _prev_todo_sid = _todo_sid_var.get(None) _prev_uid = _uid_var.get(None) _prev_role = _role_var.get() _prev_uinfo = _uinfo_var.get(None) subagent_run_id = f"subagent_{uuid.uuid4().hex[:12]}" tool_session_id = parent_session_id or subagent_run_id _sid_var.set(tool_session_id) # Scope the sub-agent's todo list to its own ephemeral run id so its # auto-populated plan and todo updates do NOT clobber the parent session's # todo (which the parent's goal-anchoring reads every iteration). _todo_sid_var.set(subagent_run_id) profile = self._profiles.get(profile_id) _model_var.set(profile.model) # The sub-agent's own profile, not the parent's: tools that default to # "the profile I run as" must see the one this run was spawned with. _profile_var.set(profile.id) exclude = set(exclude_tools or []) scope = profile.get_subagent_tools() tools = [ t for t in build_tool_list(scope.native, scope.mcp, self._tools, self._mcp_manager) if t.name not in exclude ] tool_schemas = [t.schema() for t in tools] llm = self._get_backend(profile.llm_backend) user_id = None if parent_session_id and self._sessions: parent_session = await self._sessions.get(parent_session_id) if parent_session: user_id = parent_session.user_id if user_id is not None: _uid_var.set(user_id) _role_var.set(_prev_role or "user") _uinfo_var.set(_prev_uinfo) else: _uid_var.set(None) _role_var.set("user") _uinfo_var.set(None) mem = await self._ctx_builder._memory_msg(user_id=user_id) sys_parts: list[str] = [] if inherit_system_prompt: sys_parts.append(profile.system_prompt) if profile.subagent_system_prompt: sys_parts.append(profile.subagent_system_prompt) if custom_system_prompt: sys_parts.append(custom_system_prompt) sys_parts.append(_SCOPE_ANCHOR) if briefing: sys_parts.append(f"## Task context\n\n{briefing}") if parent_session_id: # Generic session plumbing only. Server-specific calling conventions # (path shapes, session_id handling) belong to the MCP server that # owns them — in its tool schemas and INSTRUCTIONS, which reach # every agent, not just sub-agents. sys_parts.append( "[Parent session context]\n" f"Parent Session ID: {parent_session_id}\n" f"Session files directory: {settings.session_files_dir}/{parent_session_id}/\n" "For files the user should see, write to this exact session directory. " "Do not use or invent a subagent_* directory." ) if not sys_parts: sys_parts.append(profile.system_prompt) subagent_sys_msg = Message(role="system", content="\n\n---\n\n".join(sys_parts)) context: list[Message] = [] if context_transfer: context.append( Message( role="user", content=f"## Context from parent agent\n\n{context_transfer}", ) ) context.append( Message( role="assistant", content="Understood. I have the context. Ready to begin the task.", ) ) context.append( Message(role="user", content=user_message, created_at=datetime.now(timezone.utc)) ) sink = current_event_sink.get() log.info( "agent.subagent.start", profile_id=profile_id, max_iterations=max_iterations, tools=len(tools), planning=profile.subagent_planning_enabled, ) stop_event = current_stop_event.get() tool_map = {t.name: t for t in tools} subagent_think = ( profile.subagent_think_enabled if profile.subagent_think_enabled is not None else profile.think_enabled ) _turn_tokens: int = 0 _sub_tool_count: int = 0 _start_time = time.monotonic() accumulated_text = "" try: if profile.subagent_planning_enabled: async for _ev in self._planning.run( context, profile, llm, mem, tool_schemas, system_prompt_override=subagent_sys_msg.content, is_subagent=True, ): if isinstance(_ev, AIHelperTokensUsed): _turn_tokens += _ev.completion_tokens elif sink is not None: await sink.put(_ev) for iteration in range(max_iterations): if stop_event and stop_event.is_set(): return await self._finish( sink=sink, context=context, reason="user_stop", status="user_stop", iterations=iteration, text=accumulated_text, token_count=_turn_tokens, tool_call_count=_sub_tool_count, ) elapsed = time.monotonic() - _start_time if elapsed >= timeout_seconds: log.warning( "agent.subagent.timeout", elapsed=elapsed, timeout=timeout_seconds ) return await self._finish( sink=sink, context=context, reason="timeout", status="timeout", iterations=iteration, text=accumulated_text, fallback="[Sub-agent timed out]", token_count=_turn_tokens, tool_call_count=_sub_tool_count, ) log.debug("agent.subagent.iteration", iteration=iteration) accumulated_text = "" accumulated_thinking = "" turn_tool_calls: list[ToolCallRequest] | None = None thinking_started_at: float | None = None thinking_stalled_reason: str | None = None built_ctx: list[Message] = [subagent_sys_msg] if mem: built_ctx.append(mem) mcp_msg = self._ctx_builder._mcp_context_msg(profile) if mcp_msg: built_ctx.append(mcp_msg) built_ctx.extend(m for m in context if m.role != "system") try: self._compressor.check_context_size(built_ctx) except ContextTooLargeError: # The parent compresses its own context; a sub-agent has no # session to compress into, so it stops here instead of # dying — and hands back everything it had done by then. A # raised error used to reach the parent as a bare # "Sub-agent failed: …", discarding the whole run. log.warning( "agent.subagent.context_overflow", iteration=iteration, messages=len(built_ctx), ) return await self._finish( sink=sink, context=context, reason="context_overflow", status="context_overflow", iterations=iteration, text=accumulated_text, fallback="[Sub-agent ran out of context before finishing]", token_count=_turn_tokens, tool_call_count=_sub_tool_count, ) async for chunk in _iter_stream_guarded( llm.stream_complete( built_ctx, tools=tool_schemas if tools else None, temperature=profile.temperature, model=profile.model, think=subagent_think, top_k=profile.top_k, top_p=profile.top_p, num_thread=profile.num_thread, ), stop_event=stop_event, first_chunk_timeout=settings.llm_stream_first_chunk_timeout, chunk_timeout=settings.llm_stream_chunk_timeout, ): if chunk.completion_tokens is not None: _turn_tokens += chunk.completion_tokens if chunk.thinking: if thinking_started_at is None: thinking_started_at = time.monotonic() accumulated_thinking += chunk.thinking thinking_elapsed = time.monotonic() - thinking_started_at if ( thinking_elapsed >= _SUBAGENT_THINKING_STALL_SECONDS or len(accumulated_thinking) >= _SUBAGENT_THINKING_STALL_CHARS ): thinking_stalled_reason = ( "Sub-agent produced only thinking output for " f"{thinking_elapsed:.0f}s / {len(accumulated_thinking)} chars " "without text or tool calls." ) log.warning( "agent.subagent.thinking_stall", elapsed=thinking_elapsed, chars=len(accumulated_thinking), profile_id=profile_id, ) break if chunk.delta: accumulated_text += chunk.delta if chunk.tool_calls: turn_tool_calls = chunk.tool_calls if stop_event and stop_event.is_set(): return await self._finish( sink=sink, context=context, reason="user_stop", status="user_stop", iterations=iteration + 1, text=accumulated_text, token_count=_turn_tokens, tool_call_count=_sub_tool_count, ) if thinking_stalled_reason: return await self._finish( sink=sink, context=context, reason="thinking_stall", status="thinking_stall", iterations=iteration + 1, text=f"[{thinking_stalled_reason}]", token_count=_turn_tokens, tool_call_count=_sub_tool_count, ) if not turn_tool_calls: log.info( "agent.subagent.complete", iterations=iteration + 1, result_len=len(accumulated_text), ) return await self._finish( sink=sink, context=context, reason="completed", status="ok", iterations=iteration + 1, text=accumulated_text, token_count=_turn_tokens, tool_call_count=_sub_tool_count, ) if accumulated_thinking and sink is not None: log.debug( "agent.subagent.turn_thinking", length=len(accumulated_thinking) ) await sink.put( TurnThinking(thinking=accumulated_thinking, is_subagent=True) ) context.append( Message( role="assistant", content=accumulated_text or None, tool_calls=turn_tool_calls, ) ) tool_ctx = ToolContext( session_id=tool_session_id, profile_id=profile.id, event_sink=sink, stop_event=stop_event, model=profile.model, user_id=_uid_var.get(None), user_role=_role_var.get(), user_info=_uinfo_var.get(None), ) for tc in turn_tool_calls: # Cooperative stop: if the user clicked Stop before this tool, # skip remaining tools in this batch. if stop_event and stop_event.is_set(): log.info("agent.subagent.tool_batch_stopped", tool=tc.name) return await self._finish( sink=sink, context=context, reason="user_stop", status="user_stop", iterations=iteration + 1, token_count=_turn_tokens, tool_call_count=_sub_tool_count, ) _sub_tool_count += 1 if sink is not None: from navi.tools.todo import started_metadata_for_call await sink.put( ToolStarted( tool_name=tc.name, arguments=tc.arguments, is_subagent=True, tool_call_id=tc.id, metadata=await started_metadata_for_call(tc, tool_ctx), ) ) tool = tool_map.get(tc.name) image_msg = None metadata: dict = {} if tool is None: content = f"Error: tool '{tc.name}' not found." success = False else: log.info( "tool.execute.subagent", tool=tc.name, args=redact_args(tc.arguments) ) try: # Same background interception as the main loop: # background=true detaches into TaskManager (and # the flag never reaches the tool itself). None = # run inline as before. result = self._tool_executor._maybe_background( tc, tc.name, tool, tool_ctx ) if result is None: result = await tool.execute(tc.arguments, ctx=tool_ctx) content = result.to_message_content() success = result.success metadata = result.metadata or {} if ( result.success and result.metadata and result.metadata.get("is_image") ): b64 = result.metadata.get("base64") if b64: image_msg = Message( role="user", content=f"[Image loaded via {tc.name} — analyse it]", images=[b64], ) except Exception as exc: log.warning( "agent.subagent.tool_exception", tool=tc.name, error=str(exc), ) content = f"Error: {exc}" success = False metadata = {} if sink is not None: await sink.put( ToolEvent( tool_name=tc.name, arguments=tc.arguments, result=content, success=success, is_subagent=True, tool_call_id=tc.id, ) ) context.append( Message( role="tool", content=content, tool_call_id=tc.id, name=tc.name, metadata=metadata, ) ) if image_msg: context.append(image_msg) log.warning( "agent.subagent.max_iterations", max_iterations=max_iterations ) return await self._finish( sink=sink, context=context, reason="max_iterations", status="max_iterations", iterations=max_iterations, text=accumulated_text, fallback="[Sub-agent reached iteration limit without a final answer]", token_count=_turn_tokens, tool_call_count=_sub_tool_count, ) finally: _sid_var.set(_prev_sid) _model_var.set(_prev_model) _profile_var.set(_prev_profile) _todo_sid_var.set(_prev_todo_sid) _uid_var.set(_prev_uid) _role_var.set(_prev_role) _uinfo_var.set(_prev_uinfo) @staticmethod async def _finish( *, sink, context: list[Message], reason: str, status: str, iterations: int, text: str = "", fallback: str = "", token_count: int = 0, tool_call_count: int = 0, ) -> SubAgentOutcome: """Emit completion metrics and build the outcome for one way out of the loop. Every exit routes through here so token accounting is never dropped (the parent's context budget reads it) and so no exit returns a bare string: the progress report always leads, and ``fallback`` supplies a body when the sub-agent produced no text of its own. """ if sink is not None: await sink.put( SubagentComplete(token_count=token_count, tool_call_count=tool_call_count) ) report = SubAgentRunner._build_progress_report(context, reason, iterations) body = text.strip() or fallback return SubAgentOutcome(text=f"{report}\n\n{body}" if body else report, status=status) @staticmethod def _build_progress_report( context: list[Message], reason: str, iterations_used: int, ) -> str: """Serialize the sub-agent's execution history into a progress report. The report lists every tool call the sub-agent made, the arguments, and a short result snippet so the parent agent knows what was accomplished before the sub-agent stopped. """ lines = [ f"[Sub-agent stopped: {reason}]", f"Iterations used: {iterations_used}", "", ] turn_number = 0 msg_idx = 0 while msg_idx < len(context): msg = context[msg_idx] if msg.role == "assistant" and msg.tool_calls: turn_number += 1 turn_lines = [f"Turn {turn_number}:", ""] if msg.content: text = msg.content.strip().replace("\n", " ") turn_lines.append(f" Assistant text: {text[:200]}{'...' if len(text) > 200 else ''}") for tc in msg.tool_calls: # Find matching tool result result_msg = None for m in context[msg_idx + 1 :]: if m.role == "tool" and m.tool_call_id == tc.id: result_msg = m break if result_msg: status = "error" if result_msg.content.startswith("Error:") else "success" snippet = result_msg.content.strip().replace("\n", " ")[:120] turn_lines.append(f" - {tc.name} → {status}: {snippet}{'...' if len(result_msg.content) > 120 else ''}") else: args = str(redact_args(tc.arguments)).replace("\n", " ")[:80] turn_lines.append(f" - {tc.name}({args}) → (no result yet)") lines.extend(turn_lines) lines.append("") msg_idx += 1 + len(msg.tool_calls) elif msg.role == "assistant" and msg.content: turn_number += 1 text = msg.content.strip().replace("\n", " ") lines.append(f"Turn {turn_number}: Assistant response: {text[:200]}{'...' if len(text) > 200 else ''}") lines.append("") msg_idx += 1 else: msg_idx += 1 if turn_number == 0: lines.append("No tool calls recorded.") return "\n".join(lines) def _get_backend(self, backend_key: str) -> LLMBackend: return self._backends.get(backend_key)