evotraders/backend/agents/prompt_factory.py

# -*- coding: utf-8 -*-
"""Assemble system prompts from base prompts, run assets, and toolkit context."""

from pathlib import Path
from typing import Any, Optional

from .agent_workspace import load_agent_workspace_config
from backend.config.bootstrap_config import get_bootstrap_config_for_run
from .prompt_loader import PromptLoader
from .skills_manager import SkillsManager

_prompt_loader = PromptLoader()


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,
    analyst_type: Optional[str] = None,
) -> str:
    """Build the final system prompt for an agent.

    Always reads fresh from disk — no caching.
    """
    # Clear any cached templates before building (CoPaw-style, no caching)
    _prompt_loader.clear_cache()

    sections: list[str] = []
    canonical_agent_id = (
        "portfolio_manager"
        if "portfolio" in agent_id
        else "risk_manager"
        if "risk" in agent_id and not analyst_type
        else agent_id
    )

    if analyst_type:
        personas_config = _prompt_loader.load_yaml_config(
            "analyst",
            "personas",
        )
        persona = personas_config.get(analyst_type, {})
        focus_text = "\n".join(
            f"- {item}" for item in persona.get("focus", [])
        )
        description = persona.get("description", "").strip()
        base_prompt = _prompt_loader.load_prompt(
            "analyst",
            "system",
            variables={
                "analyst_type": persona.get("name", analyst_type),
                "focus": focus_text,
                "description": description,
            },
        )
    elif agent_id == "portfolio_manager":
        base_prompt = _prompt_loader.load_prompt(
            "portfolio_manager",
            "system",
        )
    elif canonical_agent_id == "portfolio_manager":
        base_prompt = _prompt_loader.load_prompt(
            "portfolio_manager",
            "system",
        )
    elif agent_id == "risk_manager":
        base_prompt = _prompt_loader.load_prompt(
            "risk_manager",
            "system",
        )
    elif canonical_agent_id == "risk_manager":
        base_prompt = _prompt_loader.load_prompt(
            "risk_manager",
            "system",
        )
    else:
        raise ValueError(f"Unsupported agent prompt build for: {agent_id}")

    sections.append(base_prompt.strip())

    skills_manager = SkillsManager()
    asset_dir = skills_manager.get_agent_asset_dir(config_name, agent_id)
    asset_dir.mkdir(parents=True, exist_ok=True)
    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",
        "HEARTBEAT.md": "Heartbeat",
        "ROLE.md": "Role",
        "STYLE.md": "Style",
    }
    for filename in prompt_files:
        _append_section(
            sections,
            title_map.get(filename, filename),
            _read_file_if_exists(asset_dir / filename),
        )

    if "ROLE.md" not in included_files:
        _append_section(
            sections,
            "Role",
            _read_file_if_exists(asset_dir / "ROLE.md"),
        )
    if "STYLE.md" not in included_files:
        _append_section(
            sections,
            "Style",
            _read_file_if_exists(asset_dir / "STYLE.md"),
        )
    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:
    """Clear cached prompt and YAML templates before hot reload."""
    _prompt_loader.clear_cache()