Claude Code Harness工程：三Agent自主编码实战

传统模式：  开发者 → Claude Code → 代码
Harness模式：需求 → Planner → Generator → Evaluator → 交付物
                                       ↑____________↓（反馈循环）

graph TB
    subgraph "Harness Orchestrator"
        P[Planner Agent<br/>任务拆解与调度]
        G[Generator Agent<br/>代码生成与实现]
        E[Evaluator Agent<br/>质量审查与反馈]
    end

    REQ[项目需求文档] --> P
    P --> |任务列表+约束条件| G
    G --> |代码变更| E
    E --> |通过✓| OUT[交付物]
    E --> |不通过✗<br/>具体问题| G

    style P fill:#4A90D9,color:#fff
    style G fill:#7B68EE,color:#fff
    style E fill:#E74C3C,color:#fff

PLANNER_SYSTEM_PROMPT = """
You are a software architect. Given a project specification:
1. Break it into atomic, ordered tasks
2. Define file paths and module boundaries for each task
3. Specify acceptance criteria for each task
4. Set dependency relationships between tasks

Output format: JSON task list with:
- task_id, description, dependencies[]
- target_files[], acceptance_criteria[]
- estimated_complexity (low/medium/high)

Rules:
- Each task should be completable in <500 lines of code
- Database tasks must precede API tasks
- API tasks must precede frontend tasks
- Include a "project scaffold" task as task_id=0
"""

GENERATOR_SYSTEM_PROMPT = """
You are an expert software developer. You will receive a task 
from the planner with clear specifications.

For each task:
1. Read existing codebase context (CLAUDE.md, related files)
2. Implement the specified feature/fix
3. Write unit tests for your implementation
4. Update CLAUDE.md if architecture decisions changed

Critical rules:
- Follow the exact file paths specified by the planner
- Do NOT modify files outside your task scope
- If you encounter a blocker, output: BLOCKED: {reason}
- Run tests after implementation. If tests fail, fix before proceeding
"""

EVALUATOR_SYSTEM_PROMPT = """
You are a senior code reviewer and QA engineer. 
Evaluate the generated code against these criteria:

1. FUNCTIONALITY: Does it satisfy the task acceptance criteria?
2. ARCHITECTURE: Does it follow project conventions (check CLAUDE.md)?
3. SECURITY: Any SQL injection, XSS, or OWASP Top 10 issues?
4. PERFORMANCE: N+1 queries, unnecessary re-renders, memory leaks?
5. TESTING: Are tests meaningful? Do they cover edge cases?

Output format:
{
  "passed": true/false,
  "score": 0-100,
  "issues": [
    {"severity": "critical|warning|info", 
     "file": "path/to/file",
     "line": 42,
     "message": "Description of the issue",
     "suggestion": "How to fix it"}
  ]
}

A score >= 80 is required to pass. 
All critical issues must be resolved.
"""

# 创建Harness项目
mkdir claude-harness && cd claude-harness
python -m venv .venv
source .venv/bin/activate
pip install claude-agent-sdk asyncio python-dotenv

# 项目结构
mkdir -p harness/agents harness/prompts harness/checkpoints output
touch harness/__init__.py
touch harness/agents/__init__.py

# harness/orchestrator.py
import json
import os
import asyncio
from datetime import datetime
from pathlib import Path
from claude_agent_sdk import Agent, AgentConfig

class HarnessOrchestrator:
    """
    三Agent Harness编排器
    管理 Planner → Generator → Evaluator 的完整生命周期
    """

    def __init__(self, project_spec: str, output_dir: str = "./output"):
        self.project_spec = project_spec
        self.output_dir = Path(output_dir)
        self.checkpoint_dir = self.output_dir / "checkpoints"
        self.checkpoint_dir.mkdir(parents=True, exist_ok=True)

        self.max_iterations = 5  # 每个任务的最大生成-评估循环
        self.quality_threshold = 80  # 质量分门槛

        # 加载Agent配置
        self.planner = self._create_planner()
        self.generator = self._create_generator()
        self.evaluator = self._create_evaluator()

        # 状态追踪
        self.task_list = []
        self.completed_tasks = []
        self.failed_tasks = []
        self.iteration_log = []

    def _create_planner(self) -> Agent:
        return Agent(
            config=AgentConfig(
                model="claude-sonnet-4-6",
                system_prompt=self._load_prompt("planner.md"),
                allowed_tools=["Read", "Glob", "Grep", "Write"],
                max_turns=20
            )
        )

    def _create_generator(self) -> Agent:
        return Agent(
            config=AgentConfig(
                model="claude-sonnet-4-6",
                system_prompt=self._load_prompt("generator.md"),
                allowed_tools=[
                    "Read", "Write", "Edit", "Bash",
                    "Glob", "Grep", "TodoWrite"
                ],
                max_turns=50
            )
        )

    def _create_evaluator(self) -> Agent:
        return Agent(
            config=AgentConfig(
                model="claude-sonnet-4-6",
                system_prompt=self._load_prompt("evaluator.md"),
                allowed_tools=["Read", "Glob", "Grep", "Bash"],
                max_turns=30
            )
        )

    def _load_prompt(self, filename: str) -> str:
        prompt_path = Path(__file__).parent / "prompts" / filename
        return prompt_path.read_text()

    async def run(self):
        """执行完整的Harness流程"""
        print(f"[Harness] 启动项目: {self.project_spec[:100]}...")
        start_time = datetime.now()

        # Phase 1: 规划
        print("[Harness] Phase 1: 任务规划中...")
        self.task_list = await self._plan()
        self._save_checkpoint("plan", self.task_list)

        print(f"[Harness] 规划完成: {len(self.task_list)} 个任务")

        # Phase 2: 逐任务执行（生成-评估循环）
        for i, task in enumerate(self.task_list):
            print(f"\n[Harness] Phase 2: 执行任务 {i+1}/{len(self.task_list)}")
            print(f"  → {task['description']}")

            success = await self._execute_task_with_eval_loop(task)

            if success:
                self.completed_tasks.append(task["task_id"])
                self._save_checkpoint(f"task_{task['task_id']}_done", {
                    "task": task, "status": "completed",
                    "timestamp": datetime.now().isoformat()
                })
            else:
                self.failed_tasks.append(task["task_id"])
                self._save_checkpoint(f"task_{task['task_id']}_fail", {
                    "task": task, "status": "failed",
                    "timestamp": datetime.now().isoformat()
                })

        # Phase 3: 最终验证
        print("\n[Harness] Phase 3: 最终集成验证...")
        final_result = await self._final_integration_test()

        elapsed = (datetime.now() - start_time).total_seconds()
        self._print_report(elapsed, final_result)

    async def _plan(self) -> list:
        """调用Planner Agent生成任务列表"""
        response = await self.planner.run(
            f"Analyze this project spec and create an ordered task list:\n\n"
            f"{self.project_spec}\n\n"
            f"Output valid JSON only."
        )

        # 解析Planner输出
        try:
            tasks = json.loads(response.content)
            assert isinstance(tasks, list)
            return tasks
        except (json.JSONDecodeError, AssertionError):
            # Fallback: 让Agent重新生成
            response = await self.planner.run(
                "Your previous output was not valid JSON. "
                "Please output ONLY a JSON array of task objects."
            )
            return json.loads(response.content)

    async def _execute_task_with_eval_loop(self, task: dict) -> bool:
        """
        对单个任务执行 Generator → Evaluator 循环
        最多尝试 max_iterations 次
        """
        task_context = json.dumps(task, indent=2)

        for iteration in range(1, self.max_iterations + 1):
            print(f"  [迭代 {iteration}/{self.max_iterations}]")

            # Step 1: Generator 执行任务
            gen_prompt = (
                f"Execute this task:\n{task_context}\n\n"
                f"Completed tasks so far: {self.completed_tasks}\n"
                f"Working directory: {self.output_dir}\n"
                f"Iteration: {iteration}"
            )

            if iteration > 1:
                gen_prompt += (
                    f"\n\nPrevious evaluation feedback:\n"
                    f"{last_eval_feedback}"
                )

            gen_result = await self.generator.run(gen_prompt)

            # Step 2: Evaluator 审查
            eval_prompt = (
                f"Review the implementation for this task:\n"
                f"{task_context}\n\n"
                f"Working directory: {self.output_dir}\n"
                f"Focus on acceptance_criteria: "
                f"{task.get('acceptance_criteria', [])}"
            )

            eval_result = await self.evaluator.run(eval_prompt)

            try:
                evaluation = json.loads(eval_result.content)
            except json.JSONDecodeError:
                evaluation = {"passed": False, "score": 0,
                            "issues": [{"message": "Invalid eval output"}]}

            self.iteration_log.append({
                "task_id": task["task_id"],
                "iteration": iteration,
                "score": evaluation.get("score", 0),
                "passed": evaluation.get("passed", False)
            })

            print(f"    质量分: {evaluation.get('score', 0)}/100")

            if (evaluation.get("passed", False) and 
                evaluation.get("score", 0) >= self.quality_threshold):
                print(f"    ✓ 通过!")
                return True

            # 未通过，记录反馈用于下一轮迭代
            last_eval_feedback = json.dumps(
                evaluation.get("issues", []), indent=2
            )
            print(f"    ✗ 未通过，进入下一轮迭代...")

        print(f"  [达到最大迭代次数 {self.max_iterations}，任务失败]")
        return False

    async def _final_integration_test(self) -> dict:
        """最终集成测试"""
        eval_prompt = (
            f"Perform final integration review on the project "
            f"in {self.output_dir}.\n"
            f"Check: build success, test pass, no critical issues.\n"
            f"Output JSON: {{\"build_passed\": bool, "
            f"\"tests_passed\": bool, \"score\": int}}"
        )
        result = await self.evaluator.run(eval_prompt)
        try:
            return json.loads(result.content)
        except json.JSONDecodeError:
            return {"build_passed": False, "tests_passed": False, 
                    "score": 0}

    def _save_checkpoint(self, name: str, data: dict):
        """保存检查点，支持从任意位置恢复"""
        path = self.checkpoint_dir / f"{name}.json"
        path.write_text(json.dumps(data, indent=2, ensure_ascii=False))

    def _print_report(self, elapsed: float, final: dict):
        """打印执行报告"""
        print("\n" + "=" * 60)
        print("  HARNESS EXECUTION REPORT")
        print("=" * 60)
        print(f"  总耗时: {elapsed:.1f}s ({elapsed/60:.1f}min)")
        print(f"  任务完成: {len(self.completed_tasks)}/{len(self.task_list)}")
        print(f"  任务失败: {len(self.failed_tasks)}/{len(self.task_list)}")
        print(f"  总迭代次数: {len(self.iteration_log)}")
        print(f"  构建状态: {'✓' if final.get('build_passed') else '✗'}")
        print(f"  测试状态: {'✓' if final.get('tests_passed') else '✗'}")
        print(f"  最终质量分: {final.get('score', 0)}/100")

        if self.failed_tasks:
            print(f"\n  失败任务: {self.failed_tasks}")
            print("  建议: 检查checkpoint目录中的失败详情")
        print("=" * 60)

<!-- harness/prompts/planner.md -->
# Role: Software Architect (Planner Agent)

You are a senior software architect. Your job is to break down 
project specifications into ordered, atomic tasks.

## Rules
1. Each task must be independently verifiable
2. Respect dependency order: infra → data → logic → interface
3. Every task has clear acceptance_criteria
4. Scaffold task (task_id=0) is always first
5. Total tasks should be 5-15 for a typical project

## Output Format
Output ONLY a JSON array:
```json
[
  {
    "task_id": 0,
    "description": "Initialize project scaffold with ...",
    "dependencies": [],
    "target_files": ["package.json", "tsconfig.json", ...],
    "acceptance_criteria": [
      "package.json exists with correct dependencies",
      "npm install succeeds without errors"
    ],
    "estimated_complexity": "low"
  }
]

```markdown
<!-- harness/prompts/generator.md -->
# Role: Expert Developer (Generator Agent)

You are a senior full-stack developer. Execute tasks precisely 
as specified by the planner.

## Workflow
1. Read task specification carefully
2. Check CLAUDE.md for project conventions
3. Read relevant existing files before editing
4. Implement the feature/fix
5. Write tests (aim for >80% coverage on new code)
6. Run tests and fix failures
7. Output summary of changes made

## Critical Rules
- ONLY modify files listed in target_files or newly created files
- Follow existing code style exactly
- If blocked, output: BLOCKED: {clear reason}
- Never skip writing tests
- Run linter after implementation

<!-- harness/prompts/evaluator.md -->
# Role: Senior Code Reviewer & QA (Evaluator Agent)

You are a strict but fair code reviewer. Evaluate implementations 
against acceptance criteria and best practices.

## Evaluation Criteria
1. **FUNCTIONALITY** (40%): Does it meet acceptance criteria?
2. **CODE QUALITY** (20%): Clean code, proper abstractions
3. **SECURITY** (20%): No OWASP Top 10 vulnerabilities
4. **TESTING** (10%): Meaningful tests with good coverage
5. **PERFORMANCE** (10%): No obvious bottlenecks

## Output Format
Output ONLY valid JSON:
```json
{
  "passed": true/false,
  "score": 0-100,
  "issues": [
    {
      "severity": "critical|warning|info",
      "file": "path/to/file",
      "line": 42,
      "message": "Issue description",
      "suggestion": "How to fix"
    }
  ]
}

### 3.4 启动Harness

```python
# main.py
import asyncio
from harness.orchestrator import HarnessOrchestrator

PROJECT_SPEC = """
Project: Task Management API (Express.js + PostgreSQL)

Requirements:
1. RESTful API with CRUD operations for tasks
2. Task fields: id, title, description, status, priority, 
   created_at, updated_at
3. Endpoints: GET/POST /tasks, GET/PUT/DELETE /tasks/:id
4. Filter by status and priority
5. Pagination support (limit/offset)
6. Input validation with Joi
7. Unit tests with Jest
8. Docker Compose for local development
9. TypeScript throughout
10. Follow conventional commit format

Tech Stack:
- Runtime: Node.js 20 + TypeScript
- Framework: Express.js
- Database: PostgreSQL 16
- ORM: Drizzle ORM
- Validation: Joi
- Testing: Jest + Supertest
- Container: Docker Compose
"""

async def main():
    harness = HarnessOrchestrator(
        project_spec=PROJECT_SPEC,
        output_dir="./output/task-api"
    )
    await harness.run()

asyncio.run(main())

# 运行
export ANTHROPIC_API_KEY=sk-ant-xxx
python main.py

output/
└── checkpoints/
    ├── plan.json              # Planner生成的任务列表
    ├── task_0_done.json       # 任务0完成检查点
    ├── task_1_done.json       # 任务1完成检查点
    ├── task_2_fail.json       # 任务2失败检查点
    └── task_3_done.json

# harness/recovery.py
import json
from pathlib import Path
from datetime import datetime

class HarnessRecovery:
    """从检查点恢复Harness执行"""

    def __init__(self, checkpoint_dir: str, orchestrator):
        self.checkpoint_dir = Path(checkpoint_dir)
        self.orchestrator = orchestrator

    def get_last_checkpoint(self) -> dict:
        """获取最新的检查点"""
        checkpoints = sorted(
            self.checkpoint_dir.glob("*.json"),
            key=lambda p: p.stat().st_mtime,
            reverse=True
        )

        if not checkpoints:
            return {"phase": "start", "task_index": 0}

        latest = checkpoints[0]
        name = latest.stem  # e.g., "task_2_done"

        if name == "plan":
            return {"phase": "planned", "task_index": 0}
        elif name.endswith("_done"):
            task_id = int(name.split("_")[1])
            return {"phase": "executing", "last_completed": task_id}
        elif name.endswith("_fail"):
            task_id = int(name.split("_")[1])
            return {"phase": "executing", "last_failed": task_id}

        return {"phase": "start", "task_index": 0}

    def load_plan(self) -> list:
        """加载已保存的任务计划"""
        plan_path = self.checkpoint_dir / "plan.json"
        if plan_path.exists():
            return json.loads(plan_path.read_text())
        return None

    async def resume(self):
        """从最新检查点恢复执行"""
        checkpoint = self.get_last_checkpoint()

        if checkpoint["phase"] == "start":
            print("[Recovery] 无检查点，从头开始...")
            await self.orchestrator.run()
            return

        # 恢复任务列表
        task_list = self.load_plan()
        if not task_list:
            print("[Recovery] 检查点损坏，从头开始...")
            await self.orchestrator.run()
            return

        self.orchestrator.task_list = task_list
        completed = checkpoint.get("last_completed", -1)

        print(f"[Recovery] 从任务 {completed + 1} 恢复执行...")

        # 跳过已完成的任务
        remaining = [
            t for t in task_list 
            if int(t["task_id"]) > completed
        ]

        for i, task in enumerate(remaining):
            print(f"  → 执行任务 {task['task_id']}: {task['description']}")
            success = await self.orchestrator._execute_task_with_eval_loop(task)
            if not success:
                print(f"  ✗ 任务 {task['task_id']} 执行失败，停止恢复")
                break

flowchart TD
    START[启动Harness] --> CHECK{存在检查点?}
    CHECK --> |否| PLAN[Phase 1: Planner生成任务]
    CHECK --> |是| LOAD[加载检查点]
    LOAD --> FIND[定位最后完成的任务]
    FIND --> SKIP[跳过已完成任务]

    PLAN --> TASKS[Phase 2: 逐任务执行]
    SKIP --> TASKS

    TASKS --> GEN[Generator执行]
    GEN --> EVAL[Evaluator审查]
    EVAL --> PASS{质量分 >= 80?}
    PASS --> |是| SAVE[保存检查点]
    PASS --> |否| ITER{迭代 < 5次?}
    ITER --> |是| FEED[反馈给Generator]
    FEED --> GEN
    ITER --> |否| FAIL[标记任务失败]

    SAVE --> NEXT{还有任务?}
    FAIL --> NEXT
    NEXT --> |是| TASKS
    NEXT --> |否| FINAL[Phase 3: 集成验证]
    FINAL --> REPORT[生成报告]

    style PASS fill:#27AE60,color:#fff
    style FAIL fill:#E74C3C,color:#fff
    style SAVE fill:#4A90D9,color:#fff

import re

def extract_json(text: str) -> dict | list:
    """从可能包含Markdown的文本中提取JSON"""
    # 尝试直接解析
    try:
        return json.loads(text)
    except json.JSONDecodeError:
        pass

    # 尝试提取代码块中的JSON
    pattern = r'```(?:json)?\s*\n?(.*?)\n?```'
    matches = re.findall(pattern, text, re.DOTALL)
    for match in matches:
        try:
            return json.loads(match.strip())
        except json.JSONDecodeError:
            continue

    # 尝试找第一个 [ 或 { 到最后一个 ] 或 }
    for start_char, end_char in [('[', ']'), ('{', '}')]:
        start = text.find(start_char)
        end = text.rfind(end_char)
        if start != -1 and end != -1:
            try:
                return json.loads(text[start:end+1])
            except json.JSONDecodeError:
                continue

    raise ValueError(f"无法从文本中提取JSON: {text[:200]}...")

def _check_file_boundary(self, task: dict, 
                         changed_files: list) -> list:
    """检查Generator是否越界修改文件"""
    allowed = set(task.get("target_files", []))
    violations = []

    for f in changed_files:
        if f not in allowed and not f.startswith("test/"):
            violations.append({
                "severity": "critical",
                "file": f,
                "message": f"Modified file outside task scope: {f}",
                "suggestion": f"Revert changes to {f}"
            })

    return violations

TOKEN_BUDGET = {
    "planner": {"input": 10000, "output": 4000},
    "generator": {"input": 30000, "output": 8000},
    "evaluator": {"input": 20000, "output": 4000},
}

class TokenBudgetManager:
    def __init__(self, budgets: dict):
        self.budgets = budgets
        self.consumed = {"planner": 0, "generator": 0, "evaluator": 0}

    def can_proceed(self, agent: str, estimated_tokens: int) -> bool:
        remaining = (self.budgets[agent]["input"] + 
                    self.budgets[agent]["output"] - 
                    self.consumed[agent])
        return estimated_tokens < remaining

    def record_usage(self, agent: str, tokens: int):
        self.consumed[agent] += tokens

    def get_remaining(self, agent: str) -> int:
        total = (self.budgets[agent]["input"] + 
                self.budgets[agent]["output"])
        return total - self.consumed[agent]

class AdaptiveEvaluator:
    """根据历史评分分布动态调整阈值"""

    def __init__(self, base_threshold: int = 80):
        self.base_threshold = base_threshold
        self.score_history = []

    def get_adaptive_threshold(self, task_type: str) -> int:
        recent = [s for s in self.score_history[-10:]]
        if not recent:
            return self.base_threshold

        avg = sum(recent) / len(recent)

        # 如果最近评分偏高，提高门槛
        if avg > 90:
            return min(self.base_threshold + 10, 95)
        # 如果最近评分偏低，适当降低
        elif avg < 60:
            return max(self.base_threshold - 10, 60)

        return self.base_threshold

    def record_score(self, score: int):
        self.score_history.append(score)

# Task Management API - Project Specification

## Overview
Build a production-ready RESTful Task Management API.

## Technical Requirements
- Runtime: Node.js 20 LTS + TypeScript 5.x
- Framework: Express.js 4.x
- Database: PostgreSQL 16
- ORM: Drizzle ORM
- Validation: Zod
- Testing: Vitest + Supertest
- Containerization: Docker Compose

## API Endpoints
### Tasks
- POST   /api/tasks          - Create task
- GET    /api/tasks          - List tasks (pagination + filters)
- GET    /api/tasks/:id      - Get task by ID
- PUT    /api/tasks/:id      - Update task
- DELETE /api/tasks/:id      - Delete task

### Query Parameters for GET /api/tasks
- status: filter by status (pending|in_progress|completed)
- priority: filter by priority (low|medium|high)
- page: page number (default: 1)
- limit: items per page (default: 20, max: 100)

## Data Model
Task {
  id: UUID (auto-generated)
  title: string (required, max 200 chars)
  description: string (optional, max 2000 chars)
  status: enum [pending, in_progress, completed]
  priority: enum [low, medium, high]
  due_date: timestamp (optional)
  created_at: timestamp (auto)
  updated_at: timestamp (auto)
}

## Non-Functional Requirements
- Input validation on all endpoints
- Proper HTTP status codes
- Error response format: {error: string, details?: any}
- Database migrations
- Health check endpoint: GET /health
- Docker Compose for local dev
- Test coverage > 80%

# 方式1: 直接运行
python main.py

# 方式2: 从检查点恢复（如果中途中断）
python recover.py --checkpoint ./output/task-api/checkpoints

# 方式3: 无头模式后台运行（长时间任务推荐）
nohup python main.py > harness.log 2>&1 &
tail -f harness.log

[Harness] 启动项目: Task Management API - Project Specification...
[Harness] Phase 1: 任务规划中...
[Harness] 规划完成: 8 个任务

[Harness] Phase 2: 执行任务 1/8
  → Initialize project scaffold with package.json, tsconfig, Docker
  [迭代 1/5]
    质量分: 85/100
    ✓ 通过!

[Harness] Phase 2: 执行任务 2/8
  → Define database schema and Drizzle ORM configuration
  [迭代 1/5]
    质量分: 72/100
    ✗ 未通过，进入下一轮迭代...
  [迭代 2/5]
    质量分: 88/100
    ✓ 通过!

[Harness] Phase 2: 执行任务 3/8
  → Implement task repository layer with CRUD operations
...

[Harness] Phase 3: 最终集成验证...

============================================================
  HARNESS EXECUTION REPORT
============================================================
  总耗时: 847.3s (14.1min)
  任务完成: 8/8
  任务失败: 0/8
  总迭代次数: 11
  构建状态: ✓
  测试状态: ✓
  最终质量分: 91/100
============================================================

output/task-api/
├── src/
│   ├── index.ts                 # 入口，Express配置
│   ├── db/
│   │   ├── schema.ts            # Drizzle ORM schema定义
│   │   ├── client.ts            # 数据库连接
│   │   └── migrations/          # 迁移文件
│   ├── routes/
│   │   └── tasks.ts             # 任务路由
│   ├── services/
│   │   └── taskService.ts       # 业务逻辑层
│   ├── middleware/
│   │   ├── validation.ts        # Zod验证中间件
│   │   └── errorHandler.ts      # 统一错误处理
│   └── types/
│       └── task.ts              # TypeScript类型定义
├── tests/
│   ├── setup.ts                 # 测试配置
│   ├── routes/
│   │   └── tasks.test.ts        # API集成测试
│   └── services/
│       └── taskService.test.ts  # 单元测试
├── docker-compose.yml
├── Dockerfile
├── package.json
├── tsconfig.json
├── vitest.config.ts
└── CLAUDE.md                    # 项目约定文档

# harness/hooks.py
import os
import subprocess

def notify_completion(report: dict):
    """Harness完成时发送桌面通知"""
    summary = (
        f"Claude Harness完成\n"
        f"任务: {report['completed']}/{report['total']}\n"
        f"质量分: {report['score']}/100"
    )

    # macOS
    if os.name != "nt" and os.uname().sysname == "Darwin":
        subprocess.run([
            "osascript", "-e",
            f'display notification "{summary}" with title "Claude Harness"'
        ])
    # Linux
    elif os.name != "nt":
        subprocess.run(["notify-send", "Claude Harness", summary])


def notify_failure(task_id: str, reason: str):
    """任务失败时发送Slack/钉钉通知"""
    webhook_url = os.getenv("NOTIFICATION_WEBHOOK")
    if not webhook_url:
        return

    import urllib.request
    payload = json.dumps({
        "text": f"⚠️ Harness任务 {task_id} 失败: {reason}"
    }).encode()

    req = urllib.request.Request(
        webhook_url,
        data=payload,
        headers={"Content-Type": "application/json"}
    )
    urllib.request.urlopen(req)

MODEL_STRATEGY = {
    # Planner需要强推理能力，使用Sonnet
    "planner": "claude-sonnet-4-6",

    # Generator需要编码能力，Sonnet性价比最优
    "generator": "claude-sonnet-4-6",

    # Evaluator需要细致分析，可以复用Sonnet
    "evaluator": "claude-sonnet-4-6",

    # 如果预算充足，关键任务可以用Opus
    # "planner": "claude-opus-4-6",
    # "generator": "claude-opus-4-6",
}

# 成本估算（基于典型Token使用）
ESTIMATED_COST_PER_TASK = {
    "claude-opus-4-6": 0.35,    # ~$0.35/任务
    "claude-sonnet-4-6": 0.04,  # ~$0.04/任务
    "claude-haiku-4-5": 0.005,  # ~$0.005/任务
}

def estimate_project_cost(task_count: int, 
                          model: str = "claude-sonnet-4-6",
                          avg_iterations: float = 1.5) -> float:
    """估算项目总成本"""
    cost_per = ESTIMATED_COST_PER_TASK[model]
    # 每个任务涉及3个Agent，平均1.5次迭代
    return task_count * avg_iterations * 3 * cost_per

async def _execute_parallel(self):
    """并行执行无依赖的任务"""
    # 构建依赖图
    dep_graph = {}
    for task in self.task_list:
        deps = task.get("dependencies", [])
        dep_graph[task["task_id"]] = set(
            int(d) for d in deps
        )

    completed_ids = set()

    while dep_graph:
        # 找出所有依赖已满足的任务
        ready = [
            tid for tid, deps in dep_graph.items()
            if deps.issubset(completed_ids)
        ]

        if not ready:
            print("[Harness] 死锁! 无法继续执行")
            break

        # 并行执行就绪任务（最多3个并发）
        batches = [ready[i:i+3] for i in range(0, len(ready), 3)]

        for batch in batches:
            tasks = [
                t for t in self.task_list
                if t["task_id"] in batch
            ]

            # 使用独立的Generator实例并行执行
            results = await asyncio.gather(*[
                self._execute_task_with_eval_loop(task)
                for task in tasks
            ])

            for task, success in zip(tasks, results):
                tid = task["task_id"]
                dep_graph.pop(tid, None)
                if success:
                    completed_ids.add(tid)
                    self.completed_tasks.append(tid)
                else:
                    self.failed_tasks.append(tid)

sequenceDiagram
    participant U as 开发者
    participant O as Orchestrator
    participant P as Planner
    participant G as Generator
    participant E as Evaluator
    participant FS as 文件系统

    U->>O: 提交项目需求
    O->>P: 解析需求，生成任务列表
    P-->>O: 返回JSON任务列表

    loop 每个任务
        O->>O: 保存检查点
        O->>G: 发送任务描述+约束

        loop 生成-评估循环 (max 5次)
            G->>FS: 读取现有代码
            G->>FS: 写入/修改文件
            G->>FS: 运行测试
            G-->>O: 实现完成

            O->>E: 发送代码进行审查
            E->>FS: 读取变更文件
            E->>FS: 运行测试
            E-->>O: 返回评分和问题

            alt 评分 >= 80
                O->>O: 任务完成，保存检查点
            else 评分 < 80
                O->>G: 发送反馈，重新实现
            end
        end
    end

    O->>E: 最终集成验证
    E-->>O: 最终评分
    O-->>U: 执行报告