games-development-ai/docs/agent-specs/2026-06-09-agent-loop-v1/orchestrator/prompts.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Prompt Registry 加载与变量渲染（D3 件）—— spec：HJ-AGENT-LOOP-EXEC-001 §7.1

铁律：
- prompt 不内嵌代码：一律从 contracts/prompts/（D1 交付）按 registry.yaml 读取后渲染，
  git 即事实源；不依赖未实现的 Java Loader；
- 渲染采用 {{input.xxx}} 占位符替换（与 01-safety 既有范本一致）；
- 渲染后若仍残留 {{input.*}} 占位符 → 立即报错（变量缺漏在联调时暴露，禁止把花括号发给 LLM）；
- frontmatter version 与 registry 注册 version 不一致 → 立即报错（Prompt 即契约，两处必须同步，
  承接「改 prompt 必升 version」四道闸纪律）。

D1↔D3 渲染变量约定（写给 D1 对齐；run_batch 按此传参）：
- config.clicker-designer：idea / template_schema / banned_list / findings（首轮 findings 传空串，
  schema 重出轮传 schema 错误文本——prompt 模板需容忍该段为空）；
- quality.adversary-review：idea / game_design（GameDesign 全文 JSON）；
- fix.design-revise：idea / template_schema / game_design / findings（P1 项 JSON）。
标准库限定：registry.yaml 用最小 YAML 子集解析（仅本注册表的「列表项+键值」形态），
解析异常带行号报错，绝不静默容错。
"""

import json
import os
import re

# 渲染后残留占位符的检测模式（任何 {{input.xxx}} 残留=变量缺漏）
_PLACEHOLDER_RE = re.compile(r"\{\{\s*input\.([A-Za-z0-9_]+)\s*\}\}")


class PromptError(Exception):
    """Prompt 契约违规（registry 解析失败/文件缺失/版本不一致/变量缺漏）。"""


def _strip_inline_comment(value):
    """剥行内注释（registry 值后的 ` # ...`）；引号包裹的值先去引号。本注册表值均为简单标量。"""
    out = []
    in_quote = None
    for ch in value:
        if in_quote:
            if ch == in_quote:
                in_quote = None
            out.append(ch)
        elif ch in ("'", '"'):
            in_quote = ch
            out.append(ch)
        elif ch == "#":
            break  # 注释起点（不在引号内）
        else:
            out.append(ch)
    s = "".join(out).strip()
    if len(s) >= 2 and s[0] == s[-1] and s[0] in ("'", '"'):
        s = s[1:-1]
    return s


def load_registry(registry_path):
    """
    解析 contracts/prompts/registry.yaml 的最小子集：顶层 `prompts:` 列表，
    每项为 `- id: xxx` 起头的键值块。返回 {prompt_id: {key: value}}。
    解析不到 prompts 列表或键值形态异常 → 带行号抛 PromptError。
    """
    if not os.path.exists(registry_path):
        raise PromptError("registry 不存在：%s（D1 交付件未就位？）" % registry_path)
    entries = {}
    current = None
    in_prompts = False
    with open(registry_path, "r", encoding="utf-8") as fh:
        for ln, raw in enumerate(fh, 1):
            line = raw.rstrip("\n")
            if not line.strip() or line.lstrip().startswith("#"):
                continue
            if re.match(r"^prompts\s*:\s*(#.*)?$", line):
                in_prompts = True
                continue
            if not in_prompts:
                continue
            # 新列表项：`  - id: xxx`
            m = re.match(r"^\s*-\s+id\s*:\s*(.+)$", line)
            if m:
                pid = _strip_inline_comment(m.group(1))
                if not pid:
                    raise PromptError("registry 第 %d 行：列表项 id 为空" % ln)
                current = {}
                entries[pid] = current
                current["id"] = pid
                continue
            # 列表项内键值：`    key: value`
            m = re.match(r"^\s+([A-Za-z_][A-Za-z0-9_]*)\s*:\s*(.*)$", line)
            if m and current is not None:
                current[m.group(1)] = _strip_inline_comment(m.group(2))
                continue
            # prompts 段内出现无法识别的行（顶层新键则视为 prompts 段结束）
            if re.match(r"^[A-Za-z_]", line):
                in_prompts = False
                current = None
                continue
            raise PromptError("registry 第 %d 行无法解析：%r" % (ln, line))
    return entries


def _split_frontmatter(text, path):
    """切 frontmatter（首行 `---` 到下一个 `---`）；返回 (meta_lines, body)。无 frontmatter → 报错（§5.2 必填）。"""
    lines = text.split("\n")
    if not lines or lines[0].strip() != "---":
        raise PromptError("prompt 缺 frontmatter（首行须为 ---）：%s" % path)
    for idx in range(1, len(lines)):
        if lines[idx].strip() == "---":
            return lines[1:idx], "\n".join(lines[idx + 1:]).strip()
    raise PromptError("prompt frontmatter 未闭合（缺第二个 ---）：%s" % path)


def _meta_value(meta_lines, key):
    """从 frontmatter 行中取顶层标量键（id/version 等）；不存在返回 None。"""
    for line in meta_lines:
        m = re.match(r"^%s\s*:\s*(.+)$" % re.escape(key), line)
        if m:
            return _strip_inline_comment(m.group(1))
    return None


class PromptStore(object):
    """Registry 驱动的 prompt 仓库：load（含版本一致性核验）+ render（含残留占位符防御）。"""

    def __init__(self, prompts_dir):
        """:param prompts_dir: contracts/prompts/ 绝对路径（registry.yaml 所在目录）"""
        self.prompts_dir = prompts_dir
        self.registry = load_registry(os.path.join(prompts_dir, "registry.yaml"))
        self._cache = {}  # prompt_id → (version, body)

    def require(self, prompt_ids):
        """批跑前置核验：所有需要的 prompt id 已注册且文件存在（fail-fast，D1 未就位即停）。"""
        missing = []
        for pid in prompt_ids:
            entry = self.registry.get(pid)
            if entry is None or not os.path.exists(os.path.join(self.prompts_dir, entry.get("file", ""))):
                missing.append(pid)
        if missing:
            raise PromptError("Registry 缺以下 prompt（D1 交付件未就位或 file 路径不实）：%s" % ", ".join(missing))

    def load(self, prompt_id):
        """
        加载 prompt 正文与版本：
        - registry 必须有该 id 且 file 存在；
        - frontmatter.version 必须与 registry.version 一致（Prompt 即契约，不一致即停）。
        返回 (version: str, body: str)。
        """
        if prompt_id in self._cache:
            return self._cache[prompt_id]
        entry = self.registry.get(prompt_id)
        if entry is None:
            raise PromptError("prompt id 未注册：%s（registry.yaml）" % prompt_id)
        path = os.path.join(self.prompts_dir, entry.get("file", ""))
        if not os.path.exists(path):
            raise PromptError("prompt 文件不存在：%s（registry file 字段不实）" % path)
        with open(path, "r", encoding="utf-8") as fh:
            text = fh.read()
        meta_lines, body = _split_frontmatter(text, path)
        fm_version = _meta_value(meta_lines, "version")
        reg_version = entry.get("version")
        if fm_version and reg_version and fm_version != reg_version:
            raise PromptError(
                "prompt %s 版本不一致：frontmatter=%s registry=%s（改 prompt 必升 version 且两处同步）"
                % (prompt_id, fm_version, reg_version))
        version = reg_version or fm_version or "0.0.0"
        self._cache[prompt_id] = (version, body)
        return version, body

    def render(self, prompt_id, variables):
        """
        渲染 prompt：替换 {{input.xxx}} 占位符。
        - variables 键不带 input. 前缀；非字符串值 JSON 序列化（ensure_ascii=False，中文原样）；
        - 渲染后残留任何 {{input.*}} → 抛 PromptError 并列出缺失变量（禁止把占位符发给 LLM）。
        返回 (text: str, version: str)。
        """
        version, body = self.load(prompt_id)
        text = body
        for key, value in (variables or {}).items():
            if not isinstance(value, str):
                value = json.dumps(value, ensure_ascii=False, indent=2)
            text = text.replace("{{input.%s}}" % key, value)
            # 兼容占位符内有空白的写法 {{ input.key }}
            text = re.sub(r"\{\{\s*input\.%s\s*\}\}" % re.escape(key), lambda _m, v=value: v, text)
        leftovers = sorted(set(_PLACEHOLDER_RE.findall(text)))
        if leftovers:
            raise PromptError(
                "prompt %s 渲染后仍残留未供给变量：%s（D1↔D3 变量约定见本模块头注释）"
                % (prompt_id, ", ".join(leftovers)))
        return text, version