sanguo_moziplus_v2/src/daemon/prompt_composer.py

"""
prompt_composer.py — PromptSection Protocol + PromptContext + PromptComposer

拼装器：有序管理 prompt 段落，按优先级排序后合并为最终 prompt。
"""

import logging
from dataclasses import dataclass, field
from typing import Dict, List, Optional, Protocol, runtime_checkable

logger = logging.getLogger("moziplus-v2.prompt_composer")

# ---------------------------------------------------------------------------
# Section 优先级范围约定
# ---------------------------------------------------------------------------
PRIORITY_CONTEXT = 10        # 任务上下文
PRIORITY_PRIOR = 20          # 前序信息
PRIORITY_ROLE = 30           # 角色规范
PRIORITY_API = 40            # API 操作指令
PRIORITY_CONSTRAINTS = 50    # 硬约束
PRIORITY_EXTENSION = 60      # 扩展段


# ---------------------------------------------------------------------------
# PromptSection Protocol
# ---------------------------------------------------------------------------
@runtime_checkable
class PromptSection(Protocol):
    """一个 prompt 段"""

    name: str                     # 段名（去重用，同名覆盖）
    priority: int                 # 排序优先级（小数字=靠前）

    def render(self, context: "PromptContext") -> str:
        """渲染此段的文本内容。返回空字符串表示不注入。"""
        ...

    def should_include(self, context: "PromptContext") -> bool:
        """是否注入此段（默认 True，条件段可覆盖）。"""
        ...


# ---------------------------------------------------------------------------
# PromptContext 数据对象
# ---------------------------------------------------------------------------
@dataclass
class PromptContext:
    """Prompt 渲染的统一上下文"""

    task_id: str
    title: str
    description: str
    must_haves: str
    project_id: str
    agent_id: str

    task: Optional[Dict] = None
    role: str = "executor"
    spawn_type: str = "executor"

    # mail 专用
    from_agent: str = ""
    mail_type: str = ""          # inform / request

    # toolchain 专用
    event_type: str = ""         # ci_failure / review_request / ...
    event_data: Dict = field(default_factory=dict)
    action_type: str = ""        # 动作分类（review_result / ci_failure / ...）
    action_steps: list = field(default_factory=list)  # 结构化编号步骤列表

    # 前序产出
    depends_on_outputs: Optional[List] = None


# ---------------------------------------------------------------------------
# PromptComposer 拼装器
# ---------------------------------------------------------------------------
class PromptComposer:
    """有序拼装 prompt sections"""

    SEPARATOR = "\n\n---\n\n"
    TOKEN_BUDGET_WARN = 800      # token 预算警告阈值
    CHARS_PER_TOKEN = 3.5        # 估算比率

    def __init__(self) -> None:
        self._sections: List[PromptSection] = []

    def add(self, section: PromptSection) -> None:
        """添加一个 section（同名覆盖）"""
        self._sections = [s for s in self._sections if s.name != section.name]
        self._sections.append(section)

    def add_many(self, sections: List[PromptSection]) -> None:
        """批量添加"""
        for s in sections:
            self.add(s)

    def compose(self, context: PromptContext) -> str:
        """拼装最终 prompt

        1. 过滤 should_include=False 的段
        2. 按 priority 排序
        3. 逐段 render
        4. 过滤空段
        5. 用分隔符连接
        6. Token 预算警告（不截断）
        """
        active = [s for s in self._sections if s.should_include(context)]
        active.sort(key=lambda s: s.priority)

        parts = [s.render(context) for s in active]
        parts = [p for p in parts if p.strip()]

        result = self.SEPARATOR.join(parts)

        # Token 估算
        tokens = max(1, int(len(result) / self.CHARS_PER_TOKEN))
        logger.debug(
            "Composed prompt from %d sections, %d tokens",
            len(parts), tokens,
        )

        if tokens > self.TOKEN_BUDGET_WARN:
            logger.warning(
                "Prompt exceeds %d token budget: %d tokens (task_id=%s)",
                self.TOKEN_BUDGET_WARN, tokens, context.task_id,
            )

        return result


# ---------------------------------------------------------------------------
class GiteaConventionSection:
    """Gitea 标题规范引导段 — 提醒 Agent 创建 Issue/PR 时遵循标题格式。"""

    name: str = "gitea_convention"
    priority: int = 55  # CONSTRAINTS(50) 和 EXTENSION(60) 之间

    CONVENTION_TEXT = (
        "## Gitea 标题规范\n"
        "创建 Issue/PR 时，标题**必须**包含项目代号前缀：\n"
        "- Issue: `[代号] type: 简述`，如 `[moz] bug: Mail API 500`\n"
        "- PR: `[代号] type(scope): 简述`，如 `[moz] impl(daemon): WikiGuideSection 注入`\n"
        "代号：moz=moziplus_v2, quant=quant_live, vnpy=vnpy\n"
        "type: bug/feat/impl/fix/docs/test/ci/refactor/chore"
    )

    def render(self, context: "PromptContext") -> str:
        return self.CONVENTION_TEXT

    def should_include(self, context: "PromptContext") -> bool:
        return True


# ---------------------------------------------------------------------------
# WikiGuideSection — 知识查询引导段
# ---------------------------------------------------------------------------
class WikiGuideSection:
    """知识查询引导段 — 引导 Agent 在关键决策点查 wiki-vault。"""

    name: str = "wiki_guide"
    priority: int = 60  # PRIORITY_EXTENSION

    WIKI_GUIDE = (
        "## 知识查询引导\n"
        "涉及方案设计、编码实现、故障排查时，先查 wiki-vault 相关实践：\n"
        "- 路径：/Volumes/KnowledgeBase/wiki-vault/\n"
        "- 速查：index.md → grep 关键词 → summary 字段 → 按需读全文\n"
        "- 查不到：在 _meta/knowledge-gaps.md 记录"
    )

    def render(self, context: "PromptContext") -> str:
        return self.WIKI_GUIDE

    def should_include(self, context: "PromptContext") -> bool:
        return True


# ---------------------------------------------------------------------------
# DeliveryChecklistSection — 交付检查清单
# ---------------------------------------------------------------------------
class DeliveryChecklistSection:
    """交付检查清单 — 提醒 Agent 完成前同步关联成果物。"""

    name: str = "delivery_checklist"
    priority: int = 55  # CONSTRAINTS(50) 和 EXTENSION(60) 之间

    CHECKLIST_TEXT = (
        "## 交付检查\n"
        "完成代码改动前确认：\n"
        "- 改了实现 → docs/design/ 对应设计文档是否需要更新\n"
        "- 改了实现 → tests/ 是否有对应测试脚本需要更新\n"
        "- 所有成果物变更通过 PR 流程：PR review 把关设计合理性，CI 把关代码质量，CD 把关部署正确性\n"
    )

    def render(self, context: "PromptContext") -> str:
        return self.CHECKLIST_TEXT

    def should_include(self, context: "PromptContext") -> bool:
        return True