"""Versioned prompt files: `prompts//v.md` (spec §125 — prompts live in code, versions are recorded on every event/job). A prompt file is Markdown; everything after the optional front-matter comment is the *system* prompt. The user message is always a compact JSON context built by the caller, so prompts never interpolate untrusted text into instructions. """ from __future__ import annotations import re from dataclasses import dataclass from functools import lru_cache from pathlib import Path from companyatlas.config import settings _REPO_ROOT = Path(__file__).resolve().parents[4] _VERSION_RE = re.compile(r"^v(\d+)\.md$") def prompts_dir() -> Path: return Path(settings.prompts_dir) if settings.prompts_dir else _REPO_ROOT / "prompts" @dataclass(frozen=True, slots=True) class Prompt: task: str version: str system: str @property def ref(self) -> str: return f"{self.task}/{self.version}" @lru_cache(maxsize=64) def load_prompt(task: str, version: str | None = None) -> Prompt: """Load `prompts//.md` (latest `vN` when version is None). Raises FileNotFoundError when missing.""" folder = prompts_dir() / task if version is None: candidates = sorted(((int(m.group(1)), p) for p in folder.glob("v*.md") if (m := _VERSION_RE.match(p.name))), reverse=True) if not candidates: raise FileNotFoundError(f"no prompt versions in {folder}") path = candidates[0][1] version = path.stem else: path = folder / f"{version}.md" text = path.read_text(encoding="utf-8") text = re.sub(r"\A\s*", "", text, flags=re.DOTALL) # optional front-matter comment return Prompt(task=task, version=version, system=text.strip()) def available_prompts() -> dict[str, list[str]]: root = prompts_dir() if not root.exists(): return {} return {d.name: sorted(p.stem for p in d.glob("v*.md")) for d in root.iterdir() if d.is_dir()} __all__ = ["Prompt", "available_prompts", "load_prompt", "prompts_dir"]