"""`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 <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 _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_read/filesystem_write 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("<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]:
"""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
