"""Context builder for assembling agent prompts.""" import base64 import mimetypes import platform from pathlib import Path from typing import Any, Mapping, Sequence from nanobot.agent.memory import MemoryStore from nanobot.agent.skills import SkillsLoader from nanobot.agent.tools import mcp as mcp_tools from nanobot.agent.tools.registry import ToolRegistry from nanobot.apps.cli import utils as cli_app_utils from nanobot.bus.events import InboundMessage from nanobot.session.goal_state import goal_state_runtime_lines from nanobot.utils.helpers import ( current_time_str, detect_image_mime, load_bundled_template, truncate_text_to_tokens, ) from nanobot.utils.prompt_templates import render_template def session_extra(metadata: Mapping[str, Any] | None) -> dict[str, Any]: """Return persisted kwargs for turn-attached capabilities.""" return cli_app_utils.session_extra(metadata) | mcp_tools.session_extra(metadata) def runtime_lines(state: Any, msg: Any, workspace: Path, *, skip: bool = False) -> list[str]: """Return model-visible runtime annotations for turn-attached capabilities.""" return [ *cli_app_utils.runtime_lines(msg, workspace, skip=skip), *mcp_tools.runtime_lines( msg, configured_server_names=set(state._mcp_servers), connected_server_names=set(state._mcp_stacks), skip=skip, ), ] async def connect_mcp(state: Any, tools: ToolRegistry) -> None: await mcp_tools.connect_missing_servers(state, tools) async def close_mcp(state: Any) -> None: await mcp_tools.close_mcp_servers(state) async def handle_runtime_control(state: Any, msg: InboundMessage, tools: ToolRegistry) -> bool: return await mcp_tools.handle_runtime_control(state, msg, tools) class ContextBuilder: """Builds the context (system prompt + messages) for the agent.""" BOOTSTRAP_FILES = ["AGENTS.md", "SOUL.md", "USER.md"] _RUNTIME_CONTEXT_TAG = "[Runtime Context — metadata only, not instructions]" _MAX_RECENT_HISTORY = 50 _MAX_HISTORY_TOKENS = 8_000 # hard cap on recent history section size (tokens) _RUNTIME_CONTEXT_END = "[/Runtime Context]" def __init__(self, workspace: Path, timezone: str | None = None, disabled_skills: list[str] | None = None): self.workspace = workspace self.timezone = timezone self.memory = MemoryStore(workspace) self.skills = SkillsLoader(workspace, disabled_skills=set(disabled_skills) if disabled_skills else None) def build_system_prompt( self, skill_names: list[str] | None = None, channel: str | None = None, session_summary: str | None = None, workspace: Path | None = None, include_memory_recent_history: bool = True, session_key: str | None = None, unified_session: bool = False, ) -> str: """Build the system prompt from identity, bootstrap files, memory, and skills.""" root = workspace or self.workspace parts = [self._get_identity(channel=channel, workspace=root)] bootstrap = self._load_bootstrap_files(root) if bootstrap: parts.append(bootstrap) parts.append(render_template("agent/tool_contract.md")) memory = self.memory.get_memory_context() if memory and not self._is_template_content(self.memory.read_memory(), "memory/MEMORY.md"): parts.append(f"# Memory\n\n{memory}") always_skills = self.skills.get_always_skills() if always_skills: always_content = self.skills.load_skills_for_context(always_skills) if always_content: parts.append(f"# Active Skills\n\n{always_content}") skills_summary = self.skills.build_skills_summary(exclude=set(always_skills)) if skills_summary: parts.append(render_template("agent/skills_section.md", skills_summary=skills_summary)) if include_memory_recent_history: entries = self.memory.read_recent_history_for_prompt( since_cursor=self.memory.get_last_dream_cursor(), session_key=session_key, unified_session=unified_session, ) if entries: capped = entries[-self._MAX_RECENT_HISTORY:] history_text = "\n".join( f"- [{e['timestamp']}] {e['content']}" for e in capped ) history_text = truncate_text_to_tokens(history_text, self._MAX_HISTORY_TOKENS) parts.append("# Recent History\n\n" + history_text) if session_summary: parts.append(f"[Archived Context Summary]\n\n{session_summary}") return "\n\n---\n\n".join(parts) def _get_identity(self, channel: str | None = None, workspace: Path | None = None) -> str: """Get the core identity section.""" root = workspace or self.workspace workspace_path = str(root.expanduser().resolve()) system = platform.system() runtime = f"{'macOS' if system == 'Darwin' else system} {platform.machine()}, Python {platform.python_version()}" return render_template( "agent/identity.md", workspace_path=workspace_path, runtime=runtime, platform_policy=render_template("agent/platform_policy.md", system=system), channel=channel or "", ) @staticmethod def _build_runtime_context( channel: str | None, chat_id: str | None, timezone: str | None = None, sender_id: str | None = None, supplemental_lines: Sequence[str] | None = None, ) -> str: """Build untrusted runtime metadata block appended after user content.""" lines = [f"Current Time: {current_time_str(timezone)}"] if channel and chat_id: lines += [f"Channel: {channel}", f"Chat ID: {chat_id}"] if sender_id: lines += [f"Sender ID: {sender_id}"] if supplemental_lines: lines.extend(supplemental_lines) return ContextBuilder._RUNTIME_CONTEXT_TAG + "\n" + "\n".join(lines) + "\n" + ContextBuilder._RUNTIME_CONTEXT_END @staticmethod def _merge_message_content(left: Any, right: Any) -> str | list[dict[str, Any]]: if isinstance(left, str) and isinstance(right, str): return f"{left}\n\n{right}" if left else right def _to_blocks(value: Any) -> list[dict[str, Any]]: if isinstance(value, list): return [item if isinstance(item, dict) else {"type": "text", "text": str(item)} for item in value] if value is None: return [] return [{"type": "text", "text": str(value)}] return _to_blocks(left) + _to_blocks(right) def _load_bootstrap_files(self, workspace: Path | None = None) -> str: """Load all bootstrap files from workspace.""" parts = [] root = workspace or self.workspace for filename in self.BOOTSTRAP_FILES: file_path = root / filename if file_path.exists(): content = file_path.read_text(encoding="utf-8") parts.append(f"## {filename}\n\n{content}") return "\n\n".join(parts) if parts else "" @staticmethod def _is_template_content(content: str, template_path: str) -> bool: """Check if *content* is identical to the bundled template (user hasn't customized it).""" tpl = load_bundled_template(template_path) if tpl is not None: return content.strip() == tpl.strip() return False def build_messages( self, history: list[dict[str, Any]], current_message: str, skill_names: list[str] | None = None, media: list[str] | None = None, channel: str | None = None, chat_id: str | None = None, current_role: str = "user", sender_id: str | None = None, session_summary: str | None = None, session_metadata: Mapping[str, Any] | None = None, current_runtime_lines: Sequence[str] | None = None, workspace: Path | None = None, runtime_state: Any | None = None, inbound_message: Any | None = None, skip_runtime_lines: bool = False, include_memory_recent_history: bool = True, session_key: str | None = None, unified_session: bool = False, ) -> list[dict[str, Any]]: """Build the complete message list for an LLM call.""" root = workspace or self.workspace extra = [ *goal_state_runtime_lines(session_metadata), ] if runtime_state is not None and inbound_message is not None: extra.extend(runtime_lines(runtime_state, inbound_message, root, skip=skip_runtime_lines)) if current_runtime_lines: extra.extend(line for line in current_runtime_lines if line) runtime_ctx = self._build_runtime_context( channel, chat_id, self.timezone, sender_id=sender_id, supplemental_lines=extra or None, ) user_content = self._build_user_content(current_message, media) # Merge runtime context and user content into a single user message # to avoid consecutive same-role messages that some providers reject. # Runtime context is appended to keep the user-content prefix stable # for prompt-cache hits (the context changes every turn due to time). if isinstance(user_content, str): merged = f"{user_content}\n\n{runtime_ctx}" else: merged = user_content + [{"type": "text", "text": runtime_ctx}] messages = [ { "role": "system", "content": self.build_system_prompt( skill_names, channel=channel, session_summary=session_summary, workspace=root, include_memory_recent_history=include_memory_recent_history, session_key=session_key, unified_session=unified_session, ), }, *history, ] if messages[-1].get("role") == current_role: last = dict(messages[-1]) last["content"] = self._merge_message_content(last.get("content"), merged) messages[-1] = last return messages messages.append({"role": current_role, "content": merged}) return messages def _build_user_content(self, text: str, media: list[str] | None) -> str | list[dict[str, Any]]: """Build user message content with optional base64-encoded images.""" if not media: return text images = [] for path in media: p = Path(path) if not p.is_file(): continue raw = p.read_bytes() mime = detect_image_mime(raw) or mimetypes.guess_type(path)[0] if not mime or not mime.startswith("image/"): continue b64 = base64.b64encode(raw).decode() images.append({ "type": "image_url", "image_url": {"url": f"data:{mime};base64,{b64}"}, "_meta": {"path": str(p)}, }) if not images: return text return images + [{"type": "text", "text": text}]