# -*- coding: utf-8 -*- """Assemble system prompts from run workspace assets and toolkit context.""" from pathlib import Path from typing import Any from .agent_workspace import load_agent_workspace_config from backend.config.bootstrap_config import get_bootstrap_config_for_run from .skills_manager import SkillsManager from .workspace_manager import RunWorkspaceManager def _read_file_if_exists(path: Path) -> str: if not path.exists() or not path.is_file(): return "" return path.read_text(encoding="utf-8").strip() def _append_section(parts: list[str], title: str, content: str) -> None: content = content.strip() if content: parts.append(f"## {title}\n{content}") def _build_skill_metadata_summary(skills_manager: SkillsManager, config_name: str, agent_id: str) -> str: """Create a compact summary of active skills for prompt routing.""" metadata_items = skills_manager.list_active_skill_metadata(config_name, agent_id) if not metadata_items: return "" lines: list[str] = [ "You can use the following active skills. Prefer the most relevant one, then read its SKILL.md if needed for detailed workflow:", ] for item in metadata_items: parts = [f"- `{item.skill_name}`"] if item.description: parts.append(item.description) if item.version: parts.append(f"version: {item.version}") parts.append(f"path: {item.path}") lines.append(" | ".join(parts)) return "\n".join(lines) def build_agent_system_prompt( agent_id: str, config_name: str, toolkit: Any, ) -> str: """Build the final system prompt for an agent. Always reads fresh from disk — no caching. """ sections: list[str] = [] skills_manager = SkillsManager() asset_dir = skills_manager.get_agent_asset_dir(config_name, agent_id) asset_dir.mkdir(parents=True, exist_ok=True) workspace_manager = RunWorkspaceManager(project_root=skills_manager.project_root) required_files = ["SOUL.md", "PROFILE.md", "AGENTS.md", "POLICY.md", "MEMORY.md"] if not all((asset_dir / filename).exists() for filename in required_files): workspace_manager.ensure_agent_assets(config_name=config_name, agent_id=agent_id) agent_config = load_agent_workspace_config(asset_dir / "agent.yaml") bootstrap_config = get_bootstrap_config_for_run( skills_manager.project_root, config_name, ) _append_section( sections, "Bootstrap", bootstrap_config.prompt_body, ) prompt_files = agent_config.prompt_files or [ "SOUL.md", "PROFILE.md", "AGENTS.md", "POLICY.md", "MEMORY.md", ] included_files = set(prompt_files) title_map = { "SOUL.md": "Soul", "PROFILE.md": "Profile", "AGENTS.md": "Agent Guide", "POLICY.md": "Policy", "MEMORY.md": "Memory", } for filename in prompt_files: _append_section( sections, title_map.get(filename, filename), _read_file_if_exists(asset_dir / filename), ) if "POLICY.md" not in included_files: _append_section( sections, "Policy", _read_file_if_exists(asset_dir / "POLICY.md"), ) skill_prompt = toolkit.get_agent_skill_prompt() if skill_prompt: _append_section(sections, "Skills", str(skill_prompt)) metadata_summary = _build_skill_metadata_summary( skills_manager=skills_manager, config_name=config_name, agent_id=agent_id, ) if metadata_summary: _append_section(sections, "Active Skill Catalog", metadata_summary) activated_notes = toolkit.get_activated_notes() if activated_notes: _append_section(sections, "Tool Usage Notes", str(activated_notes)) return "\n\n".join(section for section in sections if section.strip()) def clear_prompt_factory_cache() -> None: """No-op retained for compatibility with runtime reload hooks."""