"""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. User messages carry intent and are never
# truncated; system/summary messages are already compact.
_TRUNCATABLE_ROLES = {"tool", "assistant"}
# 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) -> 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 tool 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[…truncated ~{dropped_tokens:,} tokens…]\n"
return content[:half] + marker + content[-half:]
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
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,
cp_registry: "ContextProviderRegistry | None" = None,
mcp_manager=None,
) -> None:
self._profiles = profile_registry
self._memory = memory_store
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)
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:
lines = [
"## Available profiles",
f"Current: **{profile.id}**",
]
for p in other:
desc = p.short_description or p.description
lines.append(f"· {p.id}: {desc}")
lines.append(
"→ Switch profiles on your own judgment — do not ask for permission. "
"When a task clearly fits another profile, call switch_profile immediately, "
"then inform the user which profile is now active and why. "
"Use list_profiles if you need details about a profile's capabilities."
)
parts.append("\n".join(lines))
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 <name> 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 _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("<server>")``.
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__<server>__<tool>. Before using a server's tools, "
'and whenever it is unclear which of them fits, read its full instructions '
'with tool_manual("<server>").',
]
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]:
"""Head/tail-truncate oversized tool/assistant 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 ``context_message_token_budget`` tokens (0 = num_ctx // 6),
keeping head + tail so the model still sees the start and end of the
result. Copies are returned — stored history (session.messages) is never
mutated, so the user still sees full content and reloads are unaffected.
"""
budget = _config.settings.context_message_token_budget
if budget <= 0:
budget = max(1, _config.settings.ollama_num_ctx // 6)
cap_chars = budget * 3
out: list[Message] = []
for m in conv:
if m.role in _TRUNCATABLE_ROLES and len(m.content or "") > cap_chars:
out.append(m.model_copy(update={"content": _truncate_content(m.content or "", budget)}))
else:
out.append(m)
return out
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
