第三章：Prompt Engineering 的工程化

## YOUR ROLE - SPEC WRITER AGENT

You are the **Spec Writer Agent** in the Auto-Build spec creation pipeline.
Your ONLY job is to read the gathered context and write a complete, valid `spec.md` document.

**Key Principle**: Synthesize context into actionable spec. No user interaction needed.

---

## YOUR CONTRACT

**Inputs** (read these files):
- `project_index.json` - Project structure
- `requirements.json` - User requirements
- `context.json` - Relevant files discovered

**Output**: `spec.md` - Complete specification document

You MUST create `spec.md` with ALL required sections (see template below).

**DO NOT** interact with the user. You have all the context you need.

# 函数签名（代码世界）
def write_spec(
    project_index: dict,
    requirements: dict,
    context: dict
) -> SpecDocument:
    ...

# Prompt 中的契约（AI 世界）
"""
Inputs: project_index.json, requirements.json, context.json
Output: spec.md
"""

## PHASE 3: CREATE implementation_plan.json

**🚨 CRITICAL: YOU MUST USE THE WRITE TOOL TO CREATE THIS FILE 🚨**

You MUST use the Write tool to save the implementation plan to `implementation_plan.json`.
Do NOT just describe what the file should contain - you must actually call the Write tool
with the complete JSON content.

**Required action:** Call the Write tool with:
- file_path: `implementation_plan.json` (in the spec directory)
- content: The complete JSON plan structure shown below

**🚨 END OF PHASE 4 CHECKPOINT 🚨**

Before proceeding to PHASE 5, verify you have:
1. ✅ Created the complete implementation_plan.json structure
2. ✅ Used the Write tool to save it (not just described it)
3. ✅ Added the summary section with parallelism analysis

If you have NOT used the Write tool yet, STOP and do it now!

def generate_subtask_prompt(
    spec_dir: Path,
    project_dir: Path,
    subtask: dict,
    phase: dict,
    attempt_count: int = 0,
    recovery_hints: list[str] | None = None,
) -> str:
    """
    Generate a minimal, focused prompt for implementing a single subtask.
    Returns: A focused prompt string (~100 lines instead of 900)
    """
    sections = []

    # 1. 环境上下文（工作目录、规格文件位置）
    sections.append(generate_environment_context(project_dir, spec_dir))

    # 2. 只有这个子任务相关的信息
    sections.append(f"## 你的任务\n{subtask['description']}")

    # 3. 只需要修改的那几个文件（不是整个代码库）
    if files_to_modify:
        sections.append(f"## 要修改的文件\n{files_to_modify}")

    # 4. 参考哪些文件的模式（告诉 Agent 去哪里学）
    if patterns_from:
        sections.append(f"## 参考以下文件的代码模式\n{patterns_from}")

    # 5. 如果是重试，注入上次失败的提示
    if recovery_hints:
        hints = "\n".join(f"- {h}" for h in recovery_hints)
        sections.append(f"## 上次尝试的问题\n{hints}")

    return "\n\n".join(sections)

def generate_environment_context(project_dir: Path, spec_dir: Path) -> str:
    # 检测是否在隔离的 Worktree 中运行
    is_worktree, parent_project_path = detect_worktree_isolation(project_dir)

    sections = []

    # 只有在 Worktree 模式下，才注入这段 500 字的安全警告
    if is_worktree and parent_project_path:
        sections.append(generate_worktree_isolation_warning(
            project_dir, parent_project_path
        ))

    # 基础的环境信息（总是注入）
    sections.append(f"""## YOUR ENVIRONMENT

**Working Directory:** `{project_dir}`
**Spec Location:** `{relative_spec}/`
{"**Isolation Mode:** WORKTREE" if is_worktree else ""}
...
""")
    return "".join(sections)

You MUST create `requirements.json` with this EXACT structure:

```json
{
  "task_description": "Clear description of what to build",
  "workflow_type": "feature|refactor|investigation|migration|simple",
  "services_involved": ["service1", "service2"],
  "user_requirements": ["Requirement 1", "Requirement 2"],
  "acceptance_criteria": ["Criterion 1", "Criterion 2"],
  "constraints": ["Any constraints or limitations"],
  "created_at": "ISO timestamp"
}

cat > critique_report.json << 'EOF'
{
  "critique_completed": true,
  "issues_found": [...]
}
EOF

# core/client.py
def create_client(
    ...
    output_format: dict | None = None,  # 结构化输出格式
):
    ...
    options_kwargs = {
        ...
    }
    if output_format:
        options_kwargs["output_format"] = output_format

from pydantic import BaseModel

class QAResult(BaseModel):
    status: Literal["approved", "rejected"]
    issues: list[dict]
    timestamp: str

# 创建强制输出 Pydantic 模型的 Client
client = create_client(
    ...,
    output_format={
        "type": "json_schema",
        "schema": QAResult.model_json_schema()
    }
)

## YOUR ROLE - SPEC CRITIC AGENT

You are the **Spec Critic Agent** in the Auto-Build spec creation pipeline.
Your ONLY job is to critically review the spec.md document, find issues, and fix them.

**Key Principle**: Use extended thinking (ultrathink). Find problems BEFORE implementation.

<!-- PROJECT-SPECIFIC VALIDATION TOOLS WILL BE INJECTED HERE -->
<!-- The following sections are dynamically added based on project type: -->
<!-- - Electron validation (for Electron apps) -->
<!-- - Puppeteer browser automation (for web frontends) -->
<!-- - Database validation (for projects with databases) -->
<!-- - API validation (for projects with API endpoints) -->

# 注入逻辑（简化）
def build_qa_prompt(spec_dir, project_dir):
    base_prompt = load_file("prompts/qa_reviewer.md")
    capabilities = detect_project_capabilities(project_dir)

    injections = []
    if capabilities["has_electron"]:
        injections.append(load_file("prompts/mcp_tools/electron_validation.md"))
    if capabilities["has_database"]:
        injections.append(load_file("prompts/mcp_tools/database_validation.md"))
    if capabilities["has_api"]:
        injections.append(load_file("prompts/mcp_tools/api_validation.md"))

    # 把注入内容替换到模板中的注释位置
    injection_text = "\n\n".join(injections)
    return base_prompt.replace(
        "<!-- PROJECT-SPECIFIC VALIDATION TOOLS WILL BE INJECTED HERE -->",
        injection_text
    )

## PHASE 1: VERIFY ALL SUBTASKS COMPLETED

```bash
echo "Completed: $(grep -c '"status": "completed"' implementation_plan.json)"
echo "Pending: $(grep -c '"status": "pending"' implementation_plan.json)"

## STEP 3: FIND YOUR NEXT SUBTASK

**CRITICAL**: Never work on a subtask if its phase's dependencies aren't complete!

Phase 1: Backend     [depends_on: []]           → Can start immediately
Phase 2: Worker      [depends_on: ["phase-1"]]  → Blocked until Phase 1 done
Phase 3: Frontend    [depends_on: ["phase-1"]]  → Blocked until Phase 1 done

## STEP 5.5: GENERATE & REVIEW PRE-IMPLEMENTATION CHECKLIST

**CRITICAL**: Before writing any code, generate a predictive bug prevention checklist.
This step uses historical data and pattern analysis to predict likely issues BEFORE
they happen.

{
  "subtask": "实现用户登录 API",
  "risk_level": "high",
  "predicted_issues": [
    {
      "category": "security",
      "issue": "密码明文存储",
      "prevention": "使用 bcrypt 哈希，不存储原始密码"
    },
    {
      "category": "validation",
      "issue": "没有限制登录尝试次数",
      "prevention": "添加速率限制，防止暴力破解"
    }
  ],
  "must_verify_before_submit": [
    "密码使用哈希存储",
    "JWT token 有过期时间",
    "SQL 查询使用参数化，无注入风险"
  ]
}

"""
Lab 3.1 - 预实现检查清单生成器

工程目标：
1. 练习契约模式（明确 Inputs/Outputs）
2. 练习强制执行语言（LLM 必须写文件）
3. 练习结构化输出（JSON Schema 驱动）
"""
import asyncio
import json
from pathlib import Path
from claude_agent_sdk import ClaudeSDKClient, ClaudeAgentOptions

WORK_DIR = Path("./lab3_workspace")
WORK_DIR.mkdir(exist_ok=True)
CHECKLIST_FILE = WORK_DIR / "checklist.json"


# ============================================================
# TODO 1: 设计系统提示词（System Prompt）
# 要求：体现"契约模式"和"强制执行"原则
# ============================================================

CHECKLIST_SYSTEM_PROMPT = """
你是一名代码风险分析专家。

## YOUR CONTRACT

**Input**: 一段子任务描述（通过 user message 提供）
**Output**: checklist.json 文件（必须使用 Write 工具创建）

## 关键规则

TODO: 写出体现以下原则的规则：
1. 强制执行：LLM 必须调用 Write 工具，不能只描述内容
2. 精确的输出格式：提供精确的 JSON Schema
3. 阶段门控：分析完成后必须写文件，然后停止

## 输出格式

TODO: 在这里放入精确的 JSON Schema，包含：
- subtask (string): 子任务描述
- risk_level (string): "low" | "medium" | "high" | "critical"
- predicted_issues (array): 预测的问题列表
  - category (string): "security" | "validation" | "error_handling" | "performance"
  - issue (string): 问题描述
  - prevention (string): 预防方法
- must_verify_before_submit (array of strings): 提交前必须验证的事项

## 强制执行

TODO: 写出强制 LLM 调用 Write 工具的指令
记住：不能只展示 JSON 内容，必须实际写入文件
"""


# ============================================================
# TODO 2: 实现 generate_checklist() 函数
# ============================================================

async def generate_checklist(subtask_description: str) -> dict:
    """
    为给定的子任务生成预实现风险清单。

    工作流：
    1. 启动 Agent，让它分析 subtask_description
    2. Agent 必须将分析结果写入 checklist.json
    3. 读取 checklist.json 并返回解析后的字典

    Args:
        subtask_description: 子任务的文字描述

    Returns:
        解析后的清单字典

    Raises:
        FileNotFoundError: 如果 Agent 没有写入文件（表明强制执行失败）
        json.JSONDecodeError: 如果 Agent 写入的不是有效 JSON
    """

    # 在开始之前，清理上次运行的文件
    if CHECKLIST_FILE.exists():
        CHECKLIST_FILE.unlink()

    # TODO: 创建 ClaudeAgentOptions
    # 提示：
    # - model: "claude-haiku-4-5-20251001"（节约成本）
    # - system_prompt: 使用上面定义的 CHECKLIST_SYSTEM_PROMPT
    # - allowed_tools: 至少需要 ["Write"]（思考：还需要别的吗？）
    # - cwd: str(WORK_DIR)
    # - max_turns: 10（这个任务不需要很多轮）
    options = ClaudeAgentOptions(
        model="claude-haiku-4-5-20251001",
        system_prompt=CHECKLIST_SYSTEM_PROMPT,
        allowed_tools=["Write"],  # TODO: 这就够了吗？
        cwd=str(WORK_DIR),
        max_turns=10,
    )

    client = ClaudeSDKClient(options=options)

    # TODO: 构建 user prompt
    # 提示：要包含子任务描述，以及输出文件路径
    user_prompt = f"""
    TODO: 告诉 Agent 要分析什么，以及把结果写到哪个文件
    子任务描述：{subtask_description}
    """

    # 运行 Agent
    print(f"分析子任务：{subtask_description[:50]}...")
    async for message in client.process_query(user_prompt):
        msg_type = type(message).__name__
        # TODO: 打印有意义的进度信息
        # 提示：可以只打印 ToolUseBlock 类型的消息，查看 Agent 调用了什么工具
        pass

    # TODO: 验证 Agent 是否实际写入了文件
    # 如果文件不存在，这是强制执行失败的信号
    if not CHECKLIST_FILE.exists():
        raise FileNotFoundError(
            f"Agent 没有创建 {CHECKLIST_FILE}。\n"
            f"这说明 System Prompt 的强制执行语言不够强硬。\n"
            f"请修改 CHECKLIST_SYSTEM_PROMPT，让 LLM 明确知道必须调用 Write 工具。"
        )

    # TODO: 读取并解析文件
    content = CHECKLIST_FILE.read_text(encoding="utf-8")
    return json.loads(content)


# ============================================================
# TODO 3: 实现 validate_checklist() 验证函数
# ============================================================

def validate_checklist(checklist: dict) -> tuple[bool, list[str]]:
    """
    验证清单格式是否符合预期 Schema。

    Returns:
        (is_valid, error_messages)
    """
    errors = []

    # TODO: 验证以下字段存在且类型正确：
    # - subtask (str)
    # - risk_level (str, 必须是 "low"/"medium"/"high"/"critical" 之一)
    # - predicted_issues (list, 至少包含 1 个元素)
    #   - 每个元素有 category, issue, prevention 字段
    # - must_verify_before_submit (list of str)

    return len(errors) == 0, errors


# ============================================================
# 测试代码（补全上面的 TODO 后运行）
# ============================================================

async def main():
    test_subtasks = [
        "实现用户登录 API：接受 username/password，返回 JWT token",
        "添加文件上传功能：支持上传图片，最大 10MB",
        "实现支付接口：调用 Stripe API 处理信用卡支付",
    ]

    for subtask in test_subtasks:
        print(f"\n{'='*60}")
        print(f"子任务：{subtask}")
        print('='*60)

        try:
            checklist = await generate_checklist(subtask)

            # 验证输出格式
            is_valid, errors = validate_checklist(checklist)

            if is_valid:
                print(f"✅ 风险级别：{checklist['risk_level'].upper()}")
                print(f"预测问题数：{len(checklist['predicted_issues'])}")
                for issue in checklist['predicted_issues']:
                    print(f"  [{issue['category']}] {issue['issue']}")
                    print(f"    → 预防：{issue['prevention']}")
                print(f"\n提交前验证项：")
                for item in checklist['must_verify_before_submit']:
                    print(f"  ☐ {item}")
            else:
                print(f"❌ 格式验证失败：")
                for error in errors:
                    print(f"  - {error}")

        except FileNotFoundError as e:
            print(f"❌ 强制执行失败：\n{e}")
        except json.JSONDecodeError as e:
            print(f"❌ JSON 解析失败：{e}")
            print(f"Agent 写入的内容：{CHECKLIST_FILE.read_text()[:200]}")


if __name__ == "__main__":
    asyncio.run(main())