diff --git a/navi/core/agent.py b/navi/core/agent.py deleted file mode 100644 index 8fe1f14..0000000 --- a/navi/core/agent.py +++ /dev/null @@ -1,1608 +0,0 @@ -""" -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 time -from datetime import datetime, timezone -from pathlib import Path -from typing import TYPE_CHECKING, AsyncGenerator - -import structlog - -from navi.config import settings -from navi.exceptions import ( - ContextTooLargeError, - LLMBackendError, - LLMConnectionError, - MaxIterationsReached, - NothingToCompactError, - SessionNotFound, -) -from navi.llm.base import LLMBackend, Message -from navi.session_files import save_images as save_session_images -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 .image_tokens import context_image_tokens, estimate_image_tokens -from .planning import PlanningEngine -from navi.tools.plan import PlanRunner -from .stream_guard import _iter_stream_guarded -from .subagent_runner import SubAgentOutcome, SubAgentRunner -from .tool_utils import build_tool_list -from .events import ( - AgentEvent, - AIHelperTokensUsed, - CompressionStarted, - ModelInfo, - ProfileSwitched, - 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() - - -def _inline_images_notice(total: int, in_context: int, paths: list[str]) -> str: - """The line the model reads next to the images it was sent. - - `paths` is index-aligned with the images and holds an empty string for one that - could not be stored. The notice always says the files exist, because a model told - the session directory exists and given no path will go looking for one — that hunt - is exactly what this line is here to prevent. - """ - noun = "image" if total == 1 else "images" - if in_context < total: - notice = ( - f"[{total} {noun} attached inline — {in_context} are in your context, " - "the rest were not loaded" - ) - dropped = [path for path in paths[in_context:] if path] - if dropped: - notice += f"; open them with image_view: {', '.join(dropped)}" - else: - notice += " and could not be saved to the session directory" - else: - notice = ( - f"[{total} {noun} attached inline — already in your context, " - "no extra loading needed" - ) - stored = [path for path in paths if path] - if stored: - notice += f". On disk: {', '.join(stored)}" - return notice + "]" - - -class _TaggedSink: - """Routes one tool's live events into the shared batch queue, tagged by index. - - Mirrors the asyncio.Queue surface the sinks actually use (``put`` / - ``put_nowait``) so it is a drop-in replacement for a per-tool queue. - """ - - def __init__(self, queue: asyncio.Queue, idx: int) -> None: - self._queue = queue - self._idx = idx - - async def put(self, item) -> None: - await self._queue.put((self._idx, item)) - - def put_nowait(self, item) -> None: - self._queue.put_nowait((self._idx, item)) - - -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) - - -async def _protocol_progress_message(session_id: str) -> "Message | None": - """System reminder for an active protocol run, or None. - - The run was started in some earlier turn; compression may drop that turn and - the todo lines no longer say which of them the user pre-agreed to. This - reminder keeps the commitment — and the way to close it — in view. - """ - from navi.tools.protocol import get_protocol_message - - return await get_protocol_message(session_id) - - -# Final-turn intercept nudges — injected when the model ends a turn with bare -# text (no tool call) while the todo still has open steps (or a protocol run is -# still open), 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.", -) - -# Same intercept, different move: the todo shows no open work, but a protocol -# run is still `running` because it was never closed. -_PROTOCOL_FINAL_NUDGE = ( - "[Final-turn check] You ended the turn with text only while a protocol run is still " - "open. Close it now: protocol(op='finish', report=...) if its steps are done, or " - "protocol(op='cancel'/'fail', reason=...) if not. Do not leave a run running." -) - - -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, - memory_search=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._memory_search = memory_search - self._cp_registry = cp_registry - self._mcp_manager = mcp_manager - self._ctx_builder = ContextBuilder( - profile_registry=profile_registry, - memory_store=memory_store, - memory_search=memory_search, - 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 - # ------------------------------------------------------------------ - - def invalidate_prompt_cache(self, profile_id: str | None = None) -> None: - """Drop cached system prompts so a profile change reaches the next turn. - - The cache lives on this agent's ContextBuilder, and the agent handed out - by the container is process-wide — so an admin toggling a profile's - availability must clear it, or every other prompt keeps advertising the - profile list as it was. - """ - self._ctx_builder.invalidate_system_prompt_cache(profile_id) - - 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 _drain_task_notes(self, session_id: str, session) -> None: - """Move finished-background-task notes into the session context. - - Called once at the start of run_stream(), before the first LLM call: - the note becomes part of the LLM-visible context (and the compression - window) but is never displayed — clients learned about completions via - the live task_update event. Only the foreground run_stream mutates the - session, so there is no race with background jobs. - """ - from navi.core import task_notes - - try: - note = await task_notes.drain(session_id) - except Exception: - log.exception("agent.task_notes_drain_failed", session_id=session_id) - return - if not note: - return - session.context.append( - Message( - role="system", - content=note, - metadata={"source": "task_note"}, - is_display=False, - ) - ) - await self._sessions.save(session) - log.info("agent.task_notes_drained", session_id=session_id) - - 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, - ) -> SubAgentOutcome: - """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.") - - window = self._window_for(profile) - yield CompressionStarted( - # Real-baseline estimate when available (see _compression_events_preturn); - # the calibrated estimate is only the fallback before the first LLM - # call of a session. - context_tokens=self._compressor.real_baseline_estimate( - session.context, session.context, profile.model - ), - max_context_tokens=window, - ) - 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", - memory_store=self._memory, - 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), - window=window, - ) - 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) - _scope_token, _keyless_token = await self._apply_mcp_scope(session) - tools = self._tool_list(profile.get_agent_tools()) - tool_schemas = [t.schema() for t in tools] - llm = self._get_backend(profile.llm_backend) - # The profile these four are bound to. The post-turn reload must compare - # against this, not against session.profile_id: after a switch_profile the - # two can be equal again (the store rewrote the row, or mutated this very - # object) while the tool set above is still the old profile's. - bound_profile_id = profile.id - - 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_route as _route_var, - current_profile_id as _profile_var, - current_working_directory as _cwd_var, - current_allowed_mcp_servers as _mcp_scope_var, - current_keyless_mcp_servers as _keyless_var, - ) - - _sid_token = _sid_var.set(session_id) - _model_var.set(profile.model) - _route_var.set(getattr(profile, "llm_route", None)) - _profile_var.set(profile.id) - _cwd_token = _cwd_var.set(cwd) - - # Start the turn on the route's primary entry when it is healthy - # again: the background probe is what notices a recovered host, and - # this is where that knowledge becomes visible to the user — the next - # turn, never a running one. A turn that starts while the cloud is - # down begins on the local entry and stays there until it ends. - begin_turn = getattr(llm, "begin_turn", None) - if callable(begin_turn): - begin_turn(getattr(profile, "llm_route", None)) - - # 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). - # The incoming message is priced here because the gate runs before it - # is appended — an attachment is the one part of the next prompt this - # context cannot yet account for. - pending = Message(role="user", content=user_message, images=images or None) - pending_tokens = self._compressor.estimate_tokens_delta([pending], profile.model) - async for _ev in self._compression_events_preturn( - session, llm, profile, session_id, pending_tokens=pending_tokens - ): - 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) - - # Inline chat images land in the session directory too, and the paths go into - # the context next to the images themselves. A pasted image used to exist in - # the transcript only, so an agent told the session directory exists went - # looking there for the "attached images", found nothing, and spent turns on - # fruitless `find` calls before giving up. - saved_paths: list[str] = [] - if images: - try: - names = save_session_images(session_id, images) - except OSError: - log.exception("agent.image_save_failed", session_id=session_id) - names = [] - saved_paths = [ - str(Path(settings.session_files_dir) / session_id / name) if name else "" - for name in names - ] - - # Image token budgeting: fit as many images as possible into the LLM context. - # Images that do not fit stay reachable through the session directory, so the - # notice has to name them. Each image is priced by its own size: a flat 500 - # admitted twice as many screenshots as the model can take (a 1024x1024 image - # is ~1026 tokens on qwen3.6 — see navi/core/image_tokens.py). - images_for_context = images - context_content = user_message - if images: - current_tokens = self._compressor.estimate_tokens( - session.context, profile.model - ) - available_tokens = int(self._window_for(profile) * 0.8) - current_tokens - fitted: list[str] = [] - for img in images: - cost = estimate_image_tokens(img) - if cost > available_tokens: - break - fitted.append(img) - available_tokens -= cost - if len(fitted) < len(images): - images_for_context = fitted - context_content += "\n\n" + _inline_images_notice( - len(images), len(images_for_context or []), saved_paths - ) - - 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"), - ) - ) - # Raw excerpts compression dropped. Deduped across turns through the - # session's ephemeral metadata: the same chunk should not ride along in - # every prompt of a long session just because the topic persists. - if session.session_metadata is None: - session.session_metadata = {} - # A list, not a set: session_metadata is JSON-encoded into the - # sessions row on every save, and a set is not serializable. - injected_chunks = session.session_metadata.setdefault( - "injected_chunk_ids", [] - ) - mem_recall_task = asyncio.create_task( - self._ctx_builder._memory_recall_msg( - session, - user_message, - user_id=session.user_id, - injected_ids=injected_chunks, - ) - ) - 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) - mem_recall = await mem_recall_task - if mem_recall: - ctx_injections.append(mem_recall) - - # Drain background-task completion notes into the session context - # (persisted so compression keeps them; not displayed — the client - # already saw the live task_update event). - await self._drain_task_notes(session_id, session) - - 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) - - protocol_msg = await _protocol_progress_message(session_id) - if protocol_msg: - built_ctx.append(protocol_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, - model=profile.model, - window=self._window_for(profile), - ) - 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) - # This path was never exercised (the old chars//3 check never - # fired), so it kept an invalid StreamEnd(content=...) and died - # with a TypeError instead of reporting the overflow. - yield StreamEnd( - full_content=error_text, - max_context_tokens=self._window_for(profile), - ) - 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, - # The window the compressor accounts against must be the - # window the local model is actually loaded with. - num_ctx=self._window_for(profile), - ), - 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. - # `sent_tokens` lets the compressor notice a prompt the backend cut - # down instead of caching the cut size as the truth (Ollama keeps - # roughly half the window and reports only that). - self._compressor.record_real_baseline( - len(session.context), - state.context_tokens, - chars=self._compressor.context_chars(built_ctx), - model=turn_ctx.resolved_model or profile.model, - sent_tokens=self._compressor.real_baseline_estimate( - session.context, built_ctx, turn_ctx.resolved_model or profile.model - ), - image_tokens=context_image_tokens(built_ctx), - ) - - # 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) - if self._compressor.context_truncated: - # A truncated prompt is reported as what the backend - # evaluated — roughly half the window — while the context - # that produced it was already over the window. Showing - # that number would drop the bar to ~50% at the exact - # moment the session is full; keep the honest estimate. - session.context_token_count = max( - state.context_tokens or 0, - self._compressor.real_baseline_estimate( - session.context, session.context, profile.model - ), - ) - else: - 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 — or a protocol run is open - # even though its steps are all closed — and we haven't hit the - # limit, queue a recovery nudge and continue the loop instead of - # closing. - from navi.tools.protocol import has_active_run - from navi.tools.todo import has_open_steps - - may_intercept = ( - profile.final_intercept_enabled - and turn_ctx.final_interceptions < profile.final_intercept_limit - ) - todo_open = await has_open_steps(session_id) if may_intercept else False - protocol_open = await has_active_run(session_id) if may_intercept else False - if may_intercept and (todo_open or protocol_open): - turn_ctx.final_interceptions += 1 - if protocol_open and not todo_open: - # A run whose steps are all closed needs a different - # closing move than "there is still work to do". - nudge = _PROTOCOL_FINAL_NUDGE - else: - nudge = _FINAL_INTERCEPT_NUDGES[ - min(turn_ctx.final_interceptions, len(_FINAL_INTERCEPT_NUDGES)) - 1 - ] - pending_final_nudge = Message(role="system", content=nudge) - 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=self._window_for(profile), - 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, - profile_id=profile.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(), - ) - # Parallel batch gate: profile override wins, else the global flag. - if turn_ctx.parallel_tool_calls is False: - if profile.parallel_tool_calls is None: - from navi.config import settings as _settings - - turn_ctx.parallel_tool_calls = _settings.parallel_tool_calls - else: - turn_ctx.parallel_tool_calls = profile.parallel_tool_calls - # 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) - _mcp_scope_var.reset(_scope_token) - _keyless_var.reset(_keyless_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. - # Compare against bound_profile_id, never against session.profile_id: - # set_profile() either rewrites the row (and this run's own save() - # then writes its stale value back over it) or mutates the very - # Session object the run holds — in both cases session.profile_id - # and the stored row agree again, the check stayed false and the - # turn kept calling the old profile's tools. The bindings at the - # top of the run are the only truth about what tools are live. - fresh = await self._sessions.get(session_id) - if fresh and fresh.profile_id != bound_profile_id: - session.profile_id = fresh.profile_id - profile = self._profiles.get(session.profile_id) - bound_profile_id = profile.id - self._set_active_profile(profile) - # The rest of this run executes as the new profile — tools that - # resolve a default profile must not keep reporting the old one. - _profile_var.set(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) - 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) - _mcp_scope_var.reset(_scope_token) - _keyless_var.reset(_keyless_token) - raise MaxIterationsReached(profile.max_iterations) - - # ------------------------------------------------------------------ - # Internal helpers - # ------------------------------------------------------------------ - - async def _apply_mcp_scope(self, session): - """Restrict this run's MCP tool list to the servers its user may use. - - A BYOK-gated server with no personal key must not appear in the list at - all: the manager refuses the call anyway, and a tool the model can see - but never call is read as a transient fault and retried. Admins (and the - no-auth local user, whose role is ``admin``) get ``None`` — the shared - credential in the config is theirs, so nothing is filtered. - - Also records *which* servers were withheld for want of a key. The reason - is not stored anywhere else and is not derivable by the caller: it is the - complement of what ``visible_servers`` kept, so it is computed from that - same value rather than read a second time (a second read of the key store - could disagree with the first). - - Returns the pair of ContextVar tokens; callers reset them where they - already reset the session and cwd vars. - """ - from navi.tools._internal.base import ( - current_allowed_mcp_servers, - current_keyless_mcp_servers, - current_user_id, - current_user_role, - ) - - if self._mcp_manager is None: - allowed, keyless = None, frozenset() - else: - try: - allowed = await self._mcp_manager.visible_servers( - current_user_role.get(), current_user_id.get() or session.user_id - ) - # Withheld = gated servers the run did not get to keep. None means - # "no restriction", so nothing was withheld. - keyless = ( - frozenset() - if allowed is None - else frozenset(self._mcp_manager.byok_servers() - allowed) - ) - except Exception: - # Fail closed: an unanswerable visibility question must not fall - # back to "everything is available". Every gated server then counts - # as withheld, which matches the empty tool list the model gets. - log.warning("agent.mcp_scope_failed", exc_info=True) - allowed = frozenset() - keyless = frozenset(self._mcp_manager.byok_servers()) - - return ( - current_allowed_mcp_servers.set(allowed), - current_keyless_mcp_servers.set(keyless), - ) - - 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=self._window_for(profile), - llm=llm, - model=model, - temperature=settings.context_summary_temperature, - session_store=self._sessions, - profile=profile, - # The backend cut the last prompt down (record_real_baseline saw it). - # Its token count is below the trigger, but the context it came from - # is over the window — compress now instead of waiting for a - # threshold that the truncated count will never reach. - context_truncated=self._compressor.pop_context_truncated(), - memory_store=self._memory, - ) - 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 _tool_list_for_profile(self, profile_id: str) -> "list[Tool] | None": - """Resolve another profile's tools by id; None when it is unknown. - - Used mid-batch right after a switch_profile call, where the running - turn's own ``profile`` binding is still the one it started with. - """ - try: - profile = self._profiles.get(profile_id) - except Exception: - log.warning("agent.switch_profile_unknown", profile_id=profile_id) - return None - return self._tool_list(profile.get_agent_tools()) - - def _get_backend(self, backend_key: str) -> LLMBackend: - return self._backends.get(backend_key) - - def _window_for(self, profile: "AgentProfile | None") -> int: - """Context window this profile's model is accounted against. - - One global OLLAMA_NUM_CTX used to stand in for every model — a local - 65k model and a cloud model fed 115k prompts were held to the same - number. A profile may now declare its own ``context_window``. - - The route's active entry wins over both: when the cloud is down and a - 48k local model is answering, compressing against a cloud-sized window - would send it prompts it cannot hold. - """ - try: - backend = self._get_backend(getattr(profile, "llm_backend", "ollama")) - except KeyError: - backend = None - window_for_route = getattr(backend, "active_window", None) - if callable(window_for_route): - active = window_for_route(getattr(profile, "llm_route", None)) - if active: - return active - return ( - getattr(profile, "context_window", None) - or settings.ollama_num_ctx - ) - - 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, pending_tokens: int = 0 - ): - # real_baseline_estimate, not a character heuristic: the baseline is - # valid here (the context only grew since the last LLM call), and a - # character count undercounts code-heavy tool output — firing - # compression too late is exactly what the baseline machinery was - # built to fix. The midturn gate below does the same. - estimated_tokens = self._compressor.real_baseline_estimate( - session.context, session.context, profile.model - ) - window = self._window_for(profile) - # `pending_tokens` prices the user message that has not been appended - # yet — this gate runs before it is. Without it, a message carrying - # images could overshoot the window in one step: the context was under - # the threshold a moment ago and the images were the only thing over. - if ( - settings.context_compression_enabled - and len(session.context) > 2 - and should_compress( - estimated_tokens + pending_tokens, - window, - 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=window, - # Judge the fallback on the same total that fired the gate. - current_tokens=estimated_tokens + pending_tokens, - model=profile.model, - ): - return - yield CompressionStarted( - # The context being compressed does not include the pending - # message yet, so report the stored context's own size. - context_tokens=estimated_tokens, - max_context_tokens=window, - ) - 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", - memory_store=self._memory, - keep_recent=settings.context_keep_recent, - max_tokens=settings.context_summary_max_tokens, - window=window, - ) - 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, profile.model - ) - window = self._window_for(profile) - if should_compress( - estimated_tokens, - window, - 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=window, - current_tokens=estimated_tokens, - model=profile.model, - ): - return - yield CompressionStarted( - context_tokens=estimated_tokens, - max_context_tokens=window, - ) - 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", - memory_store=self._memory, - keep_recent=settings.context_keep_recent, - max_tokens=settings.context_summary_max_tokens, - keep_recent_messages=max(12, settings.context_keep_recent * 2), - window=window, - ) - 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, - target=getattr(chunk, "target", None), - fallback=bool(getattr(chunk, "is_fallback", False)), - ) - 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, sequentially or as a parallel batch. - - The mode is decided per-turn by ``turn_ctx.parallel_tool_calls`` - (profile override ? global setting, resolved at tool_ctx build). - """ - # A profile switch replaces the toolset mid-turn, so it must not share a - # batch: every neighbour would still be dispatched against the old - # tool_map and die with "tool 'X' not found" (seen with reload_tools called - # just after a switch). Run it alone first, then re-resolve the tools - # it switched to for the rest of the batch. The turn's own ``profile`` - # binding stays old on purpose — the end-of-iteration reload in - # run_stream() is what rebinds profile/llm/schemas. - switching = next((tc for tc in turn_tool_calls if tc.name == "switch_profile"), None) - if switching is not None and len(turn_tool_calls) > 1: - rest = [tc for tc in turn_tool_calls if tc is not switching] - switched_to = None - async for ev in self._execute_tools_sequential( - [switching], tools, turn_ctx, session, stop_event, tool_ctx - ): - if isinstance(ev, ProfileSwitched): - switched_to = ev.profile_id - yield ev - if switched_to: - new_tools = self._tool_list_for_profile(switched_to) - if new_tools is not None: - tools = new_tools - turn_tool_calls = rest - - if turn_tool_calls and turn_ctx.parallel_tool_calls and len(turn_tool_calls) > 1: - async for ev in self._execute_tools_parallel( - turn_tool_calls, tools, turn_ctx, session, stop_event, tool_ctx - ): - yield ev - else: - async for ev in self._execute_tools_sequential( - turn_tool_calls, tools, turn_ctx, session, stop_event, tool_ctx - ): - yield ev - - async def _execute_tools_parallel( - self, turn_tool_calls, tools, turn_ctx: AgentTurnContext, session, stop_event, tool_ctx=None - ): - """Execute an independent batch of tool calls concurrently. - - ToolStarted is emitted for every call up-front (in call order) so the - UI shows the whole batch immediately. Live events from all tools merge - through one tagged queue; results and tool messages are appended in - call order after the batch settles, with a single session save. - - Only batch independent calls — two actions over the same terminal in - one batch race each other. - """ - from navi.tools.todo import started_metadata_for_call - - tool_map = {t.name: t for t in tools} - for tc in turn_tool_calls: - yield ToolStarted( - tool_name=tc.name, - arguments=tc.arguments, - tool_call_id=tc.id, - metadata=await started_metadata_for_call(tc, tool_ctx), - ) - - shared: asyncio.Queue = asyncio.Queue() - holders: list[list] = [[] for _ in turn_tool_calls] - tasks: list[asyncio.Task] = [] - - for idx, tc in enumerate(turn_tool_calls): - tagged = _TaggedSink(shared, idx) - sink_token = current_event_sink.set(tagged) - - async def _run_with_sentinel(_tc=tc, _idx=idx, _holder=holders[idx], _sink=tagged): - 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) - - tasks.append(asyncio.create_task(_run_with_sentinel())) - current_event_sink.reset(sink_token) - - try: - pending = set(range(len(tasks))) - stopped = False - while pending: - try: - item = await asyncio.wait_for(shared.get(), timeout=1.0) - except asyncio.TimeoutError: - if stop_event and stop_event.is_set(): - stopped = True - break - continue - - _idx, value = item - if value is _TOOL_DONE: - pending.discard(_idx) - continue - if isinstance(value, SubagentComplete): - turn_ctx.subagent_tokens += value.token_count - turn_ctx.tool_call_count += value.tool_call_count - elif isinstance(value, AIHelperTokensUsed): - turn_ctx.subagent_tokens += value.completion_tokens - else: - yield value - - if stopped: - # Stop must actually stop work: cancel in-flight tools the same - # way the sequential path does (docs/tasks.md: stop mid-batch - # cancels every unfinished tool). Calls that already finished - # keep their real results — only unfinished ones get the - # synthetic stopped note below; erasing a completed - # backgrounded call (task_id) would orphan the task in the - # client UI. - log.info( - "agent.batch_stopped", - tools=[tc.name for tc in turn_tool_calls], - ) - for t in tasks: - if not t.done(): - t.cancel() - await asyncio.gather(*tasks, return_exceptions=True) - # Same reason as in the sequential path: whatever the cancelled - # tools announced before unwinding still belongs to cards the - # client is showing, so it is forwarded rather than dropped. - while True: - try: - _idx, value = shared.get_nowait() - except asyncio.QueueEmpty: - break - if value is _TOOL_DONE: - continue - if isinstance(value, SubagentComplete): - turn_ctx.subagent_tokens += value.token_count - turn_ctx.tool_call_count += value.tool_call_count - elif isinstance(value, AIHelperTokensUsed): - turn_ctx.subagent_tokens += value.completion_tokens - else: - yield value - - # Results and messages in call order, one save for the batch. - # A stopped batch mixes real results (finished calls) with the - # synthetic stopped note (cancelled calls). - for tc, holder in zip(turn_tool_calls, holders): - if not holder: - if stopped: - 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, - ) - ) - continue - r = RuntimeError("tool task produced no result") - else: - r = holder[0] - 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) - await self._sessions.save(session) - finally: - # Cancel whatever is still running (generator closed mid-batch or - # an outer cancellation of the run), then gather to contain each - # cancelled task — awaiting the results directly would re-raise - # CancelledError past the results already produced, while an - # outer cancellation of the run itself still propagates. - for t in tasks: - if not t.done(): - t.cancel() - await asyncio.gather(*tasks, return_exceptions=True) - - async def _execute_tools_sequential( - self, turn_tool_calls, tools, turn_ctx: AgentTurnContext, session, stop_event, tool_ctx=None - ): - """Execute tool calls one at a time 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: - # A cancelled tool may still have announced work the client - # is already showing: a sub-agent emits a closing ToolEvent - # for its in-flight tool from the CancelledError handler. - # Read those before dropping the queue, or the client keeps - # a spinner it can never resolve until the page reloads. - await asyncio.gather(tool_task, return_exceptions=True) - while True: - try: - item = sink.get_nowait() - except asyncio.QueueEmpty: - break - if item is _TOOL_DONE: - continue - 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 - # 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() - # gather(return_exceptions=True) contains the cancelled tool - # task (awaiting it directly would re-raise CancelledError and - # abort the turn before the synthetic stopped results land), - # while an outer cancellation of the run itself propagates. - await asyncio.gather(tool_task, return_exceptions=True) diff --git a/navi/core/agent/__init__.py b/navi/core/agent/__init__.py new file mode 100644 index 0000000..5724302 --- /dev/null +++ b/navi/core/agent/__init__.py @@ -0,0 +1,50 @@ +"""Agent: the tool-calling loop. + +The class is assembled from mixins: `runtime` (turn lifecycle), `streaming` (the +turn loop), `tools` (tool execution), `compression` (context gates) and +`context` (tool lists, backend lookup, context window). The module-level +helpers live in `helpers`. + +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 +""" + +from navi.exceptions import MaxIterationsReached + +from ..agent_run_context import AgentTurnContext, StreamState +from ..stream_guard import _iter_stream_guarded +from .core import Agent +from .helpers import ( + _FINAL_INTERCEPT_NUDGES, + _PROTOCOL_FINAL_NUDGE, + _TOOL_DONE, + _inline_images_notice, + _protocol_progress_message, + _TaggedSink, + _todo_progress_message, +) + +__all__ = [ + "_FINAL_INTERCEPT_NUDGES", + "_PROTOCOL_FINAL_NUDGE", + "_TOOL_DONE", + "Agent", + "AgentTurnContext", + "MaxIterationsReached", + "StreamState", + "_TaggedSink", + "_inline_images_notice", + "_iter_stream_guarded", + "_protocol_progress_message", + "_todo_progress_message", +] diff --git a/navi/core/agent/compression.py b/navi/core/agent/compression.py new file mode 100644 index 0000000..5b97cc8 --- /dev/null +++ b/navi/core/agent/compression.py @@ -0,0 +1,138 @@ +"""Context compression gates around and inside a turn.""" + +from typing import TYPE_CHECKING + +import structlog + +from navi.config import settings + +from ..compressor import should_compress +from ..events import ( + CompressionStarted, +) + + + +log = structlog.get_logger() + + +class AgentCompressionMixin: + async def _compression_events_preturn( + self, session, llm, profile, session_id, pending_tokens: int = 0 + ): + # real_baseline_estimate, not a character heuristic: the baseline is + # valid here (the context only grew since the last LLM call), and a + # character count undercounts code-heavy tool output — firing + # compression too late is exactly what the baseline machinery was + # built to fix. The midturn gate below does the same. + estimated_tokens = self._compressor.real_baseline_estimate( + session.context, session.context, profile.model + ) + window = self._window_for(profile) + # `pending_tokens` prices the user message that has not been appended + # yet — this gate runs before it is. Without it, a message carrying + # images could overshoot the window in one step: the context was under + # the threshold a moment ago and the images were the only thing over. + if ( + settings.context_compression_enabled + and len(session.context) > 2 + and should_compress( + estimated_tokens + pending_tokens, + window, + 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=window, + # Judge the fallback on the same total that fired the gate. + current_tokens=estimated_tokens + pending_tokens, + model=profile.model, + ): + return + yield CompressionStarted( + # The context being compressed does not include the pending + # message yet, so report the stored context's own size. + context_tokens=estimated_tokens, + max_context_tokens=window, + ) + 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", + memory_store=self._memory, + keep_recent=settings.context_keep_recent, + max_tokens=settings.context_summary_max_tokens, + window=window, + ) + 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, profile.model + ) + window = self._window_for(profile) + if should_compress( + estimated_tokens, + window, + 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=window, + current_tokens=estimated_tokens, + model=profile.model, + ): + return + yield CompressionStarted( + context_tokens=estimated_tokens, + max_context_tokens=window, + ) + 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", + memory_store=self._memory, + keep_recent=settings.context_keep_recent, + max_tokens=settings.context_summary_max_tokens, + keep_recent_messages=max(12, settings.context_keep_recent * 2), + window=window, + ) + if event: + yield event diff --git a/navi/core/agent/context.py b/navi/core/agent/context.py new file mode 100644 index 0000000..0d39c0c --- /dev/null +++ b/navi/core/agent/context.py @@ -0,0 +1,68 @@ +"""What the model sees: tool lists for a profile, backend lookup, context window.""" + +from typing import TYPE_CHECKING + +import structlog + +from navi.config import settings +from navi.llm.base import LLMBackend +from navi.tools._internal.base import ( + Tool, +) + +from ..tool_utils import build_tool_list + +if TYPE_CHECKING: + from navi.profiles.base import AgentProfile + + +log = structlog.get_logger() + + +class AgentContextMixin: + 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 _tool_list_for_profile(self, profile_id: str) -> "list[Tool] | None": + """Resolve another profile's tools by id; None when it is unknown. + + Used mid-batch right after a switch_profile call, where the running + turn's own ``profile`` binding is still the one it started with. + """ + try: + profile = self._profiles.get(profile_id) + except Exception: + log.warning("agent.switch_profile_unknown", profile_id=profile_id) + return None + return self._tool_list(profile.get_agent_tools()) + + def _get_backend(self, backend_key: str) -> LLMBackend: + return self._backends.get(backend_key) + + def _window_for(self, profile: "AgentProfile | None") -> int: + """Context window this profile's model is accounted against. + + One global OLLAMA_NUM_CTX used to stand in for every model — a local + 65k model and a cloud model fed 115k prompts were held to the same + number. A profile may now declare its own ``context_window``. + + The route's active entry wins over both: when the cloud is down and a + 48k local model is answering, compressing against a cloud-sized window + would send it prompts it cannot hold. + """ + try: + backend = self._get_backend(getattr(profile, "llm_backend", "ollama")) + except KeyError: + backend = None + window_for_route = getattr(backend, "active_window", None) + if callable(window_for_route): + active = window_for_route(getattr(profile, "llm_route", None)) + if active: + return active + return ( + getattr(profile, "context_window", None) + or settings.ollama_num_ctx + ) diff --git a/navi/core/agent/core.py b/navi/core/agent/core.py new file mode 100644 index 0000000..978c466 --- /dev/null +++ b/navi/core/agent/core.py @@ -0,0 +1,78 @@ +"""The assembled `Agent` class — behaviour lives in the mixins.""" + +from typing import TYPE_CHECKING + +import structlog + +from ..compressor import ContextCompressor +from ..context_builder import ContextBuilder +from ..planning import PlanningEngine +from ..registry import BackendRegistry, ProfileRegistry, ToolRegistry +from ..session import SessionStore +from ..subagent_runner import SubAgentRunner +from ..tool_executor import ToolExecutor + +if TYPE_CHECKING: + from navi.context_providers._loader import ContextProviderRegistry + from navi.memory.store import MemoryStore + from navi.workers.base import Worker + + +log = structlog.get_logger() + +from .compression import AgentCompressionMixin +from .context import AgentContextMixin +from .runtime import AgentRuntimeMixin +from .streaming import AgentStreamMixin +from .tools import AgentToolsMixin + + +class Agent( + AgentRuntimeMixin, + AgentStreamMixin, + AgentToolsMixin, + AgentCompressionMixin, + AgentContextMixin, +): + 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, + memory_search=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._memory_search = memory_search + self._cp_registry = cp_registry + self._mcp_manager = mcp_manager + self._ctx_builder = ContextBuilder( + profile_registry=profile_registry, + memory_store=memory_store, + memory_search=memory_search, + 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, + ) diff --git a/navi/core/agent/helpers.py b/navi/core/agent/helpers.py new file mode 100644 index 0000000..a24cd50 --- /dev/null +++ b/navi/core/agent/helpers.py @@ -0,0 +1,103 @@ +"""Module-level helpers, constants and the event-sink wrapper the agent uses.""" + +import asyncio + +from navi.llm.base import Message + +# Sentinel: placed in the event sink by the tool wrapper to signal completion. +_TOOL_DONE = object() + + +def _inline_images_notice(total: int, in_context: int, paths: list[str]) -> str: + """The line the model reads next to the images it was sent. + + `paths` is index-aligned with the images and holds an empty string for one that + could not be stored. The notice always says the files exist, because a model told + the session directory exists and given no path will go looking for one — that hunt + is exactly what this line is here to prevent. + """ + noun = "image" if total == 1 else "images" + if in_context < total: + notice = ( + f"[{total} {noun} attached inline — {in_context} are in your context, " + "the rest were not loaded" + ) + dropped = [path for path in paths[in_context:] if path] + if dropped: + notice += f"; open them with image_view: {', '.join(dropped)}" + else: + notice += " and could not be saved to the session directory" + else: + notice = ( + f"[{total} {noun} attached inline — already in your context, " + "no extra loading needed" + ) + stored = [path for path in paths if path] + if stored: + notice += f". On disk: {', '.join(stored)}" + return notice + "]" + + +class _TaggedSink: + """Routes one tool's live events into the shared batch queue, tagged by index. + + Mirrors the asyncio.Queue surface the sinks actually use (``put`` / + ``put_nowait``) so it is a drop-in replacement for a per-tool queue. + """ + + def __init__(self, queue: asyncio.Queue, idx: int) -> None: + self._queue = queue + self._idx = idx + + async def put(self, item) -> None: + await self._queue.put((self._idx, item)) + + def put_nowait(self, item) -> None: + self._queue.put_nowait((self._idx, item)) + + +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) + + +async def _protocol_progress_message(session_id: str) -> "Message | None": + """System reminder for an active protocol run, or None. + + The run was started in some earlier turn; compression may drop that turn and + the todo lines no longer say which of them the user pre-agreed to. This + reminder keeps the commitment — and the way to close it — in view. + """ + from navi.tools.protocol import get_protocol_message + + return await get_protocol_message(session_id) + + +# Final-turn intercept nudges — injected when the model ends a turn with bare +# text (no tool call) while the todo still has open steps (or a protocol run is +# still open), 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.", +) + +# Same intercept, different move: the todo shows no open work, but a protocol +# run is still `running` because it was never closed. +_PROTOCOL_FINAL_NUDGE = ( + "[Final-turn check] You ended the turn with text only while a protocol run is still " + "open. Close it now: protocol(op='finish', report=...) if its steps are done, or " + "protocol(op='cancel'/'fail', reason=...) if not. Do not leave a run running." +) diff --git a/navi/core/agent/runtime.py b/navi/core/agent/runtime.py new file mode 100644 index 0000000..3047669 --- /dev/null +++ b/navi/core/agent/runtime.py @@ -0,0 +1,181 @@ +"""Turn lifecycle: run, ephemeral runs, compaction and prompt-cache invalidation.""" + +from typing import TYPE_CHECKING +from collections.abc import AsyncGenerator + +import structlog + +from navi.config import settings +from navi.exceptions import ( + NothingToCompactError, + SessionNotFound, +) +from navi.llm.base import Message + +from ..events import ( + AgentEvent, + CompressionStarted, + StreamEnd, +) +from ..subagent_runner import SubAgentOutcome + + + +log = structlog.get_logger() + + +class AgentRuntimeMixin: + 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 _drain_task_notes(self, session_id: str, session) -> None: + """Move finished-background-task notes into the session context. + + Called once at the start of run_stream(), before the first LLM call: + the note becomes part of the LLM-visible context (and the compression + window) but is never displayed — clients learned about completions via + the live task_update event. Only the foreground run_stream mutates the + session, so there is no race with background jobs. + """ + from navi.core import task_notes + + try: + note = await task_notes.drain(session_id) + except Exception: + log.exception("agent.task_notes_drain_failed", session_id=session_id) + return + if not note: + return + session.context.append( + Message( + role="system", + content=note, + metadata={"source": "task_note"}, + is_display=False, + ) + ) + await self._sessions.save(session) + log.info("agent.task_notes_drained", session_id=session_id) + + 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, + ) -> SubAgentOutcome: + """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.") + + window = self._window_for(profile) + yield CompressionStarted( + # Real-baseline estimate when available (see _compression_events_preturn); + # the calibrated estimate is only the fallback before the first LLM + # call of a session. + context_tokens=self._compressor.real_baseline_estimate( + session.context, session.context, profile.model + ), + max_context_tokens=window, + ) + 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", + memory_store=self._memory, + 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), + window=window, + ) + if event is None: + raise NothingToCompactError( + "Nothing to compact yet — the context is still small." + ) + yield event + finally: + _sid_var.reset(_sid_token) + + 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) + + def invalidate_prompt_cache(self, profile_id: str | None = None) -> None: + """Drop cached system prompts so a profile change reaches the next turn. + + The cache lives on this agent's ContextBuilder, and the agent handed out + by the container is process-wide — so an admin toggling a profile's + availability must clear it, or every other prompt keeps advertising the + profile list as it was. + """ + self._ctx_builder.invalidate_system_prompt_cache(profile_id) diff --git a/navi/core/agent/streaming.py b/navi/core/agent/streaming.py new file mode 100644 index 0000000..b3bc48f --- /dev/null +++ b/navi/core/agent/streaming.py @@ -0,0 +1,655 @@ +"""The streaming turn loop: model rounds, tool dispatch and the final intercept.""" + +import asyncio +import time +from datetime import datetime, timezone +from pathlib import Path +from typing import TYPE_CHECKING +from collections.abc import AsyncGenerator + +import structlog + +from navi.config import settings +from navi.exceptions import ( + ContextTooLargeError, + MaxIterationsReached, + SessionNotFound, +) +from navi.llm.base import Message +from navi.session_files import save_images as save_session_images +from navi.tools._internal.base import ( + ToolContext, + current_plan_runner, + current_stop_event, + current_user_info, + current_user_role, +) +from navi.tools.plan import PlanRunner + +from ..agent_run_context import AgentTurnContext, StreamState +from ..anti_stall import AntiStallMonitor +from ..events import ( + AgentEvent, + ModelInfo, + StreamEnd, + StreamStopped, + TextDelta, + ThinkingDelta, + ThinkingEnd, +) +from ..image_tokens import context_image_tokens, estimate_image_tokens +from ..stream_guard import _iter_stream_guarded + + + +log = structlog.get_logger() + + +from .helpers import ( + _FINAL_INTERCEPT_NUDGES, + _PROTOCOL_FINAL_NUDGE, + _inline_images_notice, + _protocol_progress_message, + _todo_progress_message, +) + + +class AgentStreamMixin: + 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) + _scope_token, _keyless_token = await self._apply_mcp_scope(session) + tools = self._tool_list(profile.get_agent_tools()) + tool_schemas = [t.schema() for t in tools] + llm = self._get_backend(profile.llm_backend) + # The profile these four are bound to. The post-turn reload must compare + # against this, not against session.profile_id: after a switch_profile the + # two can be equal again (the store rewrote the row, or mutated this very + # object) while the tool set above is still the old profile's. + bound_profile_id = profile.id + + 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_allowed_mcp_servers as _mcp_scope_var, + ) + from navi.tools._internal.base import ( + current_keyless_mcp_servers as _keyless_var, + ) + from navi.tools._internal.base import ( + current_model as _model_var, + ) + from navi.tools._internal.base import ( + current_profile_id as _profile_var, + ) + from navi.tools._internal.base import ( + current_route as _route_var, + ) + from navi.tools._internal.base import ( + current_session_id as _sid_var, + ) + from navi.tools._internal.base import ( + current_working_directory as _cwd_var, + ) + + _sid_token = _sid_var.set(session_id) + _model_var.set(profile.model) + _route_var.set(getattr(profile, "llm_route", None)) + _profile_var.set(profile.id) + _cwd_token = _cwd_var.set(cwd) + + # Start the turn on the route's primary entry when it is healthy + # again: the background probe is what notices a recovered host, and + # this is where that knowledge becomes visible to the user — the next + # turn, never a running one. A turn that starts while the cloud is + # down begins on the local entry and stays there until it ends. + begin_turn = getattr(llm, "begin_turn", None) + if callable(begin_turn): + begin_turn(getattr(profile, "llm_route", None)) + + # 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). + # The incoming message is priced here because the gate runs before it + # is appended — an attachment is the one part of the next prompt this + # context cannot yet account for. + pending = Message(role="user", content=user_message, images=images or None) + pending_tokens = self._compressor.estimate_tokens_delta([pending], profile.model) + async for _ev in self._compression_events_preturn( + session, llm, profile, session_id, pending_tokens=pending_tokens + ): + 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) + + # Inline chat images land in the session directory too, and the paths go into + # the context next to the images themselves. A pasted image used to exist in + # the transcript only, so an agent told the session directory exists went + # looking there for the "attached images", found nothing, and spent turns on + # fruitless `find` calls before giving up. + saved_paths: list[str] = [] + if images: + try: + names = save_session_images(session_id, images) + except OSError: + log.exception("agent.image_save_failed", session_id=session_id) + names = [] + saved_paths = [ + str(Path(settings.session_files_dir) / session_id / name) if name else "" + for name in names + ] + + # Image token budgeting: fit as many images as possible into the LLM context. + # Images that do not fit stay reachable through the session directory, so the + # notice has to name them. Each image is priced by its own size: a flat 500 + # admitted twice as many screenshots as the model can take (a 1024x1024 image + # is ~1026 tokens on qwen3.6 — see navi/core/image_tokens.py). + images_for_context = images + context_content = user_message + if images: + current_tokens = self._compressor.estimate_tokens( + session.context, profile.model + ) + available_tokens = int(self._window_for(profile) * 0.8) - current_tokens + fitted: list[str] = [] + for img in images: + cost = estimate_image_tokens(img) + if cost > available_tokens: + break + fitted.append(img) + available_tokens -= cost + if len(fitted) < len(images): + images_for_context = fitted + context_content += "\n\n" + _inline_images_notice( + len(images), len(images_for_context or []), saved_paths + ) + + 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"), + ) + ) + # Raw excerpts compression dropped. Deduped across turns through the + # session's ephemeral metadata: the same chunk should not ride along in + # every prompt of a long session just because the topic persists. + if session.session_metadata is None: + session.session_metadata = {} + # A list, not a set: session_metadata is JSON-encoded into the + # sessions row on every save, and a set is not serializable. + injected_chunks = session.session_metadata.setdefault( + "injected_chunk_ids", [] + ) + mem_recall_task = asyncio.create_task( + self._ctx_builder._memory_recall_msg( + session, + user_message, + user_id=session.user_id, + injected_ids=injected_chunks, + ) + ) + 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) + mem_recall = await mem_recall_task + if mem_recall: + ctx_injections.append(mem_recall) + + # Drain background-task completion notes into the session context + # (persisted so compression keeps them; not displayed — the client + # already saw the live task_update event). + await self._drain_task_notes(session_id, session) + + 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) + + protocol_msg = await _protocol_progress_message(session_id) + if protocol_msg: + built_ctx.append(protocol_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, + model=profile.model, + window=self._window_for(profile), + ) + 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) + # This path was never exercised (the old chars//3 check never + # fired), so it kept an invalid StreamEnd(content=...) and died + # with a TypeError instead of reporting the overflow. + yield StreamEnd( + full_content=error_text, + max_context_tokens=self._window_for(profile), + ) + 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, + # The window the compressor accounts against must be the + # window the local model is actually loaded with. + num_ctx=self._window_for(profile), + ), + 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. + # `sent_tokens` lets the compressor notice a prompt the backend cut + # down instead of caching the cut size as the truth (Ollama keeps + # roughly half the window and reports only that). + self._compressor.record_real_baseline( + len(session.context), + state.context_tokens, + chars=self._compressor.context_chars(built_ctx), + model=turn_ctx.resolved_model or profile.model, + sent_tokens=self._compressor.real_baseline_estimate( + session.context, built_ctx, turn_ctx.resolved_model or profile.model + ), + image_tokens=context_image_tokens(built_ctx), + ) + + # 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) + if self._compressor.context_truncated: + # A truncated prompt is reported as what the backend + # evaluated — roughly half the window — while the context + # that produced it was already over the window. Showing + # that number would drop the bar to ~50% at the exact + # moment the session is full; keep the honest estimate. + session.context_token_count = max( + state.context_tokens or 0, + self._compressor.real_baseline_estimate( + session.context, session.context, profile.model + ), + ) + else: + 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 — or a protocol run is open + # even though its steps are all closed — and we haven't hit the + # limit, queue a recovery nudge and continue the loop instead of + # closing. + from navi.tools.protocol import has_active_run + from navi.tools.todo import has_open_steps + + may_intercept = ( + profile.final_intercept_enabled + and turn_ctx.final_interceptions < profile.final_intercept_limit + ) + todo_open = await has_open_steps(session_id) if may_intercept else False + protocol_open = await has_active_run(session_id) if may_intercept else False + if may_intercept and (todo_open or protocol_open): + turn_ctx.final_interceptions += 1 + if protocol_open and not todo_open: + # A run whose steps are all closed needs a different + # closing move than "there is still work to do". + nudge = _PROTOCOL_FINAL_NUDGE + else: + nudge = _FINAL_INTERCEPT_NUDGES[ + min(turn_ctx.final_interceptions, len(_FINAL_INTERCEPT_NUDGES)) - 1 + ] + pending_final_nudge = Message(role="system", content=nudge) + 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=self._window_for(profile), + 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, + profile_id=profile.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(), + ) + # Parallel batch gate: profile override wins, else the global flag. + if turn_ctx.parallel_tool_calls is False: + if profile.parallel_tool_calls is None: + from navi.config import settings as _settings + + turn_ctx.parallel_tool_calls = _settings.parallel_tool_calls + else: + turn_ctx.parallel_tool_calls = profile.parallel_tool_calls + # 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) + _mcp_scope_var.reset(_scope_token) + _keyless_var.reset(_keyless_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. + # Compare against bound_profile_id, never against session.profile_id: + # set_profile() either rewrites the row (and this run's own save() + # then writes its stale value back over it) or mutates the very + # Session object the run holds — in both cases session.profile_id + # and the stored row agree again, the check stayed false and the + # turn kept calling the old profile's tools. The bindings at the + # top of the run are the only truth about what tools are live. + fresh = await self._sessions.get(session_id) + if fresh and fresh.profile_id != bound_profile_id: + session.profile_id = fresh.profile_id + profile = self._profiles.get(session.profile_id) + bound_profile_id = profile.id + self._set_active_profile(profile) + # The rest of this run executes as the new profile — tools that + # resolve a default profile must not keep reporting the old one. + _profile_var.set(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) + 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) + _mcp_scope_var.reset(_scope_token) + _keyless_var.reset(_keyless_token) + raise MaxIterationsReached(profile.max_iterations) + + 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, + target=getattr(chunk, "target", None), + fallback=bool(getattr(chunk, "is_fallback", False)), + ) + 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() diff --git a/navi/core/agent/tools.py b/navi/core/agent/tools.py new file mode 100644 index 0000000..a782285 --- /dev/null +++ b/navi/core/agent/tools.py @@ -0,0 +1,502 @@ +"""Tool execution: scope resolution, workers, parallel and sequential batches.""" + +import asyncio +from typing import TYPE_CHECKING + +import structlog + +from navi.config import settings +from navi.exceptions import ( + LLMBackendError, + LLMConnectionError, +) +from navi.llm.base import LLMBackend, Message +from navi.tools._internal.base import ( + current_event_sink, + current_user_role, +) + +from ..agent_run_context import AgentTurnContext +from ..events import ( + AgentEvent, + AIHelperTokensUsed, + ProfileSwitched, + SubagentComplete, + TodoUpdated, + ToolEvent, + ToolStarted, +) + +if TYPE_CHECKING: + from navi.profiles.base import AgentProfile + + +log = structlog.get_logger() + + +from .helpers import _TOOL_DONE, _TaggedSink + + +class AgentToolsMixin: + async def _apply_mcp_scope(self, session): + """Restrict this run's MCP tool list to the servers its user may use. + + A BYOK-gated server with no personal key must not appear in the list at + all: the manager refuses the call anyway, and a tool the model can see + but never call is read as a transient fault and retried. Admins (and the + no-auth local user, whose role is ``admin``) get ``None`` — the shared + credential in the config is theirs, so nothing is filtered. + + Also records *which* servers were withheld for want of a key. The reason + is not stored anywhere else and is not derivable by the caller: it is the + complement of what ``visible_servers`` kept, so it is computed from that + same value rather than read a second time (a second read of the key store + could disagree with the first). + + Returns the pair of ContextVar tokens; callers reset them where they + already reset the session and cwd vars. + """ + from navi.tools._internal.base import ( + current_allowed_mcp_servers, + current_keyless_mcp_servers, + current_user_id, + ) + + if self._mcp_manager is None: + allowed, keyless = None, frozenset() + else: + try: + allowed = await self._mcp_manager.visible_servers( + current_user_role.get(), current_user_id.get() or session.user_id + ) + # Withheld = gated servers the run did not get to keep. None means + # "no restriction", so nothing was withheld. + keyless = ( + frozenset() + if allowed is None + else frozenset(self._mcp_manager.byok_servers() - allowed) + ) + except Exception: + # Fail closed: an unanswerable visibility question must not fall + # back to "everything is available". Every gated server then counts + # as withheld, which matches the empty tool list the model gets. + log.warning("agent.mcp_scope_failed", exc_info=True) + allowed = frozenset() + keyless = frozenset(self._mcp_manager.byok_servers()) + + return ( + current_allowed_mcp_servers.set(allowed), + current_keyless_mcp_servers.set(keyless), + ) + + 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=self._window_for(profile), + llm=llm, + model=model, + temperature=settings.context_summary_temperature, + session_store=self._sessions, + profile=profile, + # The backend cut the last prompt down (record_real_baseline saw it). + # Its token count is below the trigger, but the context it came from + # is over the window — compress now instead of waiting for a + # threshold that the truncated count will never reach. + context_truncated=self._compressor.pop_context_truncated(), + memory_store=self._memory, + ) + 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 + + async def _execute_tools_with_sink( + self, turn_tool_calls, tools, turn_ctx: AgentTurnContext, session, stop_event, tool_ctx=None + ): + """Execute tool calls, sequentially or as a parallel batch. + + The mode is decided per-turn by ``turn_ctx.parallel_tool_calls`` + (profile override ? global setting, resolved at tool_ctx build). + """ + # A profile switch replaces the toolset mid-turn, so it must not share a + # batch: every neighbour would still be dispatched against the old + # tool_map and die with "tool 'X' not found" (seen with reload_tools called + # just after a switch). Run it alone first, then re-resolve the tools + # it switched to for the rest of the batch. The turn's own ``profile`` + # binding stays old on purpose — the end-of-iteration reload in + # run_stream() is what rebinds profile/llm/schemas. + switching = next((tc for tc in turn_tool_calls if tc.name == "switch_profile"), None) + if switching is not None and len(turn_tool_calls) > 1: + rest = [tc for tc in turn_tool_calls if tc is not switching] + switched_to = None + async for ev in self._execute_tools_sequential( + [switching], tools, turn_ctx, session, stop_event, tool_ctx + ): + if isinstance(ev, ProfileSwitched): + switched_to = ev.profile_id + yield ev + if switched_to: + new_tools = self._tool_list_for_profile(switched_to) + if new_tools is not None: + tools = new_tools + turn_tool_calls = rest + + if turn_tool_calls and turn_ctx.parallel_tool_calls and len(turn_tool_calls) > 1: + async for ev in self._execute_tools_parallel( + turn_tool_calls, tools, turn_ctx, session, stop_event, tool_ctx + ): + yield ev + else: + async for ev in self._execute_tools_sequential( + turn_tool_calls, tools, turn_ctx, session, stop_event, tool_ctx + ): + yield ev + + async def _execute_tools_parallel( + self, turn_tool_calls, tools, turn_ctx: AgentTurnContext, session, stop_event, tool_ctx=None + ): + """Execute an independent batch of tool calls concurrently. + + ToolStarted is emitted for every call up-front (in call order) so the + UI shows the whole batch immediately. Live events from all tools merge + through one tagged queue; results and tool messages are appended in + call order after the batch settles, with a single session save. + + Only batch independent calls — two actions over the same terminal in + one batch race each other. + """ + from navi.tools.todo import started_metadata_for_call + + tool_map = {t.name: t for t in tools} + for tc in turn_tool_calls: + yield ToolStarted( + tool_name=tc.name, + arguments=tc.arguments, + tool_call_id=tc.id, + metadata=await started_metadata_for_call(tc, tool_ctx), + ) + + shared: asyncio.Queue = asyncio.Queue() + holders: list[list] = [[] for _ in turn_tool_calls] + tasks: list[asyncio.Task] = [] + + for idx, tc in enumerate(turn_tool_calls): + tagged = _TaggedSink(shared, idx) + sink_token = current_event_sink.set(tagged) + + async def _run_with_sentinel(_tc=tc, _idx=idx, _holder=holders[idx], _sink=tagged): + 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) + + tasks.append(asyncio.create_task(_run_with_sentinel())) + current_event_sink.reset(sink_token) + + try: + pending = set(range(len(tasks))) + stopped = False + while pending: + try: + item = await asyncio.wait_for(shared.get(), timeout=1.0) + except asyncio.TimeoutError: + if stop_event and stop_event.is_set(): + stopped = True + break + continue + + _idx, value = item + if value is _TOOL_DONE: + pending.discard(_idx) + continue + if isinstance(value, SubagentComplete): + turn_ctx.subagent_tokens += value.token_count + turn_ctx.tool_call_count += value.tool_call_count + elif isinstance(value, AIHelperTokensUsed): + turn_ctx.subagent_tokens += value.completion_tokens + else: + yield value + + if stopped: + # Stop must actually stop work: cancel in-flight tools the same + # way the sequential path does (docs/tasks.md: stop mid-batch + # cancels every unfinished tool). Calls that already finished + # keep their real results — only unfinished ones get the + # synthetic stopped note below; erasing a completed + # backgrounded call (task_id) would orphan the task in the + # client UI. + log.info( + "agent.batch_stopped", + tools=[tc.name for tc in turn_tool_calls], + ) + for t in tasks: + if not t.done(): + t.cancel() + await asyncio.gather(*tasks, return_exceptions=True) + # Same reason as in the sequential path: whatever the cancelled + # tools announced before unwinding still belongs to cards the + # client is showing, so it is forwarded rather than dropped. + while True: + try: + _idx, value = shared.get_nowait() + except asyncio.QueueEmpty: + break + if value is _TOOL_DONE: + continue + if isinstance(value, SubagentComplete): + turn_ctx.subagent_tokens += value.token_count + turn_ctx.tool_call_count += value.tool_call_count + elif isinstance(value, AIHelperTokensUsed): + turn_ctx.subagent_tokens += value.completion_tokens + else: + yield value + + # Results and messages in call order, one save for the batch. + # A stopped batch mixes real results (finished calls) with the + # synthetic stopped note (cancelled calls). + for tc, holder in zip(turn_tool_calls, holders): + if not holder: + if stopped: + 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, + ) + ) + continue + r = RuntimeError("tool task produced no result") + else: + r = holder[0] + 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) + await self._sessions.save(session) + finally: + # Cancel whatever is still running (generator closed mid-batch or + # an outer cancellation of the run), then gather to contain each + # cancelled task — awaiting the results directly would re-raise + # CancelledError past the results already produced, while an + # outer cancellation of the run itself still propagates. + for t in tasks: + if not t.done(): + t.cancel() + await asyncio.gather(*tasks, return_exceptions=True) + + async def _execute_tools_sequential( + self, turn_tool_calls, tools, turn_ctx: AgentTurnContext, session, stop_event, tool_ctx=None + ): + """Execute tool calls one at a time 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: + # A cancelled tool may still have announced work the client + # is already showing: a sub-agent emits a closing ToolEvent + # for its in-flight tool from the CancelledError handler. + # Read those before dropping the queue, or the client keeps + # a spinner it can never resolve until the page reloads. + await asyncio.gather(tool_task, return_exceptions=True) + while True: + try: + item = sink.get_nowait() + except asyncio.QueueEmpty: + break + if item is _TOOL_DONE: + continue + 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 + # 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() + # gather(return_exceptions=True) contains the cancelled tool + # task (awaiting it directly would re-raise CancelledError and + # abort the turn before the synthetic stopped results land), + # while an outer cancellation of the run itself propagates. + await asyncio.gather(tool_task, return_exceptions=True) + + 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) diff --git a/navi/core/context_builder.py b/navi/core/context_builder.py deleted file mode 100644 index 41403cc..0000000 --- a/navi/core/context_builder.py +++ /dev/null @@ -1,819 +0,0 @@ -"""Build LLM context lists for agent turns. - -Extracted from agent.py to reduce the Agent class surface area. -""" - -from datetime import datetime, timezone -from pathlib import Path -from typing import TYPE_CHECKING - -import structlog - -from navi.llm.base import Message - -import navi.config as _config - -log = structlog.get_logger() - - -def _fact_age(updated_at: str) -> str: - """How old a fact is, so a stale one reads as stale. - - Nothing in memory expires on a clock — a fact is wrong when the world moves, - not when a timer fires — so the injection says how old it is and lets the - agent weigh it. A missing or unparseable timestamp simply says nothing. - """ - try: - then = datetime.fromisoformat(updated_at) - except (TypeError, ValueError): - return "" - if then.tzinfo is None: - then = then.replace(tzinfo=timezone.utc) - days = (datetime.now(timezone.utc) - then).days - if days <= 0: - return "today" - if days < 30: - return f"{days}d ago" - return f"{days // 30}mo ago" - - -# Roles whose content may be head/tail-truncated when a single message is too -# large for the context window. System/summary messages are already compact. -# User messages carry intent, so their cap is deliberately more generous than -# tool output's (see `_truncation_budget`) — but leaving them uncapped at all -# turned a single oversized paste into a dead session: the size check refused -# every subsequent call while compression had nothing it could split. -_TRUNCATABLE_ROLES = {"tool", "assistant", "user"} - -# A user message may use up to this fraction of the window before its middle is -# elided for the LLM view (the stored history keeps the full text). -_USER_MESSAGE_WINDOW_FRAC = 1 / 3 - - -def _truncation_budget(role: str) -> int: - """Per-message view cap in tokens for a truncatable role.""" - budget = _config.settings.context_message_token_budget - if budget <= 0: - budget = max(1, _config.settings.ollama_num_ctx // 6) - if role == "user": - budget = max( - budget, - max(1, int(_config.settings.ollama_num_ctx * _USER_MESSAGE_WINDOW_FRAC)), - ) - return budget - -# Which stored system-role messages are infrastructure (and therefore dropped -# from the LLM view) and which are addressed to the model. `task_note` carries -# detached tool results; `synapse_event` is the visible record of the event a -# Synapse reaction is answering — the model must see the same request the user -# sees. Anything else in the transcript with role="system" is bookkeeping. -_LLM_SYSTEM_SOURCES = {"task_note", "synapse_event"} - - -def _truncate_content(content: str, budget_tokens: int, *, label: str = "truncated") -> str: - """Head/tail-truncate ``content`` to roughly ``budget_tokens`` tokens. - - Keeps the head and tail verbatim with a marker naming the elided middle, so - the model retains the beginning and end of a large result (usually the - most informative parts) without the whole thing blowing the context window. - """ - budget_chars = budget_tokens * 3 - if len(content) <= budget_chars: - return content - half = budget_chars // 2 - dropped_tokens = (len(content) - 2 * half) // 3 - marker = f"\n[…{label} ~{dropped_tokens:,} tokens…]\n" - return content[:half] + marker + content[-half:] - - -def truncate_oversized(conv: list[Message]) -> list[Message]: - """Head/tail-truncate oversized messages for the LLM view. - - A single huge tool result (e.g. reading a large file) can alone exceed - the context window, and the turn/message-count-based compressor cannot - shrink a single message. This per-build pass caps each truncatable - message at its role budget — ``context_message_token_budget`` tokens - (0 = num_ctx // 6) for tool/assistant output, at least num_ctx // 3 for - user text — keeping head + tail so the model still sees the start and end. - Copies are returned — stored history (session.messages) is never - mutated, so the user still sees full content and reloads are unaffected. - - Module-level because the compressor reports the session's context size in - the same units: a raw context holding a truncated-away megabyte is not what - the model is sent, and counting it as such shows a scary 800k over a 128k - window. - """ - out: list[Message] = [] - for m in conv: - if m.role not in _TRUNCATABLE_ROLES: - out.append(m) - continue - budget = _truncation_budget(m.role) - if len(m.content or "") <= budget * 3: - out.append(m) - continue - label = "user message truncated" if m.role == "user" else "truncated" - out.append( - m.model_copy( - update={"content": _truncate_content(m.content or "", budget, label=label)} - ) - ) - return out - - -_INJECTED_IDS_LIMIT = 100 - - -def _recall_query(context: list, user_message: str, max_chars: int = 1200) -> str: - """A search query made of the recent turn window, not just the last message. - - The user's "and what about the port?" is unsearchable on its own; in the - company of the previous two exchanges it names the topic. Assistant replies - are included in shortened form because the concrete names (a file, a value, a - plan) usually appear there, not in the user's own wording. - """ - parts: list[str] = [] - turns = [m for m in context if m.role in ("user", "assistant") and not m.is_summary] - for msg in turns[-4:]: - text = (msg.content or "").strip() - if text: - parts.append(text[:300]) - tail = (user_message or "").strip() - if tail: - parts.append(tail[:300]) - query = "\n".join(parts) - return query[-max_chars:].strip() - - -def _fact_value_is_abs_path_outside(value: str, cwd: str) -> bool: - """True when a fact's value is an absolute filesystem path outside `cwd`. - - Used to keep context-dependent path facts (e.g. "project_root → /home/.../navi-1") - from being injected into sessions running in a different project, where they - would misdirect the agent. Relative values are left through (we cannot know - what they were recorded relative to). Non-path values always pass. - """ - v = str(value).strip() - if not v or not (v.startswith("/") or v.startswith("~")): - return False - try: - p = Path(v).expanduser().resolve() - base = Path(cwd).expanduser().resolve() - except Exception: - return True # unresolvable path — treat as out-of-scope, drop it - try: - p.relative_to(base) - return False # within the session cwd — keep - except ValueError: - return True # absolute path outside the cwd — drop - -if TYPE_CHECKING: - from navi.context_providers._loader import ContextProviderRegistry - from navi.memory.store import MemoryStore - from navi.profiles.base import AgentProfile - - -def methodologies_block(profile: "AgentProfile") -> str | None: - """The "how to use the methodologies catalog" block for profiles that have it. - - ``skills`` and ``protocol`` are useless as bare tool schemas: the model meets - two unfamiliar tools with no reason to reach for them, and probes them - instead of working. The rules that say when to search the catalog, how to run - a protocol and when to write an entry belong in the system prompt, once, - instead of seven profile prompts drifting apart. - """ - scope = getattr(getattr(profile, "tools", None), "agent", None) - native = list(getattr(scope, "native", None) or []) - has_skills = "skills" in native - has_protocol = "protocol" in native - if not (has_skills or has_protocol): - return None - - tools = " and ".join(f"`{t}`" for t in ("skills", "protocol") if t in native) - manuals = " / ".join( - f'`tool_manual("{t}")`' for t in ("skills", "protocol") if t in native - ) - intro = ( - f"{tools} are this Navi's catalog of *how* to do things: what it has already " - f"learned and agreed on. To read up on them use {manuals} — do not call them " - f"with a guess to see what happens." - ) - lines = [ - "## Methodologies — skills and protocols", - intro, - ] - if has_skills: - lines.append( - '- **Recurring work: search first.** "как обычно", "как мы делали", a standard ' - 'shutdown, a deployment, a routine document — call `skills(op="search")` once ' - "before improvising, then `load` the entry that fits. One call replaces a whole " - "rediscovery. Not for a one-off question, never twice for the same query." - ) - lines.append( - "- **A skill is advice** — follow it as far as it fits the task, and say where " - "you departed from it." - ) - if has_protocol: - lines.append( - '- **A protocol is a commitment.** Run it with `protocol(op="start")`, never by ' - "retyping its steps into the todo. A protocol whose triggers match the request is " - "the user's pre-agreed procedure: say you are running it and start. Each injected " - "step is marked done only after its check passes; the run ends with `finish` plus " - "a report on the post-checks, or with `fail`/`cancel` and the reason. Never " - "improvise around a step — that is the error a protocol exists to prevent." - ) - if has_skills: - lines.append( - "- **Record what proved itself.** An approach that worked and will recur is one " - '`skills(op="create")` and a line to the user. A protocol is never created ' - 'directly: `skills(op="propose")` it, say what it does, and wait for the user\'s ' - "decision." - ) - return "\n\n".join(lines) - - -def profiles_block(current_id: str, others: list) -> str: - """The "Available profiles" block: the cross-profile map, plus how to act on it. - - The block is built in one place because the admin prompt-debug view renders - it too — two copies used to tell the model different things about switching. - The rules themselves live in the persona; this is the short reminder beside - the list, where the model is about to make the choice. - """ - lines = ["## Available profiles", f"Current: **{current_id}**"] - for profile in others: - desc = profile.short_description or profile.description - lines.append(f"· {profile.id}: {desc}") - lines.append( - "→ Decide on your own — do not ask permission. The task you were given belongs " - "to another profile's domain → `switch_profile`, called alone, then say which " - "profile is active and why. Another profile's tools are needed only for a side " - "piece of your own task → `spawn_agent(profile_id=…)` and keep this session's " - "profile. Details: `list_profiles`; what a candidate can actually call: " - "`list_tools(profile_id=…)`." - ) - return "\n".join(lines) - - -async def render_todo_lines(session_id: str) -> list[str]: - """Return a list of formatted todo lines for goal anchoring.""" - try: - from navi.tools.todo import render_todo_lines as _rtl - - return await _rtl(session_id) - except Exception: - return [] - - -class ContextBuilder: - """Handles construction of the full LLM context for each turn.""" - - def __init__( - self, - profile_registry, - memory_store: "MemoryStore | None" = None, - memory_search=None, - cp_registry: "ContextProviderRegistry | None" = None, - mcp_manager=None, - ) -> None: - self._profiles = profile_registry - self._memory = memory_store - self._memory_search = memory_search - self._cp_registry = cp_registry - self._mcp_manager = mcp_manager - # Keyed by (profile id, user role): the "Available profiles" block lists - # admin-only profiles to admins only, and this builder is a process-wide - # singleton, so a role-blind key would leak one role's list to the other. - self._system_prompt_cache: dict[tuple[str, str], str] = {} - - def build_system_prompt(self, profile: "AgentProfile") -> str: - """Build the system prompt string for a profile (cached per profile and role).""" - from navi.profiles.base import admin_only_blocked - from navi.tools._internal.base import current_user_role as _role_var - - role = _role_var.get() - cache_key = (profile.id, role) - cached = self._system_prompt_cache.get(cache_key) - if cached is not None: - return cached - - parts: list[str] = [] - identity = self._instance_identity() - if identity is not None: - parts.append(identity) - persona = _config.settings.navi_persona.strip() - if persona: - parts.append(persona) - parts.append(profile.system_prompt) - - methodologies = methodologies_block(profile) - if methodologies is not None: - parts.append(methodologies) - - other = [ - p - for p in self._profiles.all() - if p.id != profile.id - and not getattr(p, "is_hidden", False) - and not admin_only_blocked(p, role) - ] - if other: - parts.append(profiles_block(profile.id, other)) - - result = "\n\n---\n\n".join(parts) - self._system_prompt_cache[cache_key] = result - return result - - def invalidate_system_prompt_cache(self, profile_id: str | None = None) -> None: - if profile_id is None: - self._system_prompt_cache.clear() - else: - for key in [k for k in self._system_prompt_cache if k[0] == profile_id]: - del self._system_prompt_cache[key] - - def _instance_identity(self) -> str | None: - """The 'you are navi on host X' block for the system prompt.""" - import socket - - try: - from navi.identity import load_identity - - ident = load_identity(_config.settings.instance_file) - except Exception: - return None - lines = [ - f"You are navi '{ident.name}' (instance {ident.instance_id[:8]}) " - f"running on host {socket.gethostname()}.", - "Other navi instances in the swarm know this machine by that name — " - "identify yourself as '" + ident.name + "' when talking to remote peers.", - ] - return "\n".join(lines) - - async def _memory_msg(self, user_id: str | None = None) -> "Message | None": - if self._memory is None: - return None - summary = await self._memory.get_summary(user_id=user_id) - if not summary: - return None - return Message( - role="system", - content=f"## What I remember about the user\n\n{summary}", - ) - - async def _memory_facts_msg( - self, - user_message: str, - user_id: str | None = None, - injected_ids: set[str] | None = None, - scope_boundary_enabled: bool = False, - session_cwd: str | None = None, - ) -> "Message | None": - if self._memory is None: - log.info("memory_facts_msg.skip", reason="no_memory_store") - return None - msg = user_message.strip() - if len(msg) <= 20: - log.info("memory_facts_msg.skip", reason="too_short", length=len(msg)) - return None - words = [w for w in msg.split() if any(c.isalpha() for c in w)] - if len(words) < 2: - log.info("memory_facts_msg.skip", reason="too_few_words", words=len(words)) - return None - if len(msg) < 50: - limit = 1 - elif len(msg) <= 150: - limit = 2 - else: - limit = 3 - log.info("memory_facts_msg.search", query=msg[:60], limit=limit, user_id=user_id) - facts = await self._memory.search_facts(msg, user_id=user_id, limit=limit) - log.info("memory_facts_msg.hits", count=len(facts) if facts else 0) - if not facts: - return None - if injected_ids is not None: - new_facts = [f for f in facts if f.get("id") not in injected_ids] - skipped = len(facts) - len(new_facts) - log.info("memory_facts_msg.dedup", new=len(new_facts), skipped=skipped) - if not new_facts: - return None - for f in new_facts: - injected_ids.add(f["id"]) - else: - new_facts = facts - - # Bounded autonomy: drop facts whose value is an absolute path outside the - # session cwd (e.g. "project_root → /home/.../navi-1" while the user works - # in another project) so they cannot misdirect the agent. Facts are kept - # when working inside that path (then they are correct). Gated by the - # profile flag so free-flight stays reproducible by toggling it off. - if scope_boundary_enabled and session_cwd: - kept = [f for f in new_facts if not _fact_value_is_abs_path_outside(str(f.get("value", "")), session_cwd)] - dropped = len(new_facts) - len(kept) - if dropped: - log.info("memory_facts_msg.scope_filtered", dropped=dropped, session_cwd=session_cwd) - new_facts = kept - if not new_facts: - return None - - lines = ["Known facts about the user:"] - for f in new_facts: - conf = f.get("confidence", 70) - age = _fact_age(str(f.get("updated_at", ""))) - age_note = f", updated {age}" if age else "" - lines.append( - f"• [{f['category']}] {f['key']} → {f['value']} " - f"(confidence: {conf}%{age_note})" - ) - result = "\n".join(lines) - log.info("memory_facts_msg.injected", lines=len(new_facts), length=len(result)) - return Message(role="system", content=result) - - async def _memory_recall_msg( - self, - session, - user_message: str, - user_id: str | None = None, - injected_ids: list[str] | None = None, - ) -> "Message | None": - """Restore a few verbatim excerpts that compression dropped. - - Only fires once the session actually carries a summary: before that, the - history is still in the context and a search would just duplicate it. - The query is built from the window of recent turns, not the last message, - because "and what about the other one?" is only meaningful next to what - came before it. - - This is the recall path that does not wait for the model to call a tool. - The model can still call `memory_search` for anything wider. - """ - if self._memory_search is None or not _config.settings.memory_recall_enabled: - return None - context = getattr(session, "context", None) or [] - if not any(m.is_summary for m in context): - return None - session_id = getattr(session, "id", None) - query = _recall_query(context, user_message) - if not query: - return None - try: - hits = await self._memory_search.search( - query, - scope="raw", - k=_config.settings.memory_recall_k, - user_id=user_id, - session_id=session_id, - current_session_id=session_id, - summary_at=await self._memory_search.latest_summary_at(session_id), - ) - except Exception: - log.debug("memory_recall.search_failed", exc_info=True) - return None - - selected = [] - for hit in hits: - if hit.source_type != "raw" or not hit.chunk_id: - continue - if injected_ids is not None and hit.chunk_id in injected_ids: - continue - if hit.score is not None: - if hit.score < _config.settings.memory_recall_min_similarity: - continue - elif hit.terms_hit < 2: - # No vector score (lexical fallback): demand a real keyword - # overlap, otherwise a one-word coincidence injects a turn. - continue - if hit.superseded_by: - continue - selected.append(hit) - if len(selected) >= _config.settings.memory_recall_k: - break - if not selected: - return None - if injected_ids is not None: - for hit in selected: - injected_ids.append(hit.chunk_id) - # The ledger lives in `session_metadata`, which is JSON-encoded into - # the sessions row on every save — an unbounded list grows the row - # for the life of the session. Older ids are dropped; a chunk that - # falls out after ~100 injections may be recalled again, which is - # a much smaller cost than a row that never stops growing. - if len(injected_ids) > _INJECTED_IDS_LIMIT: - del injected_ids[: len(injected_ids) - _INJECTED_IDS_LIMIT] - - lines = [ - "## Recovered from earlier in this conversation", - "These are verbatim excerpts of turns that were compressed out of context. " - "The summary is the curated version; these carry the exact details — names, " - "commands, values, links.", - ] - for hit in selected: - turn = ( - f"turn {hit.start_seq}" if hit.start_seq == hit.end_seq - else f"turns {hit.start_seq}–{hit.end_seq}" - ) - if hit.end_seq < 0: - turn = "turn unknown" - when = hit.timestamp.strftime("%Y-%m-%d %H:%M") if hit.timestamp else "unknown time" - lines.append(f"\n• session {hit.session_id}, {turn}, {when}") - lines.append(hit.text[:1500].strip()) - result = "\n".join(lines) - log.info( - "memory_recall.injected", - session_id=session_id, - chunks=len(selected), - length=len(result), - ) - return Message(role="system", content=result) - - async def _collect_context_injections(self, profile: "AgentProfile") -> list[Message]: - if self._cp_registry is None: - return [] - out: list[Message] = [] - for provider in self._cp_registry.all(): - if not provider.global_provider and provider.name not in profile.context_providers: - continue - try: - text = await provider.get_context() - if text: - out.append(Message(role="system", content=text)) - except Exception: - pass - return out - - async def _build_goal_anchor(self, session_id: str, user_message: str) -> Message: - lines = [ - "[Goal anchor]", - f"Original request: {user_message}", - ] - todo_lines = await render_todo_lines(session_id) - if todo_lines: - lines.append("Current todo:") - lines.extend(todo_lines) - lines.append("Stay on track — complete the remaining pending/in_progress steps.") - lines.append( - "Use 1-based todo indexes. Mark completed steps done only after verification, with validation." - ) - lines.append( - "Before final response, update todo for every completed step, including the final one." - ) - return Message(role="system", content="\n".join(lines)) - - def _scope_boundary_msg(self, profile: "AgentProfile") -> Message | None: - """Bounded-autonomy rule: keep the agent within the requested scope. - - Returns None when the profile has not opted in (legacy free-flight behavior). - """ - if not profile.scope_boundary_enabled: - return None - return Message( - role="system", - content=( - "[Scope boundary]\n" - "Act strictly within the user's literally requested scope. Do NOT expand to " - "sibling or parent directories/projects beyond what was asked, and do NOT start " - "executing discovered TODO lists, roadmaps, milestones, or backlog docs unless " - "the user explicitly asks you to. If you encounter such docs while exploring, " - "report what you found — do not act on it. Autonomy means finishing the requested " - "task well, not enlarging it." - ), - ) - - def _security_policy_msg(self) -> Message | None: - """Build a dynamic security policy system message based on user role.""" - from navi.tools._internal.base import ( - current_user_id as _uid_var, - current_user_role as _role_var, - ) - - user_id = _uid_var.get(None) - role = _role_var.get() - if role == "admin": - return Message( - role="system", - content=( - "[Security policy]\n" - "Role: admin\n" - "Full system access. No restrictions apply.\n" - "You may use any tool, access any path, and execute any command." - ), - ) - if user_id: - allowed = _config.settings.terminal_user_allowed_commands_list - return Message( - role="system", - content=( - "[Security policy]\n" - f"Role: user (user_id={user_id})\n" - f"Filesystem areas: user_data/{user_id}/ and the current session directory " - "(its path is given in the session context above).\n" - "You MUST NOT attempt to access paths outside those two areas.\n" - f"Terminal allowed commands: {', '.join(allowed)}\n" - "You MUST NOT use terminal for: curl, wget, ssh, scp, sudo, system-wide destructive operations, " - "or any command not in the allowlist.\n" - "If a task requires admin privileges (e.g. system-wide changes, accessing another user's files, " - "installing packages globally), tell the user to contact an admin.\n" - "Always prefer filesystem tool over terminal for file operations." - ), - ) - # Legacy / single-user mode — no policy injected - return None - - def _user_context_msg(self) -> Message | None: - """Build a [User context] system message from current_user_info ContextVar.""" - from navi.tools._internal.base import ( - current_user_info as _uinfo_var, - current_user_role as _role_var, - ) - - info = _uinfo_var.get(None) - if not info: - return None - lines = ["[User context]"] - if info.get("display_name"): - lines.append(f"Name: {info['display_name']}") - if info.get("username"): - lines.append(f"Username: {info['username']}") - if info.get("first_name") or info.get("last_name"): - parts = [] - if info.get("first_name"): - parts.append(info["first_name"]) - if info.get("last_name"): - parts.append(info["last_name"]) - lines.append(f"Full name: {' '.join(parts)}") - if info.get("email"): - lines.append(f"Email: {info['email']}") - if info.get("phone"): - lines.append(f"Phone: {info['phone']}") - if info.get("birth_date"): - lines.append(f"Birth date: {info['birth_date']}") - if info.get("country") or info.get("city"): - parts = [] - if info.get("city"): - parts.append(info["city"]) - if info.get("country"): - parts.append(info["country"]) - lines.append(f"Location: {', '.join(parts)}") - if info.get("locale"): - lines.append(f"Locale: {info['locale']}") - lines.append(f"Role: {_role_var.get()}") - return Message(role="system", content="\n".join(lines)) - - def _mcp_context_msg(self, profile: "AgentProfile | None" = None) -> "Message | None": - """Build a system message naming the MCP servers this run can reach. - - One line per server — what it is for — and a pointer to the full text. The - instructions themselves used to be inlined here, which cost ~8 KB of prose - on *every* request of a heavy profile; they are now read on demand with - ``tool_manual("")``. - - A server whose tools were withheld for want of a personal key is still - named, followed by the reason. Listing it without one was the worst of - both: the model knew the server existed, had none of its tools, and had - nothing to tell the user. - """ - if not self._mcp_manager: - return None - if profile is not None and not profile.get_agent_tools().mcp: - return None - from navi.tools._internal.base import current_keyless_mcp_servers as _keyless_var - - server_names = set(profile.get_agent_tools().mcp.keys()) if profile is not None else None - summaries = self._mcp_manager.get_summaries(server_names) - if not summaries: - return None - keyless = _keyless_var.get() - lines = [ - "[MCP servers — external knowledge sources]", - "Their tools are named mcp____. Before using a server's tools, " - 'and whenever it is unclear which of them fits, read its full instructions ' - 'with tool_manual("").', - ] - for name in sorted(summaries): - lines.append("") - lines.append(f"## {name} — {summaries[name]}") - if name in keyless: - lines.append( - " (tools withheld: this server runs on a personal MCP key and none " - "is saved for this account — do not retry it or call tool_manual for " - "it; tell the user to add the key in Settings → MCP keys.)" - ) - return Message(role="system", content="\n".join(lines)) - - def _truncate_oversized(self, conv: list[Message]) -> list[Message]: - """Per-build pass; see ``truncate_oversized`` for the rule.""" - return truncate_oversized(conv) - - def build( - self, - session_context: list[Message], - profile: "AgentProfile", - mem: "Message | None", - iteration: int | None = None, - max_iterations: int | None = None, - extra_system: list[Message] | None = None, - session_id: str | None = None, - session_metadata: dict | None = None, - ) -> list[Message]: - system_prompt = self.build_system_prompt(profile) - if session_id: - system_prompt += ( - f"\n\n---\n\n" - f"[Session context]\n" - f"Session ID: {session_id}\n" - f"Session files directory: {_config.settings.session_files_dir}/{session_id}/\n" - f"When writing files the user should see, always use the session directory path above.\n" - f"When calling MCP navi-3d tools (compile_scad, render_stl, lint_scad), pass ONLY the filename " - f"(e.g. 'falcon9_rocket.scad') for source_path and output_path. Do NOT include the session_id or " - f"the session_files directory in those paths — the MCP server resolves them automatically." - ) - cwd = (session_metadata or {}).get("cwd") - if cwd: - system_prompt += ( - f"\n\n---\n\n" - f"[Working directory]\n" - f"The user launched Navi Code from: {cwd}\n" - f"Treat this directory as the project root. Resolve relative paths against it " - f"unless the user provides an absolute path. For filesystem/terminal/code_exec operations, " - f"prefer this directory when no explicit working_dir is given." - ) - # The standing reaction rules for this run. They arrive through a - # ContextVar rather than the transcript so that they shape every request - # of the run without being stored as a message the user would read. - from navi.tools._internal.base import ( - current_reaction_instructions as _reaction_var, - ) - - reaction_instructions = (_reaction_var.get(None) or "").strip() - if reaction_instructions: - system_prompt += ( - f"\n\n---\n\n" - f"[Reaction session]\n" - f"This run answers a platform event on the user's behalf. Their standing " - f"instructions for how to react:\n\n{reaction_instructions}" - ) - system_msg = Message(role="system", content=system_prompt) - # System-role history is normally infrastructure-only (hidden from the - # LLM), but a few sources are addressed to the model — see the constant. - conv = [ - m for m in session_context - if m.role != "system" - or (m.metadata or {}).get("source") in _LLM_SYSTEM_SOURCES - ] - conv = self._truncate_oversized(conv) - result: list[Message] = [system_msg] - if mem: - result.append(mem) - - # Inject user profile context for multi-user mode - user_ctx = self._user_context_msg() - if user_ctx: - result.append(user_ctx) - - # Inject security policy for multi-user mode - policy = self._security_policy_msg() - if policy: - result.append(policy) - - # Inject bounded-autonomy scope boundary when the profile opts in - boundary = self._scope_boundary_msg(profile) - if boundary: - result.append(boundary) - - # Inject MCP server instructions into context - mcp_msg = self._mcp_context_msg(profile) - if mcp_msg: - result.append(mcp_msg) - - if extra_system: - result.extend(extra_system) - result.extend(conv) - - if ( - profile.iteration_budget_enabled - and iteration is not None - and max_iterations is not None - ): - remaining_after_this = max_iterations - iteration - 1 - if remaining_after_this <= 2: - urgency = ( - f" CRITICAL: only {remaining_after_this} iteration(s) left after this one — finish or produce " - "a partial result now, do not start new subtasks." - ) - elif remaining_after_this <= 5: - urgency = ( - " Low iteration budget: complete the current step, continue necessary " - "verification/publishing, and avoid starting unrelated subtasks." - ) - else: - urgency = "" - result.append( - Message( - role="system", - content=( - f"[Iteration {iteration + 1}/{max_iterations} — " - f"{remaining_after_this} iteration(s) after this one.{urgency}]" - ), - ) - ) - - return result diff --git a/navi/core/context_builder/__init__.py b/navi/core/context_builder/__init__.py new file mode 100644 index 0000000..fdd1b7c --- /dev/null +++ b/navi/core/context_builder/__init__.py @@ -0,0 +1,40 @@ +"""LLM context assembly. + +`helpers` holds the pure pieces (truncation, fact freshness, prompt blocks); +`builder` holds `ContextBuilder`, which assembles the system prompt, memory +context and the per-message view the model sees. +""" + +from .builder import ContextBuilder +from .helpers import ( + _INJECTED_IDS_LIMIT, + _LLM_SYSTEM_SOURCES, + _TRUNCATABLE_ROLES, + _USER_MESSAGE_WINDOW_FRAC, + _fact_age, + _fact_value_is_abs_path_outside, + _recall_query, + _truncate_content, + _truncation_budget, + methodologies_block, + profiles_block, + render_todo_lines, + truncate_oversized, +) + +__all__ = [ + "_INJECTED_IDS_LIMIT", + "_LLM_SYSTEM_SOURCES", + "_TRUNCATABLE_ROLES", + "_USER_MESSAGE_WINDOW_FRAC", + "ContextBuilder", + "_fact_age", + "_fact_value_is_abs_path_outside", + "_recall_query", + "_truncate_content", + "_truncation_budget", + "methodologies_block", + "profiles_block", + "render_todo_lines", + "truncate_oversized", +] diff --git a/navi/core/context_builder/builder.py b/navi/core/context_builder/builder.py new file mode 100644 index 0000000..06061ed --- /dev/null +++ b/navi/core/context_builder/builder.py @@ -0,0 +1,586 @@ +"""`ContextBuilder` — assembles the system prompt, memory context and the LLM view.""" + + +from typing import TYPE_CHECKING + +import structlog + +import navi.config as _config +from navi.llm.base import Message + +from .helpers import ( + _INJECTED_IDS_LIMIT, + _LLM_SYSTEM_SOURCES, + _fact_age, + _fact_value_is_abs_path_outside, + _recall_query, + methodologies_block, + profiles_block, + render_todo_lines, + truncate_oversized, +) + +if TYPE_CHECKING: + from navi.context_providers._loader import ContextProviderRegistry + from navi.memory.store import MemoryStore + from navi.profiles.base import AgentProfile + +log = structlog.get_logger() + + +class ContextBuilder: + """Handles construction of the full LLM context for each turn.""" + + def __init__( + self, + profile_registry, + memory_store: "MemoryStore | None" = None, + memory_search=None, + cp_registry: "ContextProviderRegistry | None" = None, + mcp_manager=None, + ) -> None: + self._profiles = profile_registry + self._memory = memory_store + self._memory_search = memory_search + self._cp_registry = cp_registry + self._mcp_manager = mcp_manager + # Keyed by (profile id, user role): the "Available profiles" block lists + # admin-only profiles to admins only, and this builder is a process-wide + # singleton, so a role-blind key would leak one role's list to the other. + self._system_prompt_cache: dict[tuple[str, str], str] = {} + + def build_system_prompt(self, profile: "AgentProfile") -> str: + """Build the system prompt string for a profile (cached per profile and role).""" + from navi.profiles.base import admin_only_blocked + from navi.tools._internal.base import current_user_role as _role_var + + role = _role_var.get() + cache_key = (profile.id, role) + cached = self._system_prompt_cache.get(cache_key) + if cached is not None: + return cached + + parts: list[str] = [] + identity = self._instance_identity() + if identity is not None: + parts.append(identity) + persona = _config.settings.navi_persona.strip() + if persona: + parts.append(persona) + parts.append(profile.system_prompt) + + methodologies = methodologies_block(profile) + if methodologies is not None: + parts.append(methodologies) + + other = [ + p + for p in self._profiles.all() + if p.id != profile.id + and not getattr(p, "is_hidden", False) + and not admin_only_blocked(p, role) + ] + if other: + parts.append(profiles_block(profile.id, other)) + + result = "\n\n---\n\n".join(parts) + self._system_prompt_cache[cache_key] = result + return result + + def invalidate_system_prompt_cache(self, profile_id: str | None = None) -> None: + if profile_id is None: + self._system_prompt_cache.clear() + else: + for key in [k for k in self._system_prompt_cache if k[0] == profile_id]: + del self._system_prompt_cache[key] + + def _instance_identity(self) -> str | None: + """The 'you are navi on host X' block for the system prompt.""" + import socket + + try: + from navi.identity import load_identity + + ident = load_identity(_config.settings.instance_file) + except Exception: + return None + lines = [ + f"You are navi '{ident.name}' (instance {ident.instance_id[:8]}) " + f"running on host {socket.gethostname()}.", + "Other navi instances in the swarm know this machine by that name — " + "identify yourself as '" + ident.name + "' when talking to remote peers.", + ] + return "\n".join(lines) + + async def _memory_msg(self, user_id: str | None = None) -> "Message | None": + if self._memory is None: + return None + summary = await self._memory.get_summary(user_id=user_id) + if not summary: + return None + return Message( + role="system", + content=f"## What I remember about the user\n\n{summary}", + ) + + async def _memory_facts_msg( + self, + user_message: str, + user_id: str | None = None, + injected_ids: set[str] | None = None, + scope_boundary_enabled: bool = False, + session_cwd: str | None = None, + ) -> "Message | None": + if self._memory is None: + log.info("memory_facts_msg.skip", reason="no_memory_store") + return None + msg = user_message.strip() + if len(msg) <= 20: + log.info("memory_facts_msg.skip", reason="too_short", length=len(msg)) + return None + words = [w for w in msg.split() if any(c.isalpha() for c in w)] + if len(words) < 2: + log.info("memory_facts_msg.skip", reason="too_few_words", words=len(words)) + return None + if len(msg) < 50: + limit = 1 + elif len(msg) <= 150: + limit = 2 + else: + limit = 3 + log.info("memory_facts_msg.search", query=msg[:60], limit=limit, user_id=user_id) + facts = await self._memory.search_facts(msg, user_id=user_id, limit=limit) + log.info("memory_facts_msg.hits", count=len(facts) if facts else 0) + if not facts: + return None + if injected_ids is not None: + new_facts = [f for f in facts if f.get("id") not in injected_ids] + skipped = len(facts) - len(new_facts) + log.info("memory_facts_msg.dedup", new=len(new_facts), skipped=skipped) + if not new_facts: + return None + for f in new_facts: + injected_ids.add(f["id"]) + else: + new_facts = facts + + # Bounded autonomy: drop facts whose value is an absolute path outside the + # session cwd (e.g. "project_root → /home/.../navi-1" while the user works + # in another project) so they cannot misdirect the agent. Facts are kept + # when working inside that path (then they are correct). Gated by the + # profile flag so free-flight stays reproducible by toggling it off. + if scope_boundary_enabled and session_cwd: + kept = [f for f in new_facts if not _fact_value_is_abs_path_outside(str(f.get("value", "")), session_cwd)] + dropped = len(new_facts) - len(kept) + if dropped: + log.info("memory_facts_msg.scope_filtered", dropped=dropped, session_cwd=session_cwd) + new_facts = kept + if not new_facts: + return None + + lines = ["Known facts about the user:"] + for f in new_facts: + conf = f.get("confidence", 70) + age = _fact_age(str(f.get("updated_at", ""))) + age_note = f", updated {age}" if age else "" + lines.append( + f"• [{f['category']}] {f['key']} → {f['value']} " + f"(confidence: {conf}%{age_note})" + ) + result = "\n".join(lines) + log.info("memory_facts_msg.injected", lines=len(new_facts), length=len(result)) + return Message(role="system", content=result) + + async def _memory_recall_msg( + self, + session, + user_message: str, + user_id: str | None = None, + injected_ids: list[str] | None = None, + ) -> "Message | None": + """Restore a few verbatim excerpts that compression dropped. + + Only fires once the session actually carries a summary: before that, the + history is still in the context and a search would just duplicate it. + The query is built from the window of recent turns, not the last message, + because "and what about the other one?" is only meaningful next to what + came before it. + + This is the recall path that does not wait for the model to call a tool. + The model can still call `memory_search` for anything wider. + """ + if self._memory_search is None or not _config.settings.memory_recall_enabled: + return None + context = getattr(session, "context", None) or [] + if not any(m.is_summary for m in context): + return None + session_id = getattr(session, "id", None) + query = _recall_query(context, user_message) + if not query: + return None + try: + hits = await self._memory_search.search( + query, + scope="raw", + k=_config.settings.memory_recall_k, + user_id=user_id, + session_id=session_id, + current_session_id=session_id, + summary_at=await self._memory_search.latest_summary_at(session_id), + ) + except Exception: + log.debug("memory_recall.search_failed", exc_info=True) + return None + + selected = [] + for hit in hits: + if hit.source_type != "raw" or not hit.chunk_id: + continue + if injected_ids is not None and hit.chunk_id in injected_ids: + continue + if hit.score is not None: + if hit.score < _config.settings.memory_recall_min_similarity: + continue + elif hit.terms_hit < 2: + # No vector score (lexical fallback): demand a real keyword + # overlap, otherwise a one-word coincidence injects a turn. + continue + if hit.superseded_by: + continue + selected.append(hit) + if len(selected) >= _config.settings.memory_recall_k: + break + if not selected: + return None + if injected_ids is not None: + for hit in selected: + injected_ids.append(hit.chunk_id) + # The ledger lives in `session_metadata`, which is JSON-encoded into + # the sessions row on every save — an unbounded list grows the row + # for the life of the session. Older ids are dropped; a chunk that + # falls out after ~100 injections may be recalled again, which is + # a much smaller cost than a row that never stops growing. + if len(injected_ids) > _INJECTED_IDS_LIMIT: + del injected_ids[: len(injected_ids) - _INJECTED_IDS_LIMIT] + + lines = [ + "## Recovered from earlier in this conversation", + "These are verbatim excerpts of turns that were compressed out of context. " + "The summary is the curated version; these carry the exact details — names, " + "commands, values, links.", + ] + for hit in selected: + turn = ( + f"turn {hit.start_seq}" if hit.start_seq == hit.end_seq + else f"turns {hit.start_seq}–{hit.end_seq}" + ) + if hit.end_seq < 0: + turn = "turn unknown" + when = hit.timestamp.strftime("%Y-%m-%d %H:%M") if hit.timestamp else "unknown time" + lines.append(f"\n• session {hit.session_id}, {turn}, {when}") + lines.append(hit.text[:1500].strip()) + result = "\n".join(lines) + log.info( + "memory_recall.injected", + session_id=session_id, + chunks=len(selected), + length=len(result), + ) + return Message(role="system", content=result) + + async def _collect_context_injections(self, profile: "AgentProfile") -> list[Message]: + if self._cp_registry is None: + return [] + out: list[Message] = [] + for provider in self._cp_registry.all(): + if not provider.global_provider and provider.name not in profile.context_providers: + continue + try: + text = await provider.get_context() + if text: + out.append(Message(role="system", content=text)) + except Exception: + pass + return out + + async def _build_goal_anchor(self, session_id: str, user_message: str) -> Message: + lines = [ + "[Goal anchor]", + f"Original request: {user_message}", + ] + todo_lines = await render_todo_lines(session_id) + if todo_lines: + lines.append("Current todo:") + lines.extend(todo_lines) + lines.append("Stay on track — complete the remaining pending/in_progress steps.") + lines.append( + "Use 1-based todo indexes. Mark completed steps done only after verification, with validation." + ) + lines.append( + "Before final response, update todo for every completed step, including the final one." + ) + return Message(role="system", content="\n".join(lines)) + + def _scope_boundary_msg(self, profile: "AgentProfile") -> Message | None: + """Bounded-autonomy rule: keep the agent within the requested scope. + + Returns None when the profile has not opted in (legacy free-flight behavior). + """ + if not profile.scope_boundary_enabled: + return None + return Message( + role="system", + content=( + "[Scope boundary]\n" + "Act strictly within the user's literally requested scope. Do NOT expand to " + "sibling or parent directories/projects beyond what was asked, and do NOT start " + "executing discovered TODO lists, roadmaps, milestones, or backlog docs unless " + "the user explicitly asks you to. If you encounter such docs while exploring, " + "report what you found — do not act on it. Autonomy means finishing the requested " + "task well, not enlarging it." + ), + ) + + def _security_policy_msg(self) -> Message | None: + """Build a dynamic security policy system message based on user role.""" + from navi.tools._internal.base import ( + current_user_id as _uid_var, + ) + from navi.tools._internal.base import ( + current_user_role as _role_var, + ) + + user_id = _uid_var.get(None) + role = _role_var.get() + if role == "admin": + return Message( + role="system", + content=( + "[Security policy]\n" + "Role: admin\n" + "Full system access. No restrictions apply.\n" + "You may use any tool, access any path, and execute any command." + ), + ) + if user_id: + allowed = _config.settings.terminal_user_allowed_commands_list + return Message( + role="system", + content=( + "[Security policy]\n" + f"Role: user (user_id={user_id})\n" + f"Filesystem areas: user_data/{user_id}/ and the current session directory " + "(its path is given in the session context above).\n" + "You MUST NOT attempt to access paths outside those two areas.\n" + f"Terminal allowed commands: {', '.join(allowed)}\n" + "You MUST NOT use terminal for: curl, wget, ssh, scp, sudo, system-wide destructive operations, " + "or any command not in the allowlist.\n" + "If a task requires admin privileges (e.g. system-wide changes, accessing another user's files, " + "installing packages globally), tell the user to contact an admin.\n" + "Always prefer filesystem tool over terminal for file operations." + ), + ) + # Legacy / single-user mode — no policy injected + return None + + def _user_context_msg(self) -> Message | None: + """Build a [User context] system message from current_user_info ContextVar.""" + from navi.tools._internal.base import ( + current_user_info as _uinfo_var, + ) + from navi.tools._internal.base import ( + current_user_role as _role_var, + ) + + info = _uinfo_var.get(None) + if not info: + return None + lines = ["[User context]"] + if info.get("display_name"): + lines.append(f"Name: {info['display_name']}") + if info.get("username"): + lines.append(f"Username: {info['username']}") + if info.get("first_name") or info.get("last_name"): + parts = [] + if info.get("first_name"): + parts.append(info["first_name"]) + if info.get("last_name"): + parts.append(info["last_name"]) + lines.append(f"Full name: {' '.join(parts)}") + if info.get("email"): + lines.append(f"Email: {info['email']}") + if info.get("phone"): + lines.append(f"Phone: {info['phone']}") + if info.get("birth_date"): + lines.append(f"Birth date: {info['birth_date']}") + if info.get("country") or info.get("city"): + parts = [] + if info.get("city"): + parts.append(info["city"]) + if info.get("country"): + parts.append(info["country"]) + lines.append(f"Location: {', '.join(parts)}") + if info.get("locale"): + lines.append(f"Locale: {info['locale']}") + lines.append(f"Role: {_role_var.get()}") + return Message(role="system", content="\n".join(lines)) + + def _mcp_context_msg(self, profile: "AgentProfile | None" = None) -> "Message | None": + """Build a system message naming the MCP servers this run can reach. + + One line per server — what it is for — and a pointer to the full text. The + instructions themselves used to be inlined here, which cost ~8 KB of prose + on *every* request of a heavy profile; they are now read on demand with + ``tool_manual("")``. + + A server whose tools were withheld for want of a personal key is still + named, followed by the reason. Listing it without one was the worst of + both: the model knew the server existed, had none of its tools, and had + nothing to tell the user. + """ + if not self._mcp_manager: + return None + if profile is not None and not profile.get_agent_tools().mcp: + return None + from navi.tools._internal.base import current_keyless_mcp_servers as _keyless_var + + server_names = set(profile.get_agent_tools().mcp.keys()) if profile is not None else None + summaries = self._mcp_manager.get_summaries(server_names) + if not summaries: + return None + keyless = _keyless_var.get() + lines = [ + "[MCP servers — external knowledge sources]", + "Their tools are named mcp____. Before using a server's tools, " + 'and whenever it is unclear which of them fits, read its full instructions ' + 'with tool_manual("").', + ] + for name in sorted(summaries): + lines.append("") + lines.append(f"## {name} — {summaries[name]}") + if name in keyless: + lines.append( + " (tools withheld: this server runs on a personal MCP key and none " + "is saved for this account — do not retry it or call tool_manual for " + "it; tell the user to add the key in Settings → MCP keys.)" + ) + return Message(role="system", content="\n".join(lines)) + + def _truncate_oversized(self, conv: list[Message]) -> list[Message]: + """Per-build pass; see ``truncate_oversized`` for the rule.""" + return truncate_oversized(conv) + + def build( + self, + session_context: list[Message], + profile: "AgentProfile", + mem: "Message | None", + iteration: int | None = None, + max_iterations: int | None = None, + extra_system: list[Message] | None = None, + session_id: str | None = None, + session_metadata: dict | None = None, + ) -> list[Message]: + system_prompt = self.build_system_prompt(profile) + if session_id: + system_prompt += ( + f"\n\n---\n\n" + f"[Session context]\n" + f"Session ID: {session_id}\n" + f"Session files directory: {_config.settings.session_files_dir}/{session_id}/\n" + f"When writing files the user should see, always use the session directory path above.\n" + f"When calling MCP navi-3d tools (compile_scad, render_stl, lint_scad), pass ONLY the filename " + f"(e.g. 'falcon9_rocket.scad') for source_path and output_path. Do NOT include the session_id or " + f"the session_files directory in those paths — the MCP server resolves them automatically." + ) + cwd = (session_metadata or {}).get("cwd") + if cwd: + system_prompt += ( + f"\n\n---\n\n" + f"[Working directory]\n" + f"The user launched Navi Code from: {cwd}\n" + f"Treat this directory as the project root. Resolve relative paths against it " + f"unless the user provides an absolute path. For filesystem/terminal/code_exec operations, " + f"prefer this directory when no explicit working_dir is given." + ) + # The standing reaction rules for this run. They arrive through a + # ContextVar rather than the transcript so that they shape every request + # of the run without being stored as a message the user would read. + from navi.tools._internal.base import ( + current_reaction_instructions as _reaction_var, + ) + + reaction_instructions = (_reaction_var.get(None) or "").strip() + if reaction_instructions: + system_prompt += ( + f"\n\n---\n\n" + f"[Reaction session]\n" + f"This run answers a platform event on the user's behalf. Their standing " + f"instructions for how to react:\n\n{reaction_instructions}" + ) + system_msg = Message(role="system", content=system_prompt) + # System-role history is normally infrastructure-only (hidden from the + # LLM), but a few sources are addressed to the model — see the constant. + conv = [ + m for m in session_context + if m.role != "system" + or (m.metadata or {}).get("source") in _LLM_SYSTEM_SOURCES + ] + conv = self._truncate_oversized(conv) + result: list[Message] = [system_msg] + if mem: + result.append(mem) + + # Inject user profile context for multi-user mode + user_ctx = self._user_context_msg() + if user_ctx: + result.append(user_ctx) + + # Inject security policy for multi-user mode + policy = self._security_policy_msg() + if policy: + result.append(policy) + + # Inject bounded-autonomy scope boundary when the profile opts in + boundary = self._scope_boundary_msg(profile) + if boundary: + result.append(boundary) + + # Inject MCP server instructions into context + mcp_msg = self._mcp_context_msg(profile) + if mcp_msg: + result.append(mcp_msg) + + if extra_system: + result.extend(extra_system) + result.extend(conv) + + if ( + profile.iteration_budget_enabled + and iteration is not None + and max_iterations is not None + ): + remaining_after_this = max_iterations - iteration - 1 + if remaining_after_this <= 2: + urgency = ( + f" CRITICAL: only {remaining_after_this} iteration(s) left after this one — finish or produce " + "a partial result now, do not start new subtasks." + ) + elif remaining_after_this <= 5: + urgency = ( + " Low iteration budget: complete the current step, continue necessary " + "verification/publishing, and avoid starting unrelated subtasks." + ) + else: + urgency = "" + result.append( + Message( + role="system", + content=( + f"[Iteration {iteration + 1}/{max_iterations} — " + f"{remaining_after_this} iteration(s) after this one.{urgency}]" + ), + ) + ) + + return result diff --git a/navi/core/context_builder/helpers.py b/navi/core/context_builder/helpers.py new file mode 100644 index 0000000..dec9025 --- /dev/null +++ b/navi/core/context_builder/helpers.py @@ -0,0 +1,255 @@ +"""Pure helpers for context assembly: truncation, freshness, prompt blocks.""" + +from datetime import datetime, timezone +from pathlib import Path +from typing import TYPE_CHECKING + +import navi.config as _config +from navi.llm.base import Message + + +def _fact_age(updated_at: str) -> str: + """How old a fact is, so a stale one reads as stale. + + Nothing in memory expires on a clock — a fact is wrong when the world moves, + not when a timer fires — so the injection says how old it is and lets the + agent weigh it. A missing or unparseable timestamp simply says nothing. + """ + try: + then = datetime.fromisoformat(updated_at) + except (TypeError, ValueError): + return "" + if then.tzinfo is None: + then = then.replace(tzinfo=timezone.utc) + days = (datetime.now(timezone.utc) - then).days + if days <= 0: + return "today" + if days < 30: + return f"{days}d ago" + return f"{days // 30}mo ago" + + +# Roles whose content may be head/tail-truncated when a single message is too +# large for the context window. System/summary messages are already compact. +# User messages carry intent, so their cap is deliberately more generous than +# tool output's (see `_truncation_budget`) — but leaving them uncapped at all +# turned a single oversized paste into a dead session: the size check refused +# every subsequent call while compression had nothing it could split. +_TRUNCATABLE_ROLES = {"tool", "assistant", "user"} + +# A user message may use up to this fraction of the window before its middle is +# elided for the LLM view (the stored history keeps the full text). +_USER_MESSAGE_WINDOW_FRAC = 1 / 3 + + +def _truncation_budget(role: str) -> int: + """Per-message view cap in tokens for a truncatable role.""" + budget = _config.settings.context_message_token_budget + if budget <= 0: + budget = max(1, _config.settings.ollama_num_ctx // 6) + if role == "user": + budget = max( + budget, + max(1, int(_config.settings.ollama_num_ctx * _USER_MESSAGE_WINDOW_FRAC)), + ) + return budget + +# Which stored system-role messages are infrastructure (and therefore dropped +# from the LLM view) and which are addressed to the model. `task_note` carries +# detached tool results; `synapse_event` is the visible record of the event a +# Synapse reaction is answering — the model must see the same request the user +# sees. Anything else in the transcript with role="system" is bookkeeping. +_LLM_SYSTEM_SOURCES = {"task_note", "synapse_event"} + + +def _truncate_content(content: str, budget_tokens: int, *, label: str = "truncated") -> str: + """Head/tail-truncate ``content`` to roughly ``budget_tokens`` tokens. + + Keeps the head and tail verbatim with a marker naming the elided middle, so + the model retains the beginning and end of a large result (usually the + most informative parts) without the whole thing blowing the context window. + """ + budget_chars = budget_tokens * 3 + if len(content) <= budget_chars: + return content + half = budget_chars // 2 + dropped_tokens = (len(content) - 2 * half) // 3 + marker = f"\n[…{label} ~{dropped_tokens:,} tokens…]\n" + return content[:half] + marker + content[-half:] + + +def truncate_oversized(conv: list[Message]) -> list[Message]: + """Head/tail-truncate oversized messages for the LLM view. + + A single huge tool result (e.g. reading a large file) can alone exceed + the context window, and the turn/message-count-based compressor cannot + shrink a single message. This per-build pass caps each truncatable + message at its role budget — ``context_message_token_budget`` tokens + (0 = num_ctx // 6) for tool/assistant output, at least num_ctx // 3 for + user text — keeping head + tail so the model still sees the start and end. + Copies are returned — stored history (session.messages) is never + mutated, so the user still sees full content and reloads are unaffected. + + Module-level because the compressor reports the session's context size in + the same units: a raw context holding a truncated-away megabyte is not what + the model is sent, and counting it as such shows a scary 800k over a 128k + window. + """ + out: list[Message] = [] + for m in conv: + if m.role not in _TRUNCATABLE_ROLES: + out.append(m) + continue + budget = _truncation_budget(m.role) + if len(m.content or "") <= budget * 3: + out.append(m) + continue + label = "user message truncated" if m.role == "user" else "truncated" + out.append( + m.model_copy( + update={"content": _truncate_content(m.content or "", budget, label=label)} + ) + ) + return out + + +_INJECTED_IDS_LIMIT = 100 + + +def _recall_query(context: list, user_message: str, max_chars: int = 1200) -> str: + """A search query made of the recent turn window, not just the last message. + + The user's "and what about the port?" is unsearchable on its own; in the + company of the previous two exchanges it names the topic. Assistant replies + are included in shortened form because the concrete names (a file, a value, a + plan) usually appear there, not in the user's own wording. + """ + parts: list[str] = [] + turns = [m for m in context if m.role in ("user", "assistant") and not m.is_summary] + for msg in turns[-4:]: + text = (msg.content or "").strip() + if text: + parts.append(text[:300]) + tail = (user_message or "").strip() + if tail: + parts.append(tail[:300]) + query = "\n".join(parts) + return query[-max_chars:].strip() + + +def _fact_value_is_abs_path_outside(value: str, cwd: str) -> bool: + """True when a fact's value is an absolute filesystem path outside `cwd`. + + Used to keep context-dependent path facts (e.g. "project_root → /home/.../navi-1") + from being injected into sessions running in a different project, where they + would misdirect the agent. Relative values are left through (we cannot know + what they were recorded relative to). Non-path values always pass. + """ + v = str(value).strip() + if not v or not (v.startswith("/") or v.startswith("~")): + return False + try: + p = Path(v).expanduser().resolve() + base = Path(cwd).expanduser().resolve() + except Exception: + return True # unresolvable path — treat as out-of-scope, drop it + try: + p.relative_to(base) + return False # within the session cwd — keep + except ValueError: + return True # absolute path outside the cwd — drop + +if TYPE_CHECKING: + from navi.profiles.base import AgentProfile + + +def methodologies_block(profile: "AgentProfile") -> str | None: + """The "how to use the methodologies catalog" block for profiles that have it. + + ``skills`` and ``protocol`` are useless as bare tool schemas: the model meets + two unfamiliar tools with no reason to reach for them, and probes them + instead of working. The rules that say when to search the catalog, how to run + a protocol and when to write an entry belong in the system prompt, once, + instead of seven profile prompts drifting apart. + """ + scope = getattr(getattr(profile, "tools", None), "agent", None) + native = list(getattr(scope, "native", None) or []) + has_skills = "skills" in native + has_protocol = "protocol" in native + if not (has_skills or has_protocol): + return None + + tools = " and ".join(f"`{t}`" for t in ("skills", "protocol") if t in native) + manuals = " / ".join( + f'`tool_manual("{t}")`' for t in ("skills", "protocol") if t in native + ) + intro = ( + f"{tools} are this Navi's catalog of *how* to do things: what it has already " + f"learned and agreed on. To read up on them use {manuals} — do not call them " + f"with a guess to see what happens." + ) + lines = [ + "## Methodologies — skills and protocols", + intro, + ] + if has_skills: + lines.append( + '- **Recurring work: search first.** "как обычно", "как мы делали", a standard ' + 'shutdown, a deployment, a routine document — call `skills(op="search")` once ' + "before improvising, then `load` the entry that fits. One call replaces a whole " + "rediscovery. Not for a one-off question, never twice for the same query." + ) + lines.append( + "- **A skill is advice** — follow it as far as it fits the task, and say where " + "you departed from it." + ) + if has_protocol: + lines.append( + '- **A protocol is a commitment.** Run it with `protocol(op="start")`, never by ' + "retyping its steps into the todo. A protocol whose triggers match the request is " + "the user's pre-agreed procedure: say you are running it and start. Each injected " + "step is marked done only after its check passes; the run ends with `finish` plus " + "a report on the post-checks, or with `fail`/`cancel` and the reason. Never " + "improvise around a step — that is the error a protocol exists to prevent." + ) + if has_skills: + lines.append( + "- **Record what proved itself.** An approach that worked and will recur is one " + '`skills(op="create")` and a line to the user. A protocol is never created ' + 'directly: `skills(op="propose")` it, say what it does, and wait for the user\'s ' + "decision." + ) + return "\n\n".join(lines) + + +def profiles_block(current_id: str, others: list) -> str: + """The "Available profiles" block: the cross-profile map, plus how to act on it. + + The block is built in one place because the admin prompt-debug view renders + it too — two copies used to tell the model different things about switching. + The rules themselves live in the persona; this is the short reminder beside + the list, where the model is about to make the choice. + """ + lines = ["## Available profiles", f"Current: **{current_id}**"] + for profile in others: + desc = profile.short_description or profile.description + lines.append(f"· {profile.id}: {desc}") + lines.append( + "→ Decide on your own — do not ask permission. The task you were given belongs " + "to another profile's domain → `switch_profile`, called alone, then say which " + "profile is active and why. Another profile's tools are needed only for a side " + "piece of your own task → `spawn_agent(profile_id=…)` and keep this session's " + "profile. Details: `list_profiles`; what a candidate can actually call: " + "`list_tools(profile_id=…)`." + ) + return "\n".join(lines) + + +async def render_todo_lines(session_id: str) -> list[str]: + """Return a list of formatted todo lines for goal anchoring.""" + try: + from navi.tools.todo import render_todo_lines as _rtl + + return await _rtl(session_id) + except Exception: + return [] diff --git a/tests/unit/core/test_agent.py b/tests/unit/core/test_agent.py index cf2ff47..62aa406 100644 --- a/tests/unit/core/test_agent.py +++ b/tests/unit/core/test_agent.py @@ -712,15 +712,17 @@ @pytest.mark.asyncio async def test_compact_raises_when_compression_disabled(self, agent, session_with_history, monkeypatch): - import navi.core.agent as agent_mod import navi.config as config + from navi.core.agent import compression as compression_mod + from navi.core.agent import runtime as runtime_mod - # Settings is frozen — swap the whole object the agent module sees. + # Settings is frozen — swap the whole object the mixins see. disabled = config.Settings( database_url=config.settings.database_url, context_compression_enabled=False, ) - monkeypatch.setattr(agent_mod, "settings", disabled) + for module in (runtime_mod, compression_mod): + monkeypatch.setattr(module, "settings", disabled) with pytest.raises(NothingToCompactError): async for _ in agent.compact_stream(session_with_history.id): pass @@ -737,7 +739,7 @@ if would_compress says the partition cannot shrink the context — no misleading "compression" status with no work done. The gate is forced True here so we exercise the would_compress guard directly.""" - import navi.core.agent as agent_mod + from navi.core.agent import compression as compression_mod from navi.core.events import CompressionStarted # Small context that partition cannot shrink (would_compress False). @@ -747,7 +749,7 @@ # Force the token gate open; without this the small context would not # reach the would_compress check at all. - monkeypatch.setattr(agent_mod, "should_compress", lambda *a, **k: True) + monkeypatch.setattr(compression_mod, "should_compress", lambda *a, **k: True) events = [ ev @@ -772,7 +774,7 @@ A message carrying images could overshoot the window in a single step — the context was under the threshold the moment before and the images were the only thing over — so its cost is added to the estimate.""" - import navi.core.agent as agent_mod + from navi.core.agent import compression as compression_mod for i in range(4): session.context.append(Message(role="user", content=f"u{i}")) @@ -785,7 +787,7 @@ seen.append(tokens) return False - monkeypatch.setattr(agent_mod, "should_compress", _capture) + monkeypatch.setattr(compression_mod, "should_compress", _capture) profile = agent._profiles.get("test") baseline = agent._compressor.real_baseline_estimate( @@ -1149,15 +1151,16 @@ is computed from `settings.ollama_num_ctx` in the agent module, while the compressor's own thresholds read their own binding and stay untouched. """ - import navi.core.agent as agent_mod import navi.session_files as session_files_mod from navi.config import Settings + from navi.core.agent import compression, context, runtime, streaming, tools def _apply(**overrides): root = tmp_path / "sessions" patched = Settings(_env_file=None, session_files_dir=str(root), **overrides) monkeypatch.setattr(session_files_mod, "settings", patched) - monkeypatch.setattr(agent_mod, "settings", patched) + for module in (runtime, streaming, tools, compression, context): + monkeypatch.setattr(module, "settings", patched) return root return _apply @@ -1244,10 +1247,10 @@ """A directory that cannot be created must not take the message with it. The model is not told either: the images are in its context already, so the copy that failed is one it was never going to need. Operations hear it instead.""" - import navi.core.agent as agent_mod + from navi.core.agent import streaming as streaming_mod log = MagicMock() - monkeypatch.setattr(agent_mod, "log", log) + monkeypatch.setattr(streaming_mod, "log", log) session_files_dir = patch_agent_settings() backend = FakeLLMBackend(responses=["ok"]) agent._backends.register("ollama", backend) @@ -1452,11 +1455,11 @@ assert agent._window_for(profile) == 65000 def test_a_backend_without_routes_falls_back_to_the_global_setting(self, agent): - import navi.core.agent as agent_mod + from navi.core.agent import context as context_mod profile = make_profile("test", context_window=None) - assert agent._window_for(profile) == agent_mod.settings.ollama_num_ctx + assert agent._window_for(profile) == context_mod.settings.ollama_num_ctx class TestMemoryRecallInjection: diff --git a/tests/unit/core/test_memory_recall.py b/tests/unit/core/test_memory_recall.py index 39ee9c8..8eb0271 100644 --- a/tests/unit/core/test_memory_recall.py +++ b/tests/unit/core/test_memory_recall.py @@ -4,7 +4,7 @@ from types import SimpleNamespace import navi.config as _config -import navi.core.context_builder as _cb +import navi.core.context_builder.builder as _cb from navi.config import Settings from navi.core.context_builder import ContextBuilder, _recall_query from navi.llm.base import Message