第十章：规格创建流水线

# apps/backend/spec/complexity.py（简化）

class Complexity(Enum):
    SIMPLE   = "simple"    # 1-2 个文件，单一服务，无外部集成
    STANDARD = "standard"  # 3-10 个文件，1-2 个服务，少量集成
    COMPLEX  = "complex"   # 10+ 个文件，多个服务，外部集成

def phases_to_run(self) -> list[str]:
    """根据复杂度返回需要执行的阶段列表。"""
    # 如果 AI 给出了推荐阶段，优先使用
    if self.recommended_phases:
        return self.recommended_phases

    if self.complexity == Complexity.SIMPLE:
        return ["discovery", "historical_context", "quick_spec", "validation"]

    elif self.complexity == Complexity.STANDARD:
        phases = ["discovery", "historical_context", "requirements"]
        if self.needs_research:
            phases.append("research")  # 按需启用研究阶段
        phases.extend(["context", "spec_writing", "planning", "validation"])
        return phases

    else:  # COMPLEX
        return [
            "discovery", "historical_context", "requirements",
            "research", "context", "spec_writing",
            "self_critique",   # 只有 COMPLEX 才默认运行自我批判
            "planning", "validation",
        ]

# apps/backend/spec/complexity.py（简化）

class ComplexityAnalyzer:
    # 暗示简单任务的关键词
    SIMPLE_KEYWORDS = ["fix", "typo", "update", "change", "rename", "rename",
                       "style", "color", "text", "button", "margin", "padding"]

    # 暗示复杂任务的关键词
    COMPLEX_KEYWORDS = ["integrate", "api", "sdk", "database", "migrate",
                        "docker", "authentication", "oauth", "microservice",
                        "refactor", "architecture"]

    def _calculate_complexity(self, signals, integrations,
                               infra_changes, estimated_files, estimated_services):
        """根据所有信号计算最终复杂度。"""
        # SIMPLE 的强指标
        if (estimated_files <= 2 and estimated_services == 1
                and len(integrations) == 0 and not infra_changes
                and signals["simple_keywords"] > 0
                and signals["complex_keywords"] == 0):
            return Complexity.SIMPLE, 0.9, "Single service, minimal scope"

        # COMPLEX 的强指标
        if (len(integrations) >= 2 or infra_changes
                or estimated_services >= 3 or estimated_files >= 10
                or signals["complex_keywords"] >= 3):
            return Complexity.COMPLEX, 0.85, f"{len(integrations)} integrations detected"

        # 默认 STANDARD
        return Complexity.STANDARD, 0.75, f"{estimated_files} files estimated"

# apps/backend/spec/pipeline/orchestrator.py（简化）

async def _phase_complexity_assessment_with_requirements(self):
    if self.complexity_override:
        # 手动强制指定复杂度
        self.assessment = self._create_override_assessment()
    elif self.use_ai_assessment:
        # AI 评估（默认）
        self.assessment = await self._run_ai_assessment(task_logger)
    else:
        # 启发式评估（兜底）
        self.assessment = self._heuristic_assessment()

# apps/backend/spec/phases/models.py（简化）

MAX_RETRIES = 3  # 每个阶段最多重试 3 次

@dataclass
class PhaseResult:
    phase_name: str
    success: bool
    output_files: list[str]   # 本阶段产出的文件
    errors: list[str]         # 遇到的错误
    retries: int              # 实际重试次数

# apps/backend/spec/pipeline/orchestrator.py（简化）

async def _store_phase_summary(self, phase_name: str) -> None:
    """将已完成阶段的输出压缩为摘要，供后续阶段使用。"""
    # 收集本阶段产出的所有文件内容
    phase_output = gather_phase_outputs(self.spec_dir, phase_name)

    # 用 AI 将其压缩为约 500 字的摘要
    summary = await summarize_phase_output(
        phase_name, phase_output,
        model="sonnet",
        target_words=500,
    )

    if summary:
        self._phase_summaries[phase_name] = summary

# 在构建每个 Agent 的 prompt 时注入历史阶段摘要
prior_summaries = format_phase_summaries(self._phase_summaries)
return await runner.run_agent(
    prompt_file,
    additional_context=additional_context,
    prior_phase_summaries=prior_summaries,  # 注入历史阶段信息
    ...
)

# apps/backend/spec/critique.py（简化）

def generate_critique_prompt(subtask, files_modified, patterns_from) -> str:
    """生成让 AI 自我评估的提示词。"""
    return f"""
## MANDATORY Self-Critique: {subtask_id}

### STEP 1: Code Quality Checklist
- [ ] 遵循参考文件的代码模式
- [ ] 变量命名符合代码库惯例
- [ ] 没有遗留调试语句

### STEP 2: Implementation Completeness
Expected files: {files_to_modify}
Actual modified: {files_modified}
- [ ] 所有预期文件都已修改

### STEP 3: Potential Issues Analysis
[列出问题]

### STEP 4: Improvements Made
[已修复的问题]

### STEP 5: Final Verdict
**PROCEED:** [YES/NO]
**REASON:** [...]
**CONFIDENCE:** [High/Medium/Low]
"""

# apps/backend/spec/critique.py（简化）

def parse_critique_response(response: str) -> CritiqueResult:
    # 提取 PROCEED 判决
    proceed_match = re.search(
        r"\*\*PROCEED:\*\*\s*\[?\s*(YES|NO)",
        response, re.IGNORECASE
    )
    passes = proceed_match.group(1).upper() == "YES"

    # 提取 Step 3 中列出的问题
    issues_section = re.search(
        r"### STEP 3:.*?Potential Issues.*?\n\n(.*?)(?=###|\Z)",
        response, re.DOTALL
    )
    # 解析每行问题，过滤掉占位符文本...

    return CritiqueResult(passes=passes, issues=issues, ...)

def should_proceed(result: CritiqueResult) -> bool:
    """只有 PROCEED=YES 且没有未解决问题，才允许标记完成。"""
    if not result.passes:
        return False
    if result.issues:   # 找到问题但没修复
        return False
    return True

# apps/backend/spec/phases/spec_phases.py（简化）

def _greenfield_context() -> str:
    """为空目录项目返回额外上下文。"""
    return """
**GREENFIELD PROJECT**: 这是一个空的新项目，没有现有代码。

调整你的方法：
- 不要引用现有文件、模式或代码结构
- 聚焦于需要创建什么，而非修改什么
- 定义初始项目结构、文件和目录
- 指定技术栈、框架和依赖项安装方式
- 对于"参考文件"部分，描述行业最佳实践而非现有代码
"""

async def phase_quick_spec(self) -> PhaseResult:
    is_greenfield = _is_greenfield_project(self.spec_dir)

    context_str = f"""
**Task**: {self.task_description}
**Complexity**: SIMPLE
{_greenfield_context() if is_greenfield else ""}
"""
    success, output = await self.run_agent_fn(
        "spec_quick.md",
        additional_context=context_str,
    )

# apps/backend/spec/pipeline/orchestrator.py（简化）

def _run_review_checkpoint(self, auto_approve: bool) -> bool:
    """等待用户批准规格后再开始编码。"""
    try:
        review_state = run_review_checkpoint(
            spec_dir=self.spec_dir,
            auto_approve=auto_approve,  # 可通过 CLI 参数跳过
        )

        if not review_state.is_approved():
            print_status("Build will not proceed without approval.", "warning")
            return False

    except KeyboardInterrupt:
        print_status("Review interrupted. Run again to continue.", "info")
        return False

    return True

# apps/backend/spec/pipeline/orchestrator.py（简化）

# requirements.json 生成后，用任务描述重命名目录
new_spec_dir = rename_spec_dir_from_requirements(self.spec_dir)
if new_spec_dir != self.spec_dir:
    self.spec_dir = new_spec_dir
    # 同步更新所有引用了旧路径的对象
    self.validator = SpecValidator(self.spec_dir)
    phase_executor.spec_dir = self.spec_dir
    phase_executor.spec_validator = self.validator

# 在 SpecNumberLock 作用域内创建目录，确保序号唯一
with SpecNumberLock(self.project_dir) as lock:
    self.spec_dir = create_spec_dir(self.specs_dir, lock)
    self.spec_dir.mkdir(parents=True, exist_ok=True)

# lab_10_spec_pipeline.py
from __future__ import annotations

import json
import re
from dataclasses import dataclass, field
from enum import Enum
from pathlib import Path


class Complexity(Enum):
    SIMPLE   = "simple"
    STANDARD = "standard"
    COMPLEX  = "complex"


@dataclass
class ComplexityAssessment:
    complexity: Complexity
    confidence: float
    reasoning: str
    estimated_files: int = 3
    needs_research: bool = False
    needs_self_critique: bool = False

    def phases_to_run(self) -> list[str]:
        """
        根据复杂度返回要执行的阶段列表。

        TODO: 实现阶段选择逻辑（10-20 行）：

        SIMPLE → ["discovery", "quick_spec", "validation"]

        STANDARD → ["discovery", "requirements", "context", "spec_writing", "planning", "validation"]
          如果 needs_research=True，在 context 之前插入 "research"

        COMPLEX → ["discovery", "requirements", "research", "context",
                   "spec_writing", "self_critique", "planning", "validation"]
          self_critique 必须在 spec_writing 之后、planning 之前

        提示：用列表动态构建，需要某些阶段时使用 list.append() 或 list.insert()
        """
        # YOUR CODE HERE
        pass


SIMPLE_KEYWORDS = [
    "fix", "typo", "update", "change", "rename", "delete",
    "style", "color", "text", "button", "margin"
]

COMPLEX_KEYWORDS = [
    "integrate", "integration", "api", "sdk", "database", "migrate",
    "docker", "authentication", "oauth", "microservice",
    "refactor", "architecture", "websocket", "kafka"
]


def assess_complexity(task_description: str) -> ComplexityAssessment:
    """
    基于关键词启发式评估任务复杂度。

    TODO: 实现复杂度评估逻辑（20-30 行）：

    算法：
    1. 将 task_description 转为小写
    2. 统计 SIMPLE_KEYWORDS 和 COMPLEX_KEYWORDS 的命中数
    3. 检测外部集成（oauth, database, redis, docker, payment 等）
    4. 估算文件数（简单任务 1-2 个，标准 5 个，复杂 15 个）
    5. 根据信号决定复杂度：
       - 有 >= 2 个复杂关键词，或有外部集成 → COMPLEX（置信度 0.85）
       - 只有简单关键词，无复杂词 → SIMPLE（置信度 0.90）
       - 否则 → STANDARD（置信度 0.75）
    6. 如果检测到 database/api/oauth 等集成词，设置 needs_research=True

    返回 ComplexityAssessment 对象
    """
    # YOUR CODE HERE
    pass


# === 验证 Part A ===
def test_complexity():
    cases = [
        ("Fix the typo in the login button text", Complexity.SIMPLE),
        ("Add user authentication with OAuth and database integration", Complexity.COMPLEX),
        ("Create a new settings page with form validation", Complexity.STANDARD),
        ("Integrate Stripe payment gateway with webhook handling and database recording",
         Complexity.COMPLEX),
    ]

    for desc, expected in cases:
        result = assess_complexity(desc)
        status = "[PASS]" if result.complexity == expected else "[FAIL]"
        print(f"{status} '{desc[:50]}...' → {result.complexity.value} "
              f"(expected: {expected.value})")

# 接续 lab_10_spec_pipeline.py
import asyncio
from typing import Callable, Any


@dataclass
class PhaseResult:
    phase_name: str
    success: bool
    output: dict = field(default_factory=dict)
    error: str | None = None


@dataclass
class SpecPipelineResult:
    success: bool
    complexity: Complexity
    phases_executed: list[str]
    outputs: dict[str, dict]  # phase_name → output data
    error: str | None = None

    def get_spec(self) -> str | None:
        """从输出中提取规格内容。"""
        for phase in ["spec_writing", "quick_spec"]:
            if phase in self.outputs:
                return self.outputs[phase].get("spec_content")
        return None


class MiniSpecPipeline:
    """
    迷你规格创建流水线。

    将模拟的阶段函数按照复杂度动态组合执行。
    """

    def __init__(
        self,
        task_description: str,
        phase_handlers: dict[str, Callable],
        verbose: bool = True,
    ):
        self.task_description = task_description
        self.phase_handlers = phase_handlers
        self.verbose = verbose
        self._outputs: dict[str, dict] = {}

    async def _run_phase(self, phase_name: str) -> PhaseResult:
        """运行单个阶段，带错误处理。"""
        if phase_name not in self.phase_handlers:
            return PhaseResult(phase_name, False, error=f"No handler for phase: {phase_name}")

        try:
            handler = self.phase_handlers[phase_name]
            output = await handler(self.task_description, self._outputs)
            self._outputs[phase_name] = output
            return PhaseResult(phase_name, True, output=output)
        except Exception as e:
            return PhaseResult(phase_name, False, error=str(e))

    async def run(self) -> SpecPipelineResult:
        """
        运行完整的规格创建流水线。

        TODO: 实现流水线执行逻辑（15-25 行）：

        步骤：
        1. 调用 assess_complexity() 评估任务复杂度
        2. 获取要执行的阶段列表（assessment.phases_to_run()）
        3. 按顺序执行每个阶段（调用 self._run_phase()）
        4. 如果任意阶段失败，停止并返回失败的 SpecPipelineResult
        5. 如果 verbose=True，打印每个阶段的执行状态
        6. 所有阶段成功后，返回成功的 SpecPipelineResult

        提示：
        - 用 phases_executed 列表追踪已执行的阶段
        - SpecPipelineResult 的 outputs 字段对应 self._outputs
        """
        # YOUR CODE HERE
        pass


# === 模拟阶段处理器（用于测试）===

async def mock_discovery(task: str, prev: dict) -> dict:
    """模拟发现阶段：扫描项目文件。"""
    await asyncio.sleep(0.01)  # 模拟耗时
    return {
        "file_count": 47,
        "languages": ["python", "typescript"],
        "has_tests": True,
    }

async def mock_requirements(task: str, prev: dict) -> dict:
    """模拟需求收集：结构化任务描述。"""
    await asyncio.sleep(0.01)
    return {
        "task_description": task,
        "acceptance_criteria": [
            "功能按预期工作",
            "已有测试通过",
            "代码符合风格规范",
        ],
        "constraints": ["不破坏现有 API"],
    }

async def mock_spec_writing(task: str, prev: dict) -> dict:
    """模拟规格写作：生成 spec.md 内容。"""
    await asyncio.sleep(0.01)
    requirements = prev.get("requirements", {})
    criteria = requirements.get("acceptance_criteria", [])
    spec_content = f"""# 规格：{task}

## 目标
{task}

## 验收标准
{chr(10).join(f"- {c}" for c in criteria)}

## 实现方案
[详细实现描述]
"""
    return {"spec_content": spec_content}

async def mock_quick_spec(task: str, prev: dict) -> dict:
    """简单任务的快速规格（跳过多步流程）。"""
    await asyncio.sleep(0.01)
    return {"spec_content": f"# Quick Spec: {task}\n\n简单修改，预计 1-2 个文件。"}

async def mock_planning(task: str, prev: dict) -> dict:
    """模拟规划：生成实现计划。"""
    await asyncio.sleep(0.01)
    return {
        "subtasks": [
            {"id": "1", "description": "实现核心逻辑", "status": "pending"},
            {"id": "2", "description": "添加测试", "status": "pending"},
        ]
    }

async def mock_validation(task: str, prev: dict) -> dict:
    """模拟验证：检查规格完整性。"""
    has_spec = "spec_writing" in prev or "quick_spec" in prev
    has_plan = "planning" in prev
    return {
        "valid": has_spec and has_plan,
        "checks": {
            "has_spec": has_spec,
            "has_plan": has_plan,
        }
    }

# 其他阶段（仅返回空结果，不影响核心流程）
async def mock_research(task: str, prev: dict) -> dict:
    return {"research_notes": "相关代码已分析"}

async def mock_context(task: str, prev: dict) -> dict:
    return {"relevant_files": ["src/auth.py", "src/models.py"]}

async def mock_self_critique(task: str, prev: dict) -> dict:
    spec = prev.get("spec_writing", {}).get("spec_content", "")
    passes = len(spec) > 50  # 简单验证
    return {"passes": passes, "issues": []}


MOCK_HANDLERS = {
    "discovery": mock_discovery,
    "requirements": mock_requirements,
    "research": mock_research,
    "context": mock_context,
    "spec_writing": mock_spec_writing,
    "quick_spec": mock_quick_spec,
    "self_critique": mock_self_critique,
    "planning": mock_planning,
    "validation": mock_validation,
}


async def test_pipeline():
    """测试流水线端到端。"""
    test_cases = [
        ("Fix the typo in button label", Complexity.SIMPLE),
        ("Add user profile settings page with form validation", Complexity.STANDARD),
        ("Integrate OAuth2 authentication with database user management", Complexity.COMPLEX),
    ]

    for task, expected_complexity in test_cases:
        print(f"\n--- Task: {task[:50]} ---")
        pipeline = MiniSpecPipeline(task, MOCK_HANDLERS, verbose=True)
        result = await pipeline.run()

        assert result.complexity == expected_complexity, \
            f"Expected {expected_complexity.value}, got {result.complexity.value}"
        assert result.success, f"Pipeline should succeed for: {task}"
        assert result.get_spec() is not None, "Should have spec content"
        assert "validation" in result.phases_executed, "Validation should run"
        print(f"  Result: {result.complexity.value}, {len(result.phases_executed)} phases")

    print("\n所有流水线测试通过！")

# 接续 lab_10_spec_pipeline.py

@dataclass
class SpecValidationResult:
    valid: bool
    missing_sections: list[str] = field(default_factory=list)
    warnings: list[str] = field(default_factory=list)
    score: float = 0.0   # 0.0 到 1.0


def validate_spec(spec_content: str) -> SpecValidationResult:
    """
    验证规格文档的完整性。

    TODO: 实现规格验证逻辑（15-20 行）：

    必要章节（缺失则 valid=False）：
    - "# " 开头的标题（文档存在）
    - "目标" 或 "Goal" 或 "Overview"
    - "验收标准" 或 "Acceptance" 或 "Requirements"

    可选章节（缺失则加入 warnings）：
    - "实现" 或 "Implementation"
    - "测试" 或 "Test"
    - "约束" 或 "Constraint"

    评分规则：
    - 必要章节每个占 30%（共 3 个必要章节，满分 90%）
    - 可选章节每个占 5%（共 3 个）
    - score 限制在 0.0 到 1.0 之间

    提示：使用 re.search() 进行大小写不敏感的匹配
    """
    # YOUR CODE HERE
    pass


async def main():
    print("=== Part A: 复杂度评估测试 ===")
    test_complexity()

    print("\n=== Part B: 流水线测试 ===")
    await test_pipeline()

    print("\n=== Part C: 规格验证测试 ===")
    good_spec = """# 用户登录功能规格

## 目标
实现安全的用户登录功能。

## 验收标准
- 用户可以用邮件和密码登录
- 失败 5 次后锁定账户

## 实现方案
使用 JWT token，15 分钟过期。

## 测试要求
单元测试覆盖率 > 80%
"""
    result = validate_spec(good_spec)
    assert result.valid, f"Good spec should be valid: {result.missing_sections}"
    assert result.score >= 0.9, f"Score should be high: {result.score}"
    print(f"[PASS] Good spec: valid={result.valid}, score={result.score:.1%}")

    bad_spec = "# Something\n\nJust some text without structure."
    result = validate_spec(bad_spec)
    assert not result.valid, "Bad spec should fail"
    assert len(result.missing_sections) > 0
    print(f"[PASS] Bad spec: valid={result.valid}, missing={result.missing_sections}")

    print("\n全部测试通过！")


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