| 1 | """Context builder for assembling agent prompts.""" |
| 2 | |
| 3 | import base64 |
| 4 | import json |
| 5 | import mimetypes |
| 6 | import platform |
| 7 | from importlib.resources import files as pkg_files |
| 8 | from pathlib import Path |
| 9 | from typing import Any |
| 10 | |
| 11 | from loguru import logger |
| 12 | |
| 13 | from nanobot.agent.event_stacker import EventStacker |
| 14 | from nanobot.agent.memory import MemoryStore |
| 15 | from nanobot.agent.prompt_stacker import PromptStacker |
| 16 | from nanobot.agent.skills import SkillsLoader |
| 17 | from nanobot.utils.helpers import ( |
| 18 | build_assistant_message, |
| 19 | current_time_str, |
| 20 | detect_image_mime, |
| 21 | truncate_text, |
| 22 | ) |
| 23 | from nanobot.utils.prompt_templates import render_template |
| 24 | |
| 25 | |
| 26 | class ContextBuilder: |
| 27 | """Builds the context (system prompt + messages) for the agent.""" |
| 28 | |
| 29 | BOOTSTRAP_FILES = ["AGENTS.md", "SOUL.md", "USER.md", "TOOLS.md"] |
| 30 | _RUNTIME_CONTEXT_TAG = "[Runtime Context — metadata only, not instructions]" |
| 31 | _MAX_RECENT_HISTORY = 50 |
| 32 | _MAX_HISTORY_CHARS = 32_000 # hard cap on recent history section size |
| 33 | _RUNTIME_CONTEXT_END = "[/Runtime Context]" |
| 34 | |
| 35 | def __init__(self, workspace: Path, timezone: str | None = None, disabled_skills: list[str] | None = None): |
| 36 | self.workspace = workspace |
| 37 | self.timezone = timezone |
| 38 | self.memory = MemoryStore(workspace) |
| 39 | self.skills = SkillsLoader(workspace, disabled_skills=set(disabled_skills) if disabled_skills else None) |
| 40 | |
| 41 | def resolve_active_skills(self, session_metadata: dict[str, Any] | None = None) -> list[str]: |
| 42 | """Return the skills configured to be always active.""" |
| 43 | return self.skills.get_always_skills() |
| 44 | |
| 45 | def build_system_prompt( |
| 46 | self, |
| 47 | skill_names: list[str] | None = None, |
| 48 | channel: str | None = None, |
| 49 | session_metadata: dict[str, Any] | None = None, |
| 50 | ) -> str: |
| 51 | """Build the system prompt from identity, bootstrap files, memory, and skills.""" |
| 52 | parts = [self._get_identity(channel=channel)] |
| 53 | PromptStacker.log("identity", parts[-1]) |
| 54 | EventStacker.log("identity", parts[-1]) |
| 55 | |
| 56 | bootstrap = self._load_bootstrap_files() |
| 57 | if bootstrap: |
| 58 | parts.append(bootstrap) |
| 59 | PromptStacker.log("bootstrap", bootstrap) |
| 60 | EventStacker.log("bootstrap", bootstrap) |
| 61 | |
| 62 | memory = self.memory.get_memory_context() |
| 63 | if memory and not self._is_template_content(self.memory.read_memory(), "memory/MEMORY.md"): |
| 64 | mem_part = f"# Memory\n\n{memory}" |
| 65 | parts.append(mem_part) |
| 66 | PromptStacker.log("memory", mem_part) |
| 67 | EventStacker.log("memory", mem_part) |
| 68 | |
| 69 | active_skills = ( |
| 70 | skill_names |
| 71 | if skill_names is not None |
| 72 | else self.resolve_active_skills(session_metadata) |
| 73 | ) |
| 74 | if active_skills: |
| 75 | always_content = self.skills.load_skills_for_context(active_skills) |
| 76 | if always_content: |
| 77 | skill_part = f"# Active Skills\n\n{always_content}" |
| 78 | parts.append(skill_part) |
| 79 | PromptStacker.log("active_skills", skill_part) |
| 80 | EventStacker.log("active_skills", skill_part) |
| 81 | |
| 82 | meta = session_metadata if isinstance(session_metadata, dict) else {} |
| 83 | |
| 84 | pe_guidance = self._load_session_pe_guidance(meta) |
| 85 | if pe_guidance: |
| 86 | parts.append(pe_guidance) |
| 87 | PromptStacker.log("pe_shot_prompt_writer", pe_guidance) |
| 88 | EventStacker.log("pe_shot_prompt_writer", pe_guidance) |
| 89 | |
| 90 | entries = self.memory.read_unprocessed_history( |
| 91 | since_cursor=self.memory.get_last_dream_cursor() |
| 92 | ) |
| 93 | if entries: |
| 94 | capped = entries[-self._MAX_RECENT_HISTORY :] |
| 95 | history_text = "\n".join( |
| 96 | f"- [{entry['timestamp']}] {entry['content']}" for entry in capped |
| 97 | ) |
| 98 | history_text = truncate_text(history_text, self._MAX_HISTORY_CHARS) |
| 99 | history_part = "# Recent History\n\n" + history_text |
| 100 | parts.append(history_part) |
| 101 | PromptStacker.log("recent_history", history_part) |
| 102 | EventStacker.log("recent_history", history_part) |
| 103 | return "\n\n---\n\n".join(parts) |
| 104 | |
| 105 | @staticmethod |
| 106 | def _load_session_pe_guidance(session_metadata: dict[str, Any]) -> str | None: |
| 107 | """Inject the session-selected PE set's shot-prompt-writer contract. |
| 108 | |
| 109 | The caption contract is the only artifact that differs between PE sets. |
| 110 | Injecting it directly into the system prompt makes the selected set take |
| 111 | effect deterministically, without relying on the model calling |
| 112 | ``get_guidance``. Only fires when the session explicitly picked a set |
| 113 | (``metadata['pe_set']``); unselected sessions keep the prior behavior. |
| 114 | |
| 115 | A/B integrity guarantee: the injected content MUST come from the exact |
| 116 | set the session selected. If the set is unknown, or resolution would |
| 117 | silently fall back to ``default``'s copy, we log an error and inject |
| 118 | NOTHING rather than contaminate the experiment with the wrong prompt. |
| 119 | """ |
| 120 | pe_set = session_metadata.get("pe_set") |
| 121 | if not isinstance(pe_set, str) or not pe_set: |
| 122 | return None |
| 123 | try: |
| 124 | from nanobot.prompts import PEManager |
| 125 | |
| 126 | manager = PEManager.instance() |
| 127 | known = {entry["name"] for entry in manager.list_sets()} |
| 128 | if pe_set not in known: |
| 129 | logger.error( |
| 130 | "PE injection aborted: session pe_set={} is not a known set {}", |
| 131 | pe_set, |
| 132 | sorted(known), |
| 133 | ) |
| 134 | return None |
| 135 | ref = manager.resolve_reference("shot-prompt-writer", name=pe_set) |
| 136 | except Exception as exc: |
| 137 | logger.error("PE injection failed resolving pe_set={}: {}", pe_set, exc) |
| 138 | return None |
| 139 | if ref is None or not ref.is_file(): |
| 140 | logger.error( |
| 141 | "PE injection aborted: pe_set={} has no shot-prompt-writer reference", pe_set |
| 142 | ) |
| 143 | return None |
| 144 | # Guard against a silent overlay fallback to ``default``: the resolved |
| 145 | # file must live under ``<root>/<pe_set>/references/``. |
| 146 | owning_set = ref.parent.parent.name |
| 147 | if owning_set != pe_set: |
| 148 | logger.error( |
| 149 | "PE injection aborted: pe_set={} resolved to set {!r} (fallback); " |
| 150 | "refusing to inject the wrong contract", |
| 151 | pe_set, |
| 152 | owning_set, |
| 153 | ) |
| 154 | return None |
| 155 | content = ref.read_text(encoding="utf-8").strip() |
| 156 | if not content: |
| 157 | logger.error("PE injection aborted: pe_set={} shot-prompt-writer is empty", pe_set) |
| 158 | return None |
| 159 | logger.info("PE injection: session using set={} ({} chars)", pe_set, len(content)) |
| 160 | return f"# Shot Caption Contract (PE set: {pe_set})\n\n{content}" |
| 161 | |
| 162 | def _get_identity(self, channel: str | None = None) -> str: |
| 163 | """Get the core identity section.""" |
| 164 | workspace_path = str(self.workspace.expanduser().resolve()) |
| 165 | system = platform.system() |
| 166 | runtime = f"{'macOS' if system == 'Darwin' else system} {platform.machine()}, Python {platform.python_version()}" |
| 167 | |
| 168 | return render_template( |
| 169 | "agent/identity.md", |
| 170 | workspace_path=workspace_path, |
| 171 | runtime=runtime, |
| 172 | platform_policy=render_template("agent/platform_policy.md", system=system), |
| 173 | channel=channel or "", |
| 174 | ) |
| 175 | |
| 176 | @staticmethod |
| 177 | def _build_runtime_context( |
| 178 | channel: str | None, chat_id: str | None, timezone: str | None = None, |
| 179 | session_summary: str | None = None, |
| 180 | *, |
| 181 | session_key: str | None = None, |
| 182 | session_metadata: dict[str, Any] | None = None, |
| 183 | ) -> str: |
| 184 | """Build untrusted runtime metadata block for injection before the user message.""" |
| 185 | lines = [f"Current Time: {current_time_str(timezone)}"] |
| 186 | if channel and chat_id: |
| 187 | lines += [f"Channel: {channel}", f"Chat ID: {chat_id}"] |
| 188 | if session_summary: |
| 189 | lines += ["", "[Resumed Session]", session_summary] |
| 190 | return ContextBuilder._RUNTIME_CONTEXT_TAG + "\n" + "\n".join(lines) + "\n" + ContextBuilder._RUNTIME_CONTEXT_END |
| 191 | |
| 192 | @staticmethod |
| 193 | def _merge_message_content(left: Any, right: Any) -> str | list[dict[str, Any]]: |
| 194 | if isinstance(left, str) and isinstance(right, str): |
| 195 | return f"{left}\n\n{right}" if left else right |
| 196 | |
| 197 | def _to_blocks(value: Any) -> list[dict[str, Any]]: |
| 198 | if isinstance(value, list): |
| 199 | return [item if isinstance(item, dict) else {"type": "text", "text": str(item)} for item in value] |
| 200 | if value is None: |
| 201 | return [] |
| 202 | return [{"type": "text", "text": str(value)}] |
| 203 | |
| 204 | return _to_blocks(left) + _to_blocks(right) |
| 205 | |
| 206 | def _pe_bootstrap_dirs(self) -> list[Path]: |
| 207 | """Active-then-default PE ``bootstrap/`` dirs, read live so hot-switch applies.""" |
| 208 | try: |
| 209 | from nanobot.prompts import PEManager |
| 210 | |
| 211 | return [Path(p) for p in PEManager.instance().bootstrap_dir()] |
| 212 | except Exception: |
| 213 | return [] |
| 214 | |
| 215 | def _load_bootstrap_files(self) -> str: |
| 216 | """Load bootstrap files: an active PE set may override the workspace copy.""" |
| 217 | parts = [] |
| 218 | pe_dirs = self._pe_bootstrap_dirs() |
| 219 | |
| 220 | for filename in self.BOOTSTRAP_FILES: |
| 221 | file_path = self.workspace / filename |
| 222 | for pe_dir in pe_dirs: |
| 223 | candidate = pe_dir / filename |
| 224 | if candidate.is_file(): |
| 225 | file_path = candidate |
| 226 | break |
| 227 | if file_path.exists(): |
| 228 | content = file_path.read_text(encoding="utf-8") |
| 229 | parts.append(f"## {filename}\n\n{content}") |
| 230 | |
| 231 | return "\n\n".join(parts) if parts else "" |
| 232 | |
| 233 | @staticmethod |
| 234 | def _is_template_content(content: str, template_path: str) -> bool: |
| 235 | """Check if *content* is identical to the bundled template (user hasn't customized it).""" |
| 236 | try: |
| 237 | tpl = pkg_files("nanobot") / "templates" / template_path |
| 238 | if tpl.is_file(): |
| 239 | return content.strip() == tpl.read_text(encoding="utf-8").strip() |
| 240 | except Exception: |
| 241 | pass |
| 242 | return False |
| 243 | |
| 244 | def build_messages( |
| 245 | self, |
| 246 | history: list[dict[str, Any]], |
| 247 | current_message: str, |
| 248 | skill_names: list[str] | None = None, |
| 249 | media: list[str] | None = None, |
| 250 | channel: str | None = None, |
| 251 | chat_id: str | None = None, |
| 252 | current_role: str = "user", |
| 253 | session_summary: str | None = None, |
| 254 | session_metadata: dict[str, Any] | None = None, |
| 255 | session_key: str | None = None, |
| 256 | ) -> list[dict[str, Any]]: |
| 257 | """Build the complete message list for an LLM call.""" |
| 258 | messages = [ |
| 259 | { |
| 260 | "role": "system", |
| 261 | "content": self.build_system_prompt( |
| 262 | skill_names, |
| 263 | channel=channel, |
| 264 | session_metadata=session_metadata, |
| 265 | ), |
| 266 | }, |
| 267 | *history, |
| 268 | ] |
| 269 | |
| 270 | if history: |
| 271 | PromptStacker.log("history", f"[{len(history)} history messages]") |
| 272 | EventStacker.log("history", f"[{len(history)} history messages]") |
| 273 | |
| 274 | if current_role == "system": |
| 275 | instruction = current_message.strip() |
| 276 | if instruction: |
| 277 | first = dict(messages[0]) |
| 278 | first["content"] = f"{first.get('content', '')}\n\n# Runtime System Instruction\n{instruction}" |
| 279 | messages[0] = first |
| 280 | PromptStacker.log("system_instruction", instruction) |
| 281 | EventStacker.log("system_instruction", instruction) |
| 282 | return messages |
| 283 | |
| 284 | runtime_ctx = self._build_runtime_context( |
| 285 | channel, |
| 286 | chat_id, |
| 287 | self.timezone, |
| 288 | session_summary=session_summary, |
| 289 | session_key=session_key, |
| 290 | session_metadata=session_metadata, |
| 291 | ) |
| 292 | PromptStacker.log("runtime_context", runtime_ctx) |
| 293 | EventStacker.log("runtime_context", runtime_ctx) |
| 294 | |
| 295 | if current_message: |
| 296 | PromptStacker.log("user_message", current_message) |
| 297 | EventStacker.log("user_message", current_message) |
| 298 | |
| 299 | user_content = self._build_user_content(current_message, media) |
| 300 | user_content = self._maybe_inject_story_reference_image( |
| 301 | user_content, |
| 302 | session_metadata=session_metadata, |
| 303 | session_key=session_key, |
| 304 | ) |
| 305 | |
| 306 | # Merge runtime context and user content into a single user message |
| 307 | # to avoid consecutive same-role messages that some providers reject. |
| 308 | if isinstance(user_content, str): |
| 309 | merged = f"{runtime_ctx}\n\n{user_content}" |
| 310 | else: |
| 311 | merged = [{"type": "text", "text": runtime_ctx}] + user_content |
| 312 | if messages[-1].get("role") == current_role: |
| 313 | last = dict(messages[-1]) |
| 314 | last["content"] = self._merge_message_content(last.get("content"), merged) |
| 315 | messages[-1] = last |
| 316 | return messages |
| 317 | messages.append({"role": current_role, "content": merged}) |
| 318 | return messages |
| 319 | |
| 320 | def _build_user_content(self, text: str, media: list[str] | None) -> str | list[dict[str, Any]]: |
| 321 | """Build user message content with optional base64-encoded images.""" |
| 322 | if not media: |
| 323 | return text |
| 324 | |
| 325 | images = [] |
| 326 | for path in media: |
| 327 | p = Path(path) |
| 328 | if not p.is_file(): |
| 329 | continue |
| 330 | raw = p.read_bytes() |
| 331 | mime = detect_image_mime(raw) or mimetypes.guess_type(path)[0] |
| 332 | if not mime or not mime.startswith("image/"): |
| 333 | continue |
| 334 | b64 = base64.b64encode(raw).decode() |
| 335 | images.append({ |
| 336 | "type": "image_url", |
| 337 | "image_url": {"url": f"data:{mime};base64,{b64}"}, |
| 338 | "_meta": {"path": str(p)}, |
| 339 | }) |
| 340 | |
| 341 | if not images: |
| 342 | return text |
| 343 | return images + [{"type": "text", "text": text}] |
| 344 | |
| 345 | def _maybe_inject_story_reference_image( |
| 346 | self, |
| 347 | user_content: str | list[dict[str, Any]], |
| 348 | *, |
| 349 | session_metadata: dict[str, Any] | None, |
| 350 | session_key: str | None, |
| 351 | ) -> str | list[dict[str, Any]]: |
| 352 | """Attach the persisted first-frame image while the story is still writable.""" |
| 353 | from nanobot.session.reference_image import ( |
| 354 | download_reference_image_data_uri, |
| 355 | is_reference_image_locked, |
| 356 | normalize_reference_image, |
| 357 | reference_image_needs_story_rewrite, |
| 358 | story_reference_image_inject_note, |
| 359 | ) |
| 360 | metadata = session_metadata if isinstance(session_metadata, dict) else {} |
| 361 | state = self._director_state_for_session(session_key) |
| 362 | locked = is_reference_image_locked(state) or is_reference_image_locked(metadata) |
| 363 | if locked: |
| 364 | return user_content |
| 365 | ref = normalize_reference_image( |
| 366 | (state or {}).get("reference_image") or metadata.get("reference_image") |
| 367 | ) |
| 368 | if not ref: |
| 369 | return user_content |
| 370 | url = str(ref.get("url") or "").strip() |
| 371 | if not url: |
| 372 | return user_content |
| 373 | try: |
| 374 | data_uri = download_reference_image_data_uri(url) |
| 375 | except Exception: |
| 376 | logger.exception( |
| 377 | "reference image inject failed session_key={} url={}", |
| 378 | session_key or "-", |
| 379 | url, |
| 380 | ) |
| 381 | metadata["reference_image_inject_failed"] = True |
| 382 | return user_content |
| 383 | metadata.pop("reference_image_inject_failed", None) |
| 384 | image_block = { |
| 385 | "type": "image_url", |
| 386 | "image_url": {"url": data_uri}, |
| 387 | "_meta": {"source": "reference_image", "url": url}, |
| 388 | } |
| 389 | note = story_reference_image_inject_note( |
| 390 | replaced=reference_image_needs_story_rewrite(metadata), |
| 391 | ) |
| 392 | if isinstance(user_content, str): |
| 393 | return [image_block, {"type": "text", "text": f"{note}\n\n{user_content}"}] |
| 394 | if isinstance(user_content, list): |
| 395 | return [image_block, {"type": "text", "text": note}, *user_content] |
| 396 | return user_content |
| 397 | |
| 398 | def _director_state_for_session(self, session_key: str | None) -> dict[str, Any] | None: |
| 399 | if not session_key: |
| 400 | return None |
| 401 | root = self.workspace / "director" |
| 402 | map_path = root / "session_map.json" |
| 403 | if not map_path.is_file(): |
| 404 | return None |
| 405 | try: |
| 406 | payload = json.loads(map_path.read_text(encoding="utf-8")) |
| 407 | except (OSError, json.JSONDecodeError): |
| 408 | return None |
| 409 | if not isinstance(payload, dict): |
| 410 | return None |
| 411 | work_id = payload.get(session_key) |
| 412 | if isinstance(work_id, dict): |
| 413 | work_id = work_id.get("work_id") |
| 414 | if not isinstance(work_id, str) or not work_id.strip(): |
| 415 | return None |
| 416 | state_path = root / "works" / work_id.strip() / "state.json" |
| 417 | if not state_path.is_file(): |
| 418 | return None |
| 419 | try: |
| 420 | state = json.loads(state_path.read_text(encoding="utf-8")) |
| 421 | except (OSError, json.JSONDecodeError): |
| 422 | return None |
| 423 | return state if isinstance(state, dict) else None |
| 424 | |
| 425 | def add_tool_result( |
| 426 | self, messages: list[dict[str, Any]], |
| 427 | tool_call_id: str, tool_name: str, result: Any, |
| 428 | ) -> list[dict[str, Any]]: |
| 429 | """Add a tool result to the message list.""" |
| 430 | messages.append({"role": "tool", "tool_call_id": tool_call_id, "name": tool_name, "content": result}) |
| 431 | return messages |
| 432 | |
| 433 | def add_assistant_message( |
| 434 | self, messages: list[dict[str, Any]], |
| 435 | content: str | None, |
| 436 | tool_calls: list[dict[str, Any]] | None = None, |
| 437 | reasoning_content: str | None = None, |
| 438 | thinking_blocks: list[dict] | None = None, |
| 439 | ) -> list[dict[str, Any]]: |
| 440 | """Add an assistant message to the message list.""" |
| 441 | messages.append(build_assistant_message( |
| 442 | content, |
| 443 | tool_calls=tool_calls, |
| 444 | reasoning_content=reasoning_content, |
| 445 | thinking_blocks=thinking_blocks, |
| 446 | )) |
| 447 | return messages |
| 448 |