""" Agent: the tool-calling loop. Flow: 1. Receive user message, load session + profile 2. Build tool schemas from profile's enabled_tools 3. Loop (up to max_iterations): a. Call LLM with session.context (may be compressed) + tool schemas b. If finish_reason == "stop" -> stream final response c. If finish_reason == "tool_calls" -> execute tools, append to both session.messages (display history) and session.context (LLM context) 4. After StreamEnd: run workers sequentially (e.g. context compression) session.messages — full display history, never compressed session.context — what the LLM sees; workers may compress this """ import asyncio import base64 import io import time from datetime import datetime, timezone from pathlib import Path from typing import TYPE_CHECKING, AsyncGenerator import structlog from PIL import Image from navi.config import settings from navi.exceptions import ( ContextTooLargeError, LLMBackendError, LLMConnectionError, MaxIterationsReached, NothingToCompactError, SessionNotFound, ) from navi.llm.base import LLMBackend, Message from navi.tools._internal.base import ( Tool, ToolContext, current_event_sink, current_plan_runner, current_stop_event, current_user_role, current_user_info, ) from .agent_run_context import AgentTurnContext, StreamState from .anti_stall import AntiStallMonitor from .compressor import ContextCompressor, should_compress from .context_builder import ContextBuilder from .planning import PlanningEngine from navi.tools.plan import PlanRunner from .stream_guard import _iter_stream_guarded from .subagent_runner import SubAgentRunner from .tool_utils import build_tool_list from .events import ( AgentEvent, AIHelperTokensUsed, CompressionStarted, ModelInfo, StreamEnd, StreamStopped, SubagentComplete, TextDelta, ThinkingDelta, ThinkingEnd, TodoUpdated, ToolEvent, ToolStarted, ) from .registry import BackendRegistry, ProfileRegistry, ToolRegistry from .session import SessionStore from .tool_executor import ToolExecutor if TYPE_CHECKING: from navi.context_providers._loader import ContextProviderRegistry from navi.memory.store import MemoryStore from navi.profiles.base import AgentProfile from navi.workers.base import Worker log = structlog.get_logger() # Sentinel: placed in the event sink by the tool wrapper to signal completion. _TOOL_DONE = object() async def _todo_progress_message( session_id: str, *, first_iteration: bool = False ) -> "Message | None": """Build a compact system reminder with current todo state and update discipline.""" from navi.tools.todo import get_progress_message return await get_progress_message(session_id, first_iteration=first_iteration) # Final-turn intercept nudges — injected when the model ends a turn with bare # text (no tool call) while the todo still has open steps, then the loop is # continued instead of closed. Escalating tone: 1st is a soft pointer to the # available recovery tools, 2nd is a hard "act now, no bare text" instruction. # Indexed by turn_ctx.final_interceptions (clamped to the last/hardest). _FINAL_INTERCEPT_NUDGES = ( "[Final-turn check] You ended the turn with text only, but the todo list still has open steps " "(pending/in_progress). Do not stop with bare text while work remains. Either mark finished " "steps done/skipped via the todo tool, reflect on what is blocking, re-plan with the `plan` " "tool if the approach is dead, or call a tool to continue — then keep going.", "[Final-turn check — second stop] You already stopped once without acting and the todo still " "has open steps. You must act now: call a tool — todo (close steps as done/skipped/failed with " "validation), reflect, `plan` (re-plan with a reason), or continue execution. Do not output " "text without a tool call.", ) class Agent: def __init__( self, session_store: "SessionStore | None", profile_registry: ProfileRegistry, tool_registry: ToolRegistry, backend_registry: BackendRegistry, workers: list["Worker"] | None = None, memory_store: "MemoryStore | None" = None, cp_registry: "ContextProviderRegistry | None" = None, mcp_manager=None, ) -> None: self._sessions = session_store self._profiles = profile_registry self._tools = tool_registry self._backends = backend_registry self._workers: list["Worker"] = workers or [] self._memory = memory_store self._cp_registry = cp_registry self._mcp_manager = mcp_manager self._ctx_builder = ContextBuilder( profile_registry=profile_registry, memory_store=memory_store, cp_registry=cp_registry, mcp_manager=mcp_manager, ) self._tool_executor = ToolExecutor(tool_registry) self._planning = PlanningEngine(self._ctx_builder) self._compressor = ContextCompressor() self._subagent = SubAgentRunner( profile_registry=profile_registry, tool_registry=tool_registry, backend_registry=backend_registry, ctx_builder=self._ctx_builder, compressor=self._compressor, planning=self._planning, tool_executor=self._tool_executor, session_store=session_store, mcp_manager=mcp_manager, ) def _set_active_profile(self, profile) -> None: """Update cached profile and propagate to subsystems that depend on it.""" self._compressor.set_profile(profile) self._subagent.set_profile(profile) # ------------------------------------------------------------------ # Public interface # ------------------------------------------------------------------ async def run( self, session_id: str, user_message: str, images: list[str] | None = None, files: list[dict] | None = None, is_recall: bool = False, ) -> str: """Non-streaming: run the full tool-calling loop and return the final text.""" full_content = "" async for event in self.run_stream( session_id, user_message, images=images, files=files, is_recall=is_recall ): if isinstance(event, StreamEnd): full_content = event.full_content or "" return full_content async def run_ephemeral( 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]: """Delegate to SubAgentRunner.""" return await self._subagent.run( user_message=user_message, profile_id=profile_id, max_iterations=max_iterations, exclude_tools=exclude_tools, briefing=briefing, custom_system_prompt=custom_system_prompt, inherit_system_prompt=inherit_system_prompt, context_transfer=context_transfer, parent_session_id=parent_session_id, timeout_seconds=timeout_seconds, ) async def compact_stream(self, session_id: str) -> AsyncGenerator[AgentEvent, None]: """Force context compression now, bypassing the token threshold. Triggered by the TUI ``/compact`` command. Unlike the pre/mid-turn auto-compression (gated by ``should_compress``), this always attempts a summary pass as long as there is enough to summarize. Yields ``CompressionStarted`` then ``ContextCompressed`` on success; raises ``NothingToCompactError`` when the context is too small to compress (the orchestrator surfaces it as an error event so the user gets feedback). Emits NO ``StreamStart``/``StreamEnd`` — the client distinguishes a forced compact from an in-turn auto-compress by the absence of a streaming turn (the TUI keys off its ``_streaming`` flag). """ from navi.tools._internal.base import current_session_id as _sid_var session = await self._sessions.get(session_id) if session is None: raise SessionNotFound(session_id) profile = self._profiles.get(session.profile_id) self._set_active_profile(profile) llm = self._get_backend(profile.llm_backend) _sid_token = _sid_var.set(session_id) try: if not settings.context_compression_enabled: raise NothingToCompactError("Context compression is disabled.") yield CompressionStarted( context_tokens=self._compressor.estimate_context_tokens(session.context), max_context_tokens=settings.ollama_num_ctx, ) event = await self._compressor.compress_and_save_session( session=session, session_store=self._sessions, llm=llm, model=profile.model, temperature=settings.context_summary_temperature, session_id=session_id, reason="forced", keep_recent=settings.context_keep_recent, max_tokens=settings.context_summary_max_tokens, # Enable the intra-turn fallback so a single long autonomous turn # (one user message + many tool iterations = one "turn", the # common navi_code shape) can still be compressed — without this # keep_recent_messages, partition_messages finds nothing to # summarize when turns <= keep_recent and forced compact always # reports "nothing to compact". Mirrors the midturn auto-compress. keep_recent_messages=max(12, settings.context_keep_recent * 2), ) if event is None: raise NothingToCompactError( "Nothing to compact yet — the context is still small." ) yield event finally: _sid_var.reset(_sid_token) async def run_stream( self, session_id: str, user_message: str, images: list[str] | None = None, display_message: str | None = None, files: list[dict] | None = None, is_recall: bool = False, cwd: str | None = None, hidden: bool = False, ) -> AsyncGenerator[AgentEvent, None]: """ Streaming variant. Yields AgentEvent objects: - ThinkingDelta / ThinkingEnd: reasoning chunks - ToolEvent: tool call + result - TextDelta / StreamEnd: final streamed response - ContextCompressed: emitted by workers after compression Args: user_message: The text sent to the LLM (may contain injected hints). display_message: Optional text shown to the user in the chat UI. When omitted, user_message is used for both LLM context and display. hidden: If True, create a single user message with is_display=False and is_context=True. Used for form submissions and other invisible user input that must reach the LLM but should never appear in the UI. """ session = await self._sessions.get(session_id) if session is None: raise SessionNotFound(session_id) profile = self._profiles.get(session.profile_id) tools = self._tool_list(profile.get_agent_tools()) tool_schemas = [t.schema() for t in tools] llm = self._get_backend(profile.llm_backend) mem = await self._ctx_builder._memory_msg(user_id=session.user_id) # Expose session_id, model and cwd to tools via ContextVar from navi.tools._internal.base import ( current_session_id as _sid_var, current_model as _model_var, current_working_directory as _cwd_var, ) _sid_token = _sid_var.set(session_id) _model_var.set(profile.model) _cwd_token = _cwd_var.set(cwd) # Pre-turn compression: if the last turn filled the context past the # threshold, compress NOW before calling the LLM. This prevents the # model from seeing an over-full context and generating gibberish # (e.g. a "summary of the conversation" instead of a real answer). async for _ev in self._compression_events_preturn(session, llm, profile, session_id): yield _ev if hidden: # Hidden user input (e.g. form submission): one message, context-only, # never shown in the UI. No image budgeting is applied because hidden # messages are expected to carry plain JSON payloads. user_msg = Message( role="user", content=user_message, images=images or None, files=files or None, created_at=datetime.now(timezone.utc), is_recall=is_recall, is_display=False, is_context=True, ) session.messages.append(user_msg) session.context.append(user_msg) await self._sessions.save(session) else: display_text = display_message if display_message is not None else user_message user_msg_display = Message(role="user", content=display_text, images=images or None, files=files or None, created_at=datetime.now(timezone.utc), is_recall=is_recall, is_context=False) # Image token budgeting: fit as many images as possible into the LLM context. # Overflow images are saved to the session directory so Navi can view them # later via image_view if needed. images_for_context = images context_content = user_message if images: current_tokens = self._compressor.estimate_context_tokens(session.context) available_tokens = int(settings.ollama_num_ctx * 0.8) - current_tokens max_images = max(0, available_tokens // 500) if max_images < len(images): images_for_context = images[:max_images] overflow = images[max_images:] saved_names = [] try: session_dir = Path(settings.session_files_dir) / session_id session_dir.mkdir(parents=True, exist_ok=True) for idx, b64 in enumerate(overflow): raw = base64.b64decode(b64) img = Image.open(io.BytesIO(raw)) img = img.convert("RGB") w, h = img.size if w > 1024 or h > 1024: ratio = 1024 / max(w, h) img = img.resize((int(w * ratio), int(h * ratio)), Image.LANCZOS) name = f"uploaded_{idx}.jpg" path = session_dir / name img.save(path, format="JPEG", quality=85, optimize=True) saved_names.append(name) except Exception: pass if saved_names: context_content += ( f"\n\n[Additional images saved to session directory: {', '.join(saved_names)}]" ) user_msg_context = Message(role="user", content=context_content, images=images_for_context or None, files=files or None, created_at=datetime.now(timezone.utc), is_recall=is_recall, is_display=False) session.messages.append(user_msg_display) session.messages.append(user_msg_context) session.context.append(user_msg_context) # Persist user message immediately so it survives a client disconnect # before the assistant reply is ready. await self._sessions.save(session) stop_event = current_stop_event.get() turn_ctx = AgentTurnContext(turn_start=time.monotonic()) # Final-turn intercept nudge pending injection into the next iteration's # built_ctx. Lives in the generator's local scope so it survives across # iterations of the for-loop (injections are ephemeral — built_ctx is # rebuilt each iteration — so this is the channel that carries the # nudge from the intercept site to the next turn's context). pending_final_nudge: Message | None = None ctx_task = asyncio.create_task(self._ctx_builder._collect_context_injections(profile)) mem_facts_task = asyncio.create_task( self._ctx_builder._memory_facts_msg( user_message, user_id=session.user_id, injected_ids=turn_ctx.injected_fact_ids, scope_boundary_enabled=profile.scope_boundary_enabled, session_cwd=(session.session_metadata or {}).get("cwd"), ) ) ctx_injections = await ctx_task mem_facts = await mem_facts_task if mem_facts: ctx_injections.append(mem_facts) log.debug( "agent.memory_facts_injected", session_id=session_id, facts_msg_length=len(mem_facts.content or ""), ) else: log.debug("agent.memory_facts_none", session_id=session_id) anti_stall = AntiStallMonitor(profile) if profile.anti_stall_enabled: await anti_stall.init(session_id) # Tool-calling loop — uses stream_complete() for every turn so thinking # is captured in real-time via ThinkingDelta/ThinkingEnd events. for iteration in range(profile.max_iterations): # Cooperative stop: check before each LLM call if stop_event and stop_event.is_set(): await self._sessions.save(session) yield StreamStopped() return async for _ev in self._compression_events_midturn( session, llm, profile, session_id, iteration, ctx_injections, mem ): yield _ev state = StreamState() built_ctx = self._ctx_builder.build( session.context, profile, mem, iteration=iteration, max_iterations=profile.max_iterations, extra_system=ctx_injections, session_id=session_id, session_metadata=session.session_metadata, ) if ( profile.goal_anchoring_enabled and iteration > 0 and iteration % profile.goal_anchoring_interval == 0 ): built_ctx.append( await self._ctx_builder._build_goal_anchor(session_id, user_message) ) todo_msg = await _todo_progress_message(session_id, first_iteration=(iteration == 0)) if todo_msg: built_ctx.append(todo_msg) # Anti-stall / adaptive re-plan: inject intervention message if needed _stall_msg = await anti_stall.pre_turn(session_id, iteration) if _stall_msg: built_ctx.append(_stall_msg) # Final-turn intercept nudge carried over from a previous bare-text # turn: the model stopped to narrate while the todo had open steps, # so we inject the recovery pointer and continue the loop. if pending_final_nudge is not None: built_ctx.append(pending_final_nudge) pending_final_nudge = None try: self._compressor.check_context_size(built_ctx, session_context=session.context) except ContextTooLargeError as e: # Surface the error as a Navi response (not a raw system error) so the # user sees a coherent message and the exchange is saved to history. error_text = str(e) assistant_msg = Message(role="assistant", content=error_text) session.context.append(assistant_msg) session.messages.append(assistant_msg) await self._sessions.save(session) yield TextDelta(delta=error_text) yield StreamEnd(content=error_text) return async for _ev in self._consume_stream( _iter_stream_guarded( llm.stream_complete( built_ctx, tools=tool_schemas if tools else None, temperature=profile.temperature, model=profile.model, think=profile.think_enabled, 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, ), stop_event, turn_ctx, state, ): yield _ev # Record the real prompt-token count for the context we just sent so # the next iteration's midturn/check estimates use real tokens for # the bulk of the context (not the chars//3 heuristic, which # undercounts code-heavy tool output and fires compression too late). # len(session.context) here is exactly what was sent — the new # assistant/tool messages are appended only below. self._compressor.record_real_baseline(len(session.context), state.context_tokens) # Stopped mid-stream — save partial response and exit if stop_event and stop_event.is_set(): if state.accumulated_text: session.messages.append( Message( role="assistant", content=state.accumulated_text, created_at=datetime.now(timezone.utc), is_context=False, ) ) await self._sessions.save(session) yield StreamStopped() return turn_tool_calls = state.turn_tool_calls if not turn_tool_calls: # Final response — text already streamed above. The base message # is built without metrics so it is shared by the intercept and the # genuine-final paths; the model must SEE its own bare text next # iteration when we intercept, so it goes into session.context now. _elapsed = round(time.monotonic() - turn_ctx.turn_start, 1) # Net tokens = accumulated across all iterations + subagent tokens _net_tokens = turn_ctx.turn_tokens + turn_ctx.subagent_tokens assistant_msg = Message( role="assistant", content=state.accumulated_text or None, thinking=state.accumulated_thinking or None, created_at=datetime.now(timezone.utc), ) session.messages.append(assistant_msg) session.context.append(assistant_msg) session.context_token_count = state.context_tokens or 0 # Final-turn intercept: the model stopped with bare text, but the # anti-stall/adaptive nudges never fired (they live inside the # tool-loop; a bare-text turn closes the run before post_turn). # If the todo still has open steps and we haven't hit the limit, # queue a recovery nudge and continue the loop instead of closing. from navi.tools.todo import has_open_steps if ( profile.final_intercept_enabled and turn_ctx.final_interceptions < profile.final_intercept_limit and await has_open_steps(session_id) ): turn_ctx.final_interceptions += 1 pending_final_nudge = Message( role="system", content=_FINAL_INTERCEPT_NUDGES[ min(turn_ctx.final_interceptions, len(_FINAL_INTERCEPT_NUDGES)) - 1 ], ) await self._sessions.save(session) continue # No StreamEnd, no workers — the run continues. # Genuine final: stamp metrics on the already-appended message # (so history carries them) and close the run. assistant_msg.elapsed_seconds = _elapsed assistant_msg.tool_call_count = turn_ctx.tool_call_count if turn_ctx.tool_call_count else None assistant_msg.token_count = _net_tokens if _net_tokens else None await self._sessions.save(session) yield StreamEnd( full_content=state.accumulated_text, context_tokens=state.context_tokens, max_context_tokens=settings.ollama_num_ctx, elapsed_seconds=_elapsed, tool_call_count=turn_ctx.tool_call_count, token_count=_net_tokens if _net_tokens else None, ) for event in await self._run_workers( session, llm, profile.model, state.context_tokens, profile=profile ): yield event return # Tool calls turn — record to session and execute assistant_msg = Message( role="assistant", content=state.accumulated_text or None, thinking=state.accumulated_thinking or None, tool_calls=turn_tool_calls, ) session.messages.append(assistant_msg) session.context.append(assistant_msg) # Persist the assistant's tool-call decision before executing tools, so # a crash / CancelledError mid-tool-execution doesn't lose it. await self._sessions.save(session) tool_ctx = ToolContext( session_id=session_id, event_sink=None, # set per-tool inside _execute_tools_with_sink stop_event=stop_event, model=profile.model, user_id=session.user_id, user_role=current_user_role.get(), user_info=current_user_info.get(), cwd=_cwd_var.get(), ) # Expose a per-iteration PlanRunner so the `plan` tool can run the # planner over the live session. Constructed inside the loop (not # once before) using this iteration's profile/llm/tool_schemas so a # post-switch_profile plan call uses the reloaded profile. reset on # exit so the runner never leaks beyond the tool-execution window. _plan_runner = PlanRunner( self._planning, session, profile, llm, mem, tool_schemas ) _plan_token = current_plan_runner.set(_plan_runner) try: async for _ev in self._execute_tools_with_sink( turn_tool_calls, tools, turn_ctx, session, stop_event, tool_ctx ): yield _ev finally: current_plan_runner.reset(_plan_token) # The todo tool may have been called this turn (set/update/clear). # Re-read and push the latest state to the UI's side panel. yield await self._todo_updated(session) # 6. Cooperative stop: check after tool execution before next LLM call if stop_event and stop_event.is_set(): await self._sessions.save(session) # Reset context vars before returning so they don't leak. _cwd_var.reset(_cwd_token) _sid_var.reset(_sid_token) yield StreamStopped() return # Update anti-stall counters await anti_stall.post_turn(session_id, turn_tool_calls) # 7. If switch_profile was called this iteration, reload profile + tools. # switch_profile updates the DB but run_stream() holds a local session # object — without this check the final save would overwrite the change # and the next LLM call would still use the old tool schemas. fresh = await self._sessions.get(session_id) if fresh and fresh.profile_id != session.profile_id: session.profile_id = fresh.profile_id profile = self._profiles.get(session.profile_id) self._set_active_profile(profile) tools = self._tool_list(profile.get_agent_tools()) tool_schemas = [t.schema() for t in tools] llm = self._get_backend(profile.llm_backend) log.info( "agent.profile_reloaded", session_id=session_id, new_profile=session.profile_id, ) await self._sessions.save(session) # Reset cwd ContextVar so it does not leak into other calls. _cwd_var.reset(_cwd_token) _sid_var.reset(_sid_token) raise MaxIterationsReached(profile.max_iterations) # ------------------------------------------------------------------ # Internal helpers # ------------------------------------------------------------------ async def _run_workers( self, session, llm: LLMBackend, model: str, context_tokens: int | None, profile: "AgentProfile | None" = None, ) -> list[AgentEvent]: """Run all workers sequentially; collect their events.""" from navi.workers.base import WorkerContext ctx = WorkerContext( session_id=session.id, context_tokens=context_tokens, max_context_tokens=settings.ollama_num_ctx, llm=llm, model=model, temperature=settings.context_summary_temperature, session_store=self._sessions, profile=profile, ) events: list[AgentEvent] = [] for worker in self._workers: try: result = await worker.run(session, ctx) events.extend(result.events) except Exception: log.warning("agent.worker_failed", worker=type(worker).__name__, exc_info=True) return events def _tool_list( self, scope: "navi.profiles.base.ToolScopeConfig", # noqa: F821 ) -> list[Tool]: return build_tool_list(scope.native, scope.mcp, self._tools, self._mcp_manager) def _get_backend(self, backend_key: str) -> LLMBackend: return self._backends.get(backend_key) async def _todo_updated(self, session) -> TodoUpdated: """Build a TodoUpdated event from the session's current todo state. Reads via explicit (user_id, session_id) so it works regardless of the ContextVar state at the call site. Sub-agent todos are isolated in their own KV row and never reach here (the parent's session.id is the parent's). """ from navi.tools.todo import load_tasks_for tasks = await load_tasks_for(session.id, session.user_id) return TodoUpdated(session_id=session.id, tasks=tasks) async def _compression_events_preturn(self, session, llm, profile, session_id): if ( settings.context_compression_enabled and len(session.context) > 2 and should_compress( self._compressor.estimate_context_tokens(session.context), settings.ollama_num_ctx, settings.context_compression_threshold, ) ): # Confirm there is actually something to compress before announcing # it. The token gate can fire for a context that the partition cannot # shrink (e.g. a few very large messages in a single turn — handled # instead by per-message view truncation in context_builder.build). # Emitting CompressionStarted without a following ContextCompressed # would show a misleading "compression" status with no work done. if not self._compressor.would_compress( session.context, keep_recent=settings.context_keep_recent, threshold=settings.context_compression_threshold, max_context_tokens=settings.ollama_num_ctx, ): return yield CompressionStarted( context_tokens=self._compressor.estimate_context_tokens(session.context), max_context_tokens=settings.ollama_num_ctx, ) event = await self._compressor.compress_and_save_session( session=session, session_store=self._sessions, llm=llm, model=profile.model, temperature=settings.context_summary_temperature, session_id=session_id, reason="preturn", keep_recent=settings.context_keep_recent, max_tokens=settings.context_summary_max_tokens, ) if event: yield event async def _compression_events_midturn( self, session, llm, profile, session_id, iteration, ctx_injections, mem ): if settings.context_compression_enabled and iteration > 0: preflight_ctx = self._ctx_builder.build( session.context, profile, mem, extra_system=ctx_injections, session_id=session_id, ) estimated_tokens = self._compressor.real_baseline_estimate( session.context, preflight_ctx ) if should_compress( estimated_tokens, settings.ollama_num_ctx, settings.context_compression_threshold, ): # See _compression_events_preturn: only announce compression when # the partition (or token-budget fallback) can actually shrink the # stored context. The per-message view truncation in build() # already caps single huge messages for the LLM, so a no-op here # no longer risks an overflow — but announcing it would still lie # to the UI. if not self._compressor.would_compress( session.context, keep_recent=settings.context_keep_recent, keep_recent_messages=max(12, settings.context_keep_recent * 2), threshold=settings.context_compression_threshold, max_context_tokens=settings.ollama_num_ctx, ): return yield CompressionStarted( context_tokens=estimated_tokens, max_context_tokens=settings.ollama_num_ctx, ) event = await self._compressor.compress_and_save_session( session=session, session_store=self._sessions, llm=llm, model=profile.model, temperature=settings.context_summary_temperature, session_id=session_id, reason="midturn", keep_recent=settings.context_keep_recent, max_tokens=settings.context_summary_max_tokens, keep_recent_messages=max(12, settings.context_keep_recent * 2), ) if event: yield event async def _consume_stream( self, stream_gen, stop_event, turn_ctx: AgentTurnContext, state: StreamState ): async for chunk in stream_gen: if stop_event and stop_event.is_set(): if state.thinking_active: yield ThinkingEnd() break if chunk.prompt_tokens is not None: state.context_tokens = chunk.prompt_tokens if chunk.completion_tokens is not None: turn_ctx.turn_tokens += chunk.completion_tokens # Surface the model that actually served this turn to the UI. The # backend stamps it on the first successful chunk (fallback may pick # a different model than the profile's first priority). Re-emit only # when it changes across iterations within the same turn. if chunk.model and chunk.model != turn_ctx.resolved_model: turn_ctx.resolved_model = chunk.model yield ModelInfo(model=chunk.model) if chunk.thinking: state.accumulated_thinking += chunk.thinking if not state.thinking_active: state.thinking_active = True yield ThinkingDelta(delta=chunk.thinking) if chunk.delta: if state.thinking_active: state.thinking_active = False yield ThinkingEnd() state.accumulated_text += chunk.delta yield TextDelta(delta=chunk.delta) if chunk.tool_calls: state.turn_tool_calls = chunk.tool_calls if chunk.finish_reason and state.thinking_active: state.thinking_active = False yield ThinkingEnd() async def _execute_tools_with_sink( self, turn_tool_calls, tools, turn_ctx: AgentTurnContext, session, stop_event, tool_ctx=None ): """Execute tool calls with cooperative stop support. Polls *stop_event* every second while draining the event sink so the Stop button works even during long-running tools (terminal, SSH, web search, sub-agent spawn). """ tool_map = {t.name: t for t in tools} for tc in turn_tool_calls: from navi.tools.todo import started_metadata_for_call yield ToolStarted( tool_name=tc.name, arguments=tc.arguments, tool_call_id=tc.id, metadata=await started_metadata_for_call(tc, tool_ctx), ) sink: asyncio.Queue = asyncio.Queue() sink_token = current_event_sink.set(sink) result_holder: list = [] async def _run_with_sentinel(_tc=tc, _holder=result_holder, _sink=sink): try: _holder.append( await self._tool_executor._run_single_tool(_tc, tool_map, ctx=tool_ctx) ) except Exception as exc: _holder.append(exc) finally: await _sink.put(_TOOL_DONE) tool_task = asyncio.create_task(_run_with_sentinel()) current_event_sink.reset(sink_token) try: # Poll sink with 1s timeout so stop_event is checked during long tools stopped = False while True: try: item = await asyncio.wait_for(sink.get(), timeout=1.0) except asyncio.TimeoutError: if stop_event and stop_event.is_set(): stopped = True if not tool_task.done(): tool_task.cancel() break continue if item is _TOOL_DONE: break if isinstance(item, SubagentComplete): turn_ctx.subagent_tokens += item.token_count turn_ctx.tool_call_count += item.tool_call_count elif isinstance(item, AIHelperTokensUsed): turn_ctx.subagent_tokens += item.completion_tokens else: yield item if stopped: # Tool was interrupted by user — record a synthetic cancellation result log.info("agent.tool_stopped", tool=tc.name) yield ToolEvent( tool_name=tc.name, arguments=tc.arguments, result="Tool execution was stopped by the user.", success=False, tool_call_id=tc.id, ) session.messages.append( Message( role="tool", content="Tool execution was stopped by the user.", tool_call_id=tc.id, name=tc.name, metadata={}, is_context=False, ) ) await self._sessions.save(session) return r = ( result_holder[0] if result_holder else RuntimeError("tool task produced no result") ) if isinstance(r, Exception): if isinstance(r, (LLMBackendError, LLMConnectionError)): raise r log.warning("agent.tool_exception", tool=tc.name, error=str(r)) tool_event = ToolEvent( tool_name=tc.name, arguments=tc.arguments, result=f"Error: {r}", success=False, tool_call_id=tc.id, ) msg = Message( role="tool", content=f"Error: {r}", tool_call_id=tc.id, name=tc.name, metadata={}, ) image_msg = None else: tool_event, msg, image_msg = r turn_ctx.tool_call_count += 1 yield tool_event session.messages.append(msg) session.context.append(msg) if image_msg: session.messages.append(image_msg) session.context.append(image_msg) # Persist each tool result as it completes, so a crash / # CancelledError (server restart) between tool calls cannot lose # already-finished work — only the in-flight single tool is at risk. await self._sessions.save(session) finally: if not tool_task.done(): tool_task.cancel() try: await tool_task except Exception: pass