"""Context builder for assembling agent prompts.""" import base64 import mimetypes import platform from pathlib import Path from typing import Any, Mapping, Sequence, cast from nanobot.agent.memory import MemoryStore from nanobot.agent.skills import SkillsLoader from nanobot.agent.tools import image_generation as image_generation_tools from nanobot.agent.tools import mcp as mcp_tools from nanobot.agent.tools import sessions as session_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.runtime_context import ( RUNTIME_CONTEXT_END, RUNTIME_CONTEXT_MESSAGE_META, RUNTIME_CONTEXT_TAG, RuntimeContextBlock, append_runtime_context, ) from nanobot.utils.helpers import ( 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) | session_tools.session_extra(metadata) ) 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: for handler in ( image_generation_tools.handle_runtime_control, mcp_tools.handle_runtime_control, ): if await handler(state, msg, tools): return True return False class ContextBuilder: """Builds the context (system prompt + messages) for the agent.""" BOOTSTRAP_FILES = ["AGENTS.md", "SOUL.md", "USER.md"] _SKIPPABLE_DEFAULTS = {"AGENTS.md", "USER.md"} _RUNTIME_CONTEXT_TAG = RUNTIME_CONTEXT_TAG _MAX_RECENT_HISTORY = 50 _MAX_HISTORY_TOKENS = 8_000 # hard cap on recent history section size (tokens) _RUNTIME_CONTEXT_END = RUNTIME_CONTEXT_END 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, *, active_skill_names: Sequence[str] | None = None, channel: str | None = None, session_summary: str | None = None, workspace: Path | None = None, include_long_term_memory: bool = True, 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")) if include_long_term_memory: memory = self.memory.read_memory() if memory and not self._is_template_content(memory, "memory/MEMORY.md"): parts.append(f"# Memory\n\n## Long-term Memory\n{memory}") active_skills = self.skills.get_always_skills() active_skills.extend( name for name in (active_skill_names or ()) if name not in active_skills ) if active_skills: active_content = self.skills.load_skills_for_context(active_skills) if active_content: parts.append(f"# Active Skills\n\n{active_content}") skills_summary = self.skills.build_skills_summary(exclude=set(active_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()) agent_workspace_path = str(self.workspace.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, agent_workspace_path=agent_workspace_path, runtime=runtime, platform_policy=render_template("agent/platform_policy.md", system=system), channel=channel or "", ) @staticmethod def _merge_message_content(left: Any, right: Any) -> str | list[dict[str, Any]]: if isinstance(left, str) and isinstance(right, str): if not left: return right if not right: return left return f"{left}\n\n{right}" def _to_blocks(value: Any) -> list[dict[str, Any]]: if isinstance(value, list): return [ cast(dict[str, Any], item) if isinstance(item, dict) else {"type": "text", "text": str(item)} for item in cast(list[Any], 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 project instructions plus the agent's global profile files.""" parts: list[str] = [] project_root = workspace or self.workspace sources = [ ("AGENTS.md", project_root), ("SOUL.md", self.workspace), ("USER.md", self.workspace), ] for filename, root in sources: file_path = root / filename if file_path.exists(): content = file_path.read_text(encoding="utf-8") if filename == "SOUL.md" and self._is_template_content( content, "legacy/SOUL.md", ): content = load_bundled_template("SOUL.md") or content if not content.strip(): continue if filename in self._SKIPPABLE_DEFAULTS and self._is_template_content( content, filename ): continue 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, *, media: list[str] | None = None, channel: str | None = None, current_role: str = "user", session_summary: str | None = None, runtime_context_blocks: Sequence[RuntimeContextBlock] | None = None, workspace: Path | None = None, include_long_term_memory: bool = True, 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 active_skill_names = ( self.skills.get_explicitly_invoked_skills(current_message) if current_role == "user" else [] ) messages: list[dict[str, Any]] = [ { "role": "system", "content": self.build_system_prompt( active_skill_names=active_skill_names, channel=channel, session_summary=session_summary, workspace=root, include_long_term_memory=include_long_term_memory, include_memory_recent_history=include_memory_recent_history, session_key=session_key, unified_session=unified_session, ), }, *history, ] current = self.build_current_message( current_message, media=media, current_role=current_role, runtime_context_blocks=runtime_context_blocks, ) if messages[-1].get("role") == current_role: last = dict(messages[-1]) last["content"] = self._merge_message_content( last.get("content"), current.get("content"), ) current_meta = current.get("_meta") if current_role == "user" and isinstance(current_meta, dict): internal_meta = dict(last.get("_meta") or {}) internal_meta.update(cast(dict[str, Any], current_meta)) last["_meta"] = internal_meta messages[-1] = last return messages messages.append(current) return messages def build_current_message( self, current_message: str, *, media: list[str] | None = None, current_role: str = "user", runtime_context_blocks: Sequence[RuntimeContextBlock] | None = None, ) -> dict[str, Any]: """Build only the fresh turn message without merging it into history.""" content = self.build_user_content(current_message, image_paths=media) blocks = list(runtime_context_blocks or ()) if current_role == "user" else [] merged, runtime_context_meta = append_runtime_context(content, blocks) current: dict[str, Any] = {"role": current_role, "content": merged} if current_role == "user" and runtime_context_meta is not None: current["_meta"] = { RUNTIME_CONTEXT_MESSAGE_META: runtime_context_meta, } return current def build_user_content( self, text: str, image_paths: list[str] | None, ) -> str | list[dict[str, Any]]: """Build user message content from prefiltered image paths.""" if not image_paths: return text image_blocks: list[dict[str, Any]] = [] for path in image_paths: p = Path(path) if not p.is_file(): continue raw = p.read_bytes() # Re-detect from the bytes used for the request: the file may have # changed since attachment routing, and the data URL needs its MIME. 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() image_blocks.append({ "type": "image_url", "image_url": {"url": f"data:{mime};base64,{b64}"}, "_meta": {"path": str(p)}, }) if not image_blocks: return text return image_blocks + [{"type": "text", "text": text}]