"""Sub-agent execution — standalone loop without a persistent session.""" from __future__ import annotations import time import uuid 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 current_event_sink, current_stop_event 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 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 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, ) -> tuple[str, bool]: """ Run a sub-agent loop without a persistent session. Returns (result_text, completed_normally). completed_normally is False if the sub-agent hit the iteration limit or timed out. """ from navi.tools._internal.base import ( current_session_id as _sid_var, current_model as _model_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_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) profile = self._profiles.get(profile_id) _model_var.set(profile.model) exclude = set(exclude_tools or []) tool_source = profile.subagent_tools if profile.subagent_tools else profile.enabled_tools tools = [ t for t in build_tool_list(tool_source, profile.mcp_servers, 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) if briefing: sys_parts.append(f"## Task context\n\n{briefing}") if parent_session_id: 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.\n" "When calling MCP navi-3d tools (compile_scad, render_stl, lint_scad), pass ONLY the filename " "(e.g. 'falcon9_rocket.scad') for source_path and output_path. Do NOT include the session_id or " "the session_files directory in those paths — the MCP server resolves them automatically." ) 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.total elif sink is not None: await sink.put(_ev) for iteration in range(max_iterations): if stop_event and stop_event.is_set(): return accumulated_text, False elapsed = time.monotonic() - _start_time if elapsed >= timeout_seconds: log.warning( "agent.subagent.timeout", elapsed=elapsed, timeout=timeout_seconds ) if sink is not None: await sink.put( SubagentComplete( token_count=_turn_tokens, tool_call_count=_sub_tool_count ) ) return accumulated_text or "[Sub-agent timed out]", False 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") self._compressor.check_context_size(built_ctx) 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.prompt_tokens is not None or chunk.completion_tokens is not None ): _turn_tokens += (chunk.prompt_tokens or 0) + ( chunk.completion_tokens or 0 ) 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 accumulated_text, False if thinking_stalled_reason: if sink is not None: await sink.put( SubagentComplete( token_count=_turn_tokens, tool_call_count=_sub_tool_count, ) ) return f"[{thinking_stalled_reason}]", False if not turn_tool_calls: log.info( "agent.subagent.complete", iterations=iteration + 1, result_len=len(accumulated_text), ) if sink is not None: await sink.put( SubagentComplete( token_count=_turn_tokens, tool_call_count=_sub_tool_count, ) ) return accumulated_text, True 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, ) ) for tc in turn_tool_calls: _sub_tool_count += 1 if sink is not None: await sink.put( ToolStarted( tool_name=tc.name, arguments=tc.arguments, is_subagent=True, ) ) 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=tc.arguments ) try: result = await tool.execute(tc.arguments) 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, ) ) 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 ) if sink is not None: await sink.put( SubagentComplete( token_count=_turn_tokens, tool_call_count=_sub_tool_count ) ) return ( accumulated_text or "[Sub-agent reached iteration limit without a final answer]", False, ) finally: _sid_var.set(_prev_sid) _model_var.set(_prev_model) _uid_var.set(_prev_uid) _role_var.set(_prev_role) _uinfo_var.set(_prev_uinfo) def _get_backend(self, backend_key: str) -> LLMBackend: return self._backends.get(backend_key)