auto-sync: 2026-06-07 01:29:27

src/daemon/bootstrap.py

@@ -1,25 +1,21 @@
-"""Bootstrap 拼装 — L0-L3 四层上下文构建
+"""Bootstrap 拼装 — L2 引擎注入层
 
-L0 铁律层（~500 tokens） → Hook 注入，不占 bootstrap
-L1 角色层（~2000 tokens） → SOUL.md / IDENTITY.md（Agent 自带）
-L2 引擎注入层（~1500 tokens） → prompt_templates 按 role 拼装
-L3 被动参考层（按需加载） → Skill description 四要素
+4 段结构（~800 tokens）：
+  1. 任务上下文：title / description / must_haves / status
+  2. 前序产出：depends_on 产出摘要 + handoff comment
+  3. 角色操作规范全文：通过 ROLE_SKILL_MAP 从 Skill 文件读取
+  4. 硬约束：状态流转约束
 
-Bootstrap 按 role 精确注入：执行者注入 Guardrail + 审查协议，
-审查者注入审查协议，庞统注入审查协议。
+A 类 Skill 由引擎确定性注入全文，不靠 Description 触发。
 """
 
-from __future__ import annotations
-
-import json
 import logging
-import re
+import os
 from pathlib import Path
 from typing import Any, Dict, List, Optional
 
 logger = logging.getLogger("moziplus-v2.bootstrap")
 
-# 粗略 token 估算：英文 ~4 chars/token，中文 ~2 chars/token
 CHARS_PER_TOKEN = 3.0
 
 
@@ -29,188 +25,135 @@ def estimate_tokens(text: str) -> int:
 
 
 class BootstrapBuilder:
-    """Bootstrap 四层上下文构建器"""
+    """L2 引擎注入层构建器（v2.1 四段式）"""
 
-    def __init__(
-        self,
-        template_dir: Optional[Path] = None,
-        max_tokens: int = 4096,
-    ):
-        """
-        Args:
-            template_dir: prompt_templates 目录路径
-            max_tokens: bootstrap 最大 token 数
-        """
-        self.template_dir = template_dir
+    ROLE_SKILL_MAP = {
+        "executor":          "blackboard-executor",
+        "reviewer":          "blackboard-reviewer",
+        "reviewer-simayi":   "blackboard-reviewer-simayi",
+        "reviewer-pangtong": "blackboard-reviewer-pangtong",
+        "planner":           "blackboard-planner",
+        "claim":             "blackboard-claim",
+    }
+
+    # 默认从环境变量或配置读取，fallback 到默认路径
+    SKILL_BASE_PATH = os.environ.get(
+        "MOZI_SKILL_PATH",
+        os.path.expanduser("~/.sanguo_projects/sanguo_mozi/skills")
+    )
+
+    def __init__(self, max_tokens: int = 4096):
         self.max_tokens = max_tokens
 
-    def build(
-        self,
-        role: str,
-        task_context: Optional[Dict[str, Any]] = None,
-        project_context: Optional[Dict[str, Any]] = None,
-        guardrail_rules: Optional[str] = None,
-        review_protocols: Optional[str] = None,
-        experiences: Optional[List[Dict[str, Any]]] = None,
-        skill_descriptions: Optional[List[Dict[str, Any]]] = None,
-    ) -> str:
-        """构建 bootstrap 文本
+    def build(self, task: dict, role: str) -> str:
+        """构建 4 段 bootstrap 文本
 
         Args:
-            role: 角色（executor / reviewer / planner / pangtong）
-            task_context: 任务上下文（黑板数据）
-            project_context: 项目背景
-            guardrail_rules: Guardrail 规则（仅执行者）
-            review_protocols: 审查协议（执行者+审查者+庞统）
-            experiences: 相关经验列表
-            skill_descriptions: Skill 描述列表
+            task: 任务数据字典，需包含 title/description/must_haves/status 等字段
+            role: 角色，对应 ROLE_SKILL_MAP 的 key
 
         Returns:
             拼装好的 bootstrap 文本
         """
-        layers: List[str] = []
+        sections: List[str] = []
 
-        # L2a: 操作规范
-        role_template = self._load_template(f"{role}.md")
-        if role_template:
-            layers.append(role_template)
+        # 段 1: 任务上下文
+        sections.append(self._format_task_context(task))
 
-        # L2b: 项目背景
-        if project_context:
-            layers.append(self._format_project_context(project_context))
+        # 段 2: 前序产出（有依赖时注入）
+        if task.get("depends_on_outputs"):
+            sections.append(self._format_prior_outputs(task["depends_on_outputs"]))
 
-        # L2c: 任务上下文
-        if task_context:
-            layers.append(self._format_task_context(task_context))
+        # 段 3: 角色操作规范全文（通过 ROLE_SKILL_MAP 从 Skill 文件读取）
+        skill_name = self.ROLE_SKILL_MAP.get(role)
+        if skill_name:
+            skill_content = self._read_skill(skill_name)
+            if skill_content:
+                sections.append(skill_content)
+        elif role not in ("discussion",):
+            logger.warning("No skill mapping for role: %s", role)
 
-        # L2d: 前序信息（depends_on 产出摘要）
-        if task_context and task_context.get("depends_on_outputs"):
-            layers.append(self._format_depends_on(task_context["depends_on_outputs"]))
+        # 段 4: 硬约束
+        sections.append(self._format_constraints(role))
 
-        # L2e: Guardrail 规则（仅执行者）
-        if role == "executor" and guardrail_rules:
-            layers.append(guardrail_rules)
+        bootstrap = "\n\n---\n\n".join(sections)
 
-        # L2f: 审查协议（执行者+审查者+庞统）
-        if role in ("executor", "reviewer", "pangtong") and review_protocols:
-            layers.append(review_protocols)
-
-        # L2g: 经验注入
-        if experiences:
-            layers.append(self._format_experiences(experiences))
-
-        # L3: Skill descriptions
-        if skill_descriptions:
-            layers.append(self._format_skills(skill_descriptions))
-
-        if not layers:
-            return f"# Role: {role}"
-        bootstrap = "\n\n---\n\n".join(layers)
-
-        # 检查 token 限制
-        tokens = estimate_tokens(bootstrap)
-        if tokens > self.max_tokens:
-            logger.warning("Bootstrap exceeds token limit: %d > %d, truncating",
-                           tokens, self.max_tokens)
-            bootstrap = self._truncate_to_tokens(bootstrap, self.max_tokens)
+        # Token 预算检查（warn 但不截断）
+        tokens = max(1, int(len(bootstrap) / CHARS_PER_TOKEN))
+        if tokens > 800:
+            logger.warning("Bootstrap exceeds 800 token budget: %d tokens (role=%s)",
+                           tokens, role)
 
         return bootstrap
 
-    def _load_template(self, filename: str) -> Optional[str]:
-        """加载 prompt template"""
-        if not self.template_dir:
-            return None
-        path = self.template_dir / filename
-        if path.exists():
-            return path.read_text(encoding="utf-8")
-        return None
+    def build_for_task(self, task: Any, role: str, **kwargs) -> str:
+        """从 Task 对象构建 bootstrap（便捷方法，兼容旧调用）
 
-    def _format_project_context(self, ctx: Dict[str, Any]) -> str:
-        """格式化项目背景"""
-        parts = ["## 项目背景"]
-        if ctx.get("name"):
-            parts.append(f"项目: {ctx['name']}")
-        if ctx.get("description"):
-            parts.append(f"描述: {ctx['description']}")
-        if ctx.get("agents"):
-            parts.append(f"团队: {', '.join(ctx['agents'])}")
-        return "\n".join(parts)
+        忽略 kwargs 中的 project_config/experiences 等旧参数。
+        must_haves 包含验收标准（不需要 project_context）。
+        """
+        task_dict = {
+            "task_id": task.id,
+            "title": task.title,
+            "description": task.description,
+            "must_haves": task.must_haves,
+            "status": getattr(task, 'status', ''),
+        }
+        # depends_on_outputs 如果有的话
+        if hasattr(task, 'depends_on_outputs'):
+            task_dict["depends_on_outputs"] = task.depends_on_outputs
 
-    def _format_task_context(self, ctx: Dict[str, Any]) -> str:
-        """格式化任务上下文"""
+        return self.build(task=task_dict, role=role)
+
+    def _read_skill(self, skill_name: str) -> str:
+        """从 Skill 文件读取全文（带 fallback）"""
+        path = os.path.join(self.SKILL_BASE_PATH, skill_name, "SKILL.md")
+        try:
+            with open(path, encoding="utf-8") as f:
+                return f.read()
+        except FileNotFoundError:
+            logger.error("Skill file not found: %s", path)
+            return ""
+
+    def _format_task_context(self, task: dict) -> str:
+        """格式化任务上下文（段 1）"""
         parts = ["## 任务上下文"]
-        if ctx.get("task_id"):
-            parts.append(f"任务ID: {ctx['task_id']}")
-        if ctx.get("title"):
-            parts.append(f"标题: {ctx['title']}")
-        if ctx.get("description"):
-            parts.append(f"描述: {ctx['description']}")
-        if ctx.get("task_type"):
-            parts.append(f"类型: {ctx['task_type']}")
-        if ctx.get("must_haves"):
-            parts.append(f"必须完成: {ctx['must_haves']}")
-        if ctx.get("risk_level"):
-            parts.append(f"风险级别: {ctx['risk_level']}")
+        if task.get("task_id"):
+            parts.append(f"任务ID: {task['task_id']}")
+        if task.get("title"):
+            parts.append(f"标题: {task['title']}")
+        if task.get("description"):
+            parts.append(f"描述: {task['description']}")
+        if task.get("must_haves"):
+            parts.append(f"必须完成: {task['must_haves']}")
+        if task.get("status"):
+            parts.append(f"当前状态: {task['status']}")
         return "\n".join(parts)
 
-    def _format_depends_on(self, outputs: List[Dict[str, Any]]) -> str:
-        """格式化前序产出摘要"""
+    def _format_prior_outputs(self, outputs: list) -> str:
+        """格式化前序产出摘要（段 2）"""
         parts = ["## 前序产出"]
         for out in outputs:
             parts.append(f"- [{out.get('task_id', '?')}] {out.get('summary', '无摘要')}")
         return "\n".join(parts)
 
-    def _format_experiences(self, experiences: List[Dict[str, Any]]) -> str:
-        """格式化经验注入"""
-        parts = ["## 相关经验"]
-        for exp in experiences[:5]:  # 最多 5 条
-            parts.append(f"- [{exp.get('category', '?')}] {exp.get('summary', '')}")
-        return "\n".join(parts)
-
-    def _format_skills(self, skills: List[Dict[str, Any]]) -> str:
-        """格式化 Skill 描述"""
-        parts = ["## 可用技能"]
-        for skill in skills[:10]:
-            parts.append(f"- **{skill.get('name', '?')}**: {skill.get('description', '')}")
-        return "\n".join(parts)
-
-    def _truncate_to_tokens(self, text: str, max_tokens: int) -> str:
-        """截断到指定 token 数"""
-        max_chars = int(max_tokens * CHARS_PER_TOKEN)
-        if len(text) <= max_chars:
-            return text
-        trunc_marker = "\n\n[... bootstrap truncated]"
-        return text[:max_chars - len(trunc_marker)] + trunc_marker
-
-    def build_for_task(
-        self,
-        task: Any,
-        role: str,
-        project_config: Optional[Dict] = None,
-        experiences: Optional[List[Dict]] = None,
-    ) -> str:
-        """从 Task 对象构建 bootstrap（便捷方法）"""
-        task_context = {
-            "task_id": task.id,
-            "title": task.title,
-            "description": task.description,
-            "task_type": task.task_type,
-            "must_haves": task.must_haves,
-            "risk_level": task.risk_level,
-        }
-
-        project_context = None
-        if project_config:
-            project_context = {
-                "name": project_config.get("name"),
-                "description": project_config.get("description"),
-                "agents": project_config.get("agents", []),
-            }
-
-        return self.build(
-            role=role,
-            task_context=task_context,
-            project_context=project_context,
-            experiences=experiences,
-        )
-
+    def _format_constraints(self, role: str) -> str:
+        """格式化硬约束（段 4）"""
+        constraints = ["## 硬约束"]
+        if role == "executor":
+            constraints.extend([
+                "- 完成后必须标 review",
+                "- 产出物不能为空（系统会验证）",
+                "- handoff comment ≥ 50 字符",
+            ])
+        elif role.startswith("reviewer"):
+            constraints.extend([
+                "- 审查结果必须明确 pass/fail",
+                "- 评审意见须附证据（文件:行号）",
+            ])
+        elif role == "planner":
+            constraints.extend([
+                "- 需求不清时提问，不要猜",
+                "- 子任务必须有明确的终态定义",
+            ])
+        return "\n".join(constraints)