研究提案

HarnessForge 开发文档：基于 OpenHarness 的实现规划

HarnessForge 多 Agent 任务工厂的具体开发计划，基于 HKUDS/OpenHarness 框架实现，包含架构设计、API 调用方式、目录结构和分阶段交付计划

2026年4月12日 HarnessForge OpenHarness 开发文档实现 Agent SDK

HarnessForge 开发文档

状态：待确认 → 开发中依赖：OpenHarness v0.1.6+（HKUDS/OpenHarness, MIT） 预计工期：Phase 1 约 3-5 天

🔴 需要确认的事项

在开始开发前，请确认以下几点：

1. API Key 和模型选择

角色	推荐模型	原因	需要确认
Task Designer	claude-sonnet-4.5	分析 repo 架构，中等能力够用	是否用其他模型？
PM	claude-opus-4.6	需要最强理解力写需求和 hint	OK？
Dev	claude-sonnet-4.5	主力执行，平衡成本和能力	OK？还是用更便宜的？
Reviewer	claude-opus-4.6	需要判断力	OK？

所有模型通过 ANTHROPIC_API_KEY 调用。如果要用其他 provider（OpenAI、本地 vLLM），需要额外配置。

2. 首批 Repo

Phase 1 验证选 3 个 repo。建议：

Repo	语言	Release 数	理由
`django/django`	Python	100+	最经典的 Web 框架，release notes 详尽
`vercel/next.js`	TypeScript	200+	前端框架，feature 丰富
`gin-gonic/gin`	Go	20+	Go Web 框架，代码量适中

要换其他 repo 吗？

3. 运行环境

本机 macOS 执行？还是服务器？
Docker Desktop 已安装？
预算上限？（估算 Phase 1 约 $50-100 API 费用）

4. 产出存放位置

代码放在 tuntang 知识库内（tools/harnessforge/）？
还是独立仓库？

架构概览

tools/harnessforge/
│
├── forge.py                 # 主入口：python forge.py --repo django/django
├── config.yaml              # 全局配置（模型、预算、repo 列表）
│
├── agents/                  # 各 Agent 角色实现
│   ├── __init__.py
│   ├── task_designer.py     # Step 1: 分析 repo，设计 feature 任务
│   ├── pm.py                # Step 2: 需求重写 + 渐进提示
│   ├── dev.py               # Step 3: 在 Docker 沙箱中执行开发
│   ├── test_runner.py       # Step 4: 运行测试，反馈给 Dev
│   └── reviewer.py          # Step 5: 审核轨迹，输出最终 task
│
├── orchestrator.py          # Dev + PM 交互循环（核心编排逻辑）
├── sandbox.py               # Docker 沙箱管理（基于 OpenHarness SandboxPolicy）
├── trajectory.py            # 轨迹记录和序列化
│
├── prompts/                 # System prompt 模板
│   ├── task_designer.md
│   ├── pm.md
│   ├── dev.md
│   └── reviewer.md
│
├── schemas/                 # 输出 JSON Schema
│   ├── blueprint.json
│   ├── task.json
│   └── verdict.json
│
└── output/                  # 运行时产出
    ├── blueprints/          # Task Designer 输出
    ├── tasks/               # PM 输出
    ├── trajectories/        # Dev 完整轨迹
    └── verified/            # Reviewer 验证后的最终 task

技术选型：为什么用 OpenHarness

不是因为它是 v1 proposal 里提到的，而是因为它恰好提供了我们需要的三个核心能力：

1. `harness.run()` — 一行代码启动 Agent

import harness

async for msg in harness.run(
    prompt="完成以下开发任务...",
    provider="anthropic",
    model="claude-sonnet-4-20250514",
    permission_mode="bypass",    # 沙箱内自动批准
    sandbox_mode="docker",       # Docker 隔离
    max_turns=50,
):
    if isinstance(msg, harness.ToolUse):
        # 每次工具调用都能捕获
        record_trajectory(msg)
    elif isinstance(msg, harness.Result):
        final_output = msg.text

不需要自己实现文件读写、代码搜索、shell 执行——OpenHarness 内置 43+ 工具，Dev Agent 开箱即用。

2. Docker 沙箱 — 安全隔离

from harness.types.sandbox import SandboxPolicy, SandboxMode, ResourceLimits, NetworkPolicy

sandbox = SandboxPolicy(
    mode=SandboxMode.DOCKER,
    docker_image="python:3.12-slim",
    resource_limits=ResourceLimits(
        max_memory_mb=4096,
        max_cpu_seconds=300,
        max_processes=64,
    ),
    network=NetworkPolicy(allow_network=False),
    allowed_paths=("/workspace",),
    blocked_commands=("curl", "wget", "nc"),
)

Dev Agent 在完全隔离的容器中工作：不能上网、不能访问宿主文件系统、API key 自动剥离。

3. 流式事件流 — 实时轨迹捕获

harness.run() 返回异步事件流，每次工具调用（读文件、写代码、跑测试）都是一个独立事件。不需要事后解析日志——轨迹在生成时就被结构化记录。

可以不用 OpenHarness 吗？

可以。前面的 Claude Code CLI 版本也能工作。但区别是：

维度	Claude Code CLI	OpenHarness
轨迹粒度	轮级（每轮一个 JSON）	事件级（每个 tool call）
沙箱控制	通过 `--allowedTools`	完整 Docker 容器 + 资源限制
PM 介入时机	只能在每轮之间	可在任意 tool call 之后
自定义工具	不支持	支持（继承 BaseTool）
依赖	需要 claude CLI	pip install openharness-ai

核心实现

1. `sandbox.py` — 沙箱管理

"""为每个 repo 创建和管理 Docker 沙箱环境"""

import asyncio
import subprocess
from dataclasses import dataclass
from harness.types.sandbox import SandboxPolicy, SandboxMode, ResourceLimits, NetworkPolicy


@dataclass
class RepoSandbox:
    repo: str
    base_tag: str
    container_name: str
    workspace: str = "/workspace"

    def get_policy(self) -> SandboxPolicy:
        return SandboxPolicy(
            mode=SandboxMode.DOCKER,
            docker_image=f"harnessforge-{self.repo.replace('/', '-')}:latest",
            resource_limits=ResourceLimits(
                max_memory_mb=4096,
                max_cpu_seconds=600,  # 10 分钟
                max_processes=128,
            ),
            network=NetworkPolicy(allow_network=False),
            allowed_paths=(self.workspace,),
            blocked_commands=("curl", "wget", "nc", "ssh"),
        )

    async def setup(self):
        """克隆 repo，切到 base_tag，构建 Docker 镜像"""
        image_name = f"harnessforge-{self.repo.replace('/', '-')}"

        # 构建包含 repo 的 Docker 镜像
        dockerfile = f"""
FROM python:3.12-slim
RUN apt-get update && apt-get install -y git ripgrep && rm -rf /var/lib/apt/lists/*
WORKDIR /workspace
RUN git clone --depth=200 https://github.com/{self.repo}.git .
RUN git checkout {self.base_tag}
# 安装依赖（如果有）
RUN if [ -f requirements.txt ]; then pip install -r requirements.txt; fi
RUN if [ -f setup.py ]; then pip install -e .; fi
RUN if [ -f pyproject.toml ]; then pip install -e .; fi
"""
        # 写 Dockerfile 并构建
        proc = await asyncio.create_subprocess_exec(
            "docker", "build", "-t", f"{image_name}:latest", "-f-", ".",
            stdin=asyncio.subprocess.PIPE,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE,
        )
        stdout, stderr = await proc.communicate(dockerfile.encode())
        if proc.returncode != 0:
            raise RuntimeError(f"Docker build failed: {stderr.decode()}")
        return self

2. `orchestrator.py` — Dev + PM 交互循环

这是最关键的部分。

"""Dev + PM 交互循环编排器"""

import asyncio
import harness
from dataclasses import dataclass, field
from typing import Optional
from .trajectory import TrajectoryRecorder


@dataclass
class PMState:
    """PM 观察 Dev 状态，决定是否介入"""
    turn: int = 0
    test_results: list = field(default_factory=list)
    hint_level: int = 0
    hint_released_at: int = 0
    direction_changes: int = 0
    files_touched: set = field(default_factory=set)

    # 配置
    COOLDOWN: int = 8
    L1_TRIGGER_FAILS: int = 3
    L2_TRIGGER_TURNS: int = 5
    L3_TRIGGER_TURNS: int = 5

    def observe_test_result(self, result: dict):
        self.test_results.append(result)
        self.turn += 1

    def observe_tool_call(self, tool_name: str, args: dict):
        """实时观察 Dev 的每次工具调用"""
        if tool_name in ("Write", "Edit"):
            path = args.get("path", "")
            self.files_touched.add(path)

    @property
    def consecutive_same_failures(self) -> int:
        if len(self.test_results) < 2:
            return 0
        count = 1
        for i in range(len(self.test_results) - 2, -1, -1):
            r1, r2 = self.test_results[i], self.test_results[i + 1]
            if (r1.get("status") == "fail" and r2.get("status") == "fail"
                    and r1.get("error_key") == r2.get("error_key")):
                count += 1
            else:
                break
        return count

    @property
    def has_progress(self) -> bool:
        recent = [r for r in self.test_results[-5:] if r.get("status") == "fail"]
        if len(recent) < 2:
            return True
        return recent[-1].get("pass_count", 0) > recent[0].get("pass_count", 0)

    def should_release_hint(self) -> Optional[int]:
        if self.turn < self.COOLDOWN:
            return None
        turns_since = self.turn - self.hint_released_at if self.hint_released_at else 999

        if self.hint_level == 0 and self.consecutive_same_failures >= self.L1_TRIGGER_FAILS:
            return 1
        if self.hint_level == 1 and turns_since >= self.L2_TRIGGER_TURNS and not self.has_progress:
            return 2
        if self.hint_level == 2 and turns_since >= self.L3_TRIGGER_TURNS and not self.has_progress:
            return 3
        return None


async def run_dev_task(task: dict, sandbox: "RepoSandbox") -> dict:
    """执行一个完整的 Dev 任务，带 PM 提示介入"""

    pm = PMState()
    recorder = TrajectoryRecorder(task["task_id"])
    hints = task.get("hints", {})
    test_cmd = task["test_cmd"]

    # 构建初始 prompt
    prompt = f"""你是一个高级软件工程师，在一个真实的开源项目中工作。

## 任务
{task['description']}

## 验收标准
{chr(10).join('- ' + c for c in task['acceptance_criteria'])}

## 规则
1. 先浏览项目结构，理解架构
2. 修改代码后用 `{test_cmd}` 验证
3. 测试全部通过后，输出 TASK_COMPLETE
4. 如果卡住了，分析原因，换个思路
"""

    # 用 harness.run() 启动 Dev Agent
    session_messages = []
    current_prompt = prompt
    final_status = "timeout"

    for attempt in range(50):  # 最多 50 轮
        # ── 检查是否需要释放 hint ──
        new_hint = pm.should_release_hint()
        if new_hint and new_hint > pm.hint_level:
            hint_key = f"L{new_hint}"
            hint_content = hints.get(hint_key, [])
            if hint_content:
                hint_text = "\n".join(f"- {h}" for h in hint_content)
                current_prompt = f"""[技术负责人反馈]
你当前的方向需要调整。这里有一些额外信息：

{hint_text}

请根据这些提示重新审视你的方案。"""
                pm.hint_level = new_hint
                pm.hint_released_at = pm.turn
                recorder.record_hint(pm.turn, new_hint, hint_content)

        # ── 执行一轮 ──
        turn_events = []
        async for msg in harness.run(
            prompt=current_prompt,
            provider="anthropic",
            model=task.get("dev_model", "claude-sonnet-4-20250514"),
            permission_mode="bypass",
            sandbox_mode="docker",
            max_turns=1,  # 每次只执行 1 轮，以便 PM 观察
            # TODO: 传入 session 以保持上下文
        ):
            if isinstance(msg, harness.ToolUse):
                pm.observe_tool_call(msg.tool_name, msg.arguments)
                turn_events.append({
                    "type": "tool_use",
                    "tool": msg.tool_name,
                    "args_preview": str(msg.arguments)[:200],
                })

            elif isinstance(msg, harness.ToolResult):
                turn_events.append({
                    "type": "tool_result",
                    "output_preview": str(msg.output)[:500],
                })
                # 检测是否运行了测试
                if test_cmd_in_output(msg, test_cmd):
                    test_result = parse_test_output(msg.output)
                    pm.observe_test_result(test_result)
                    turn_events.append({"type": "test_result", **test_result})

            elif isinstance(msg, harness.Result):
                if "TASK_COMPLETE" in msg.text:
                    # 最终验证
                    final_test = await run_final_test(sandbox, test_cmd)
                    if final_test["status"] == "pass":
                        final_status = "success"
                        recorder.record_turn(pm.turn, turn_events, final_test)
                        break
                    else:
                        current_prompt = f"测试未全部通过：{final_test['output'][:500]}。请继续修复。"
                        pm.observe_test_result(final_test)

        recorder.record_turn(pm.turn, turn_events, None)
        pm.turn += 1

        # 默认下一轮 prompt
        if final_status != "success":
            current_prompt = "继续你的开发。如果之前有错误，分析原因并修复。"

    # ── 保存结果 ──
    return {
        "task_id": task["task_id"],
        "status": final_status,
        "total_turns": pm.turn,
        "max_hint_level": pm.hint_level,
        "hint_released_at": pm.hint_released_at,
        "trajectory_file": recorder.save(),
    }

3. `agents/task_designer.py`

"""分析 repo release 历史，设计 feature 任务"""

import harness
import json


SYSTEM_PROMPT = """你是一个资深软件架构师。你的任务是分析一个开源仓库的 release 历史，
设计能训练 AI 编码 Agent 的高质量开发任务。

好的任务应该：
1. 基于 repo 真实的 release 特性演进
2. 涉及 3+ 文件的修改
3. 有明确的测试验证方式
4. 难度中高（优秀工程师需 2-8 小时）

对于每个任务：
- single: 基于单个 release 的一个独立特性
- compound: 跨 2 个 release 的关联特性合并

输出严格的 JSON 格式。"""


async def design_tasks(repo: str, base_tag: str, next_tags: list[str]) -> dict:
    """分析 repo 并设计任务"""

    prompt = f"""分析以下仓库的 release 历史并设计开发任务。

仓库: {repo}
基准版本: {base_tag}
后续版本: {', '.join(next_tags)}

步骤：
1. 用 `git log {base_tag}..{next_tags[0]} --oneline` 查看第一个版本的变更
2. 用 `git diff {base_tag}..{next_tags[0]} --stat` 查看文件变更统计
3. 读 CHANGELOG.md 或 release notes
4. 设计 3-5 个 feature 任务

每个任务输出：
{{
  "task_id": "repo-feature-name",
  "type": "single" 或 "compound",
  "description": "技术层面的特性描述",
  "base_commit": "{base_tag}",
  "ground_truth_tag": "对应的目标版本",
  "test_cmd": "测试命令",
  "estimated_difficulty": 1-5,
  "estimated_files": 预估修改文件数,
  "estimated_lines": 预估修改行数
}}"""

    result = None
    async for msg in harness.run(
        prompt=prompt,
        system_prompt=SYSTEM_PROMPT,
        provider="anthropic",
        model="claude-sonnet-4-20250514",
        sandbox_mode="docker",
        permission_mode="bypass",
        max_turns=10,
        output_format="json",
    ):
        if isinstance(msg, harness.Result):
            result = msg.text

    return json.loads(result) if result else {}

4. `agents/pm.py`

"""将技术 blueprint 转换为用户视角需求 + 渐进提示"""

import harness
import json

SYSTEM_PROMPT = """你是一个优秀的产品经理。你的任务是将技术性的 feature 描述
改写为用户/开发者视角的需求文档，并为可能卡住的开发者准备渐进式提示。

需求描述规则：
- 描述期望行为，不暴露实现细节
- 包含验收标准（用户可感知的行为变化）
- 可以包含背景信息（"用户反馈..."、"性能问题..."）
- 不给文件路径、不给代码片段

提示层级：
- L1（方向级）：涉及哪个模块、用什么思路
- L2（模块级）：具体哪些文件、核心问题是什么
- L3（代码级）：具体哪一行、参考哪个 PR"""


async def generate_task(blueprint_feature: dict, repo: str) -> dict:
    prompt = f"""将以下技术 feature 转换为产品需求 + 渐进提示。

仓库: {repo}
Feature:
{json.dumps(blueprint_feature, ensure_ascii=False, indent=2)}

输出 JSON：
{{
  "task_id": "...",
  "description": "用户视角的需求描述（100-300字）",
  "acceptance_criteria": ["验收标准1", "验收标准2", ...],
  "test_cmd": "测试命令",
  "hints": {{
    "L1": ["方向提示1", "方向提示2"],
    "L2": ["模块提示1", "模块提示2"],
    "L3": ["代码级提示1"]
  }},
  "ground_truth_tag": "..."
}}"""

    result = None
    async for msg in harness.run(
        prompt=prompt,
        system_prompt=SYSTEM_PROMPT,
        provider="anthropic",
        model="claude-opus-4-20260514",
        max_turns=5,
        output_format="json",
    ):
        if isinstance(msg, harness.Result):
            result = msg.text

    return json.loads(result) if result else {}

5. `agents/reviewer.py`

"""审核 Dev 轨迹，判断训练价值，输出最终 task 卡片"""

import harness
import json

SYSTEM_PROMPT = """你是一个 AI 训练数据质量审核专家。你的任务是审核编码 Agent
的执行轨迹，判断其作为 RL 训练数据的价值。

判断维度：
1. 难度真实性（最重要）：Dev 是否真的卡住了？需要什么级别的提示？
2. 过程质量：探索是否有逻辑（不是随机瞎试）
3. 提示有效性：收到提示后是否有效调整方向
4. 任务描述清晰度：PM 的描述是否让人类也能理解

verdict:
- "keep_high": 难度 4-5，有完整探索-提示-突破链路
- "keep_medium": 难度 3，有有意义的迭代
- "keep_low": 难度 1-2，可作为 curriculum 起点
- "reject_env": 环境问题导致的失败
- "reject_trivial": 太简单，无训练价值
- "reject_noise": Dev 在简单问题上打转"""


async def review_task(task: dict, trajectory_file: str, summary: dict) -> dict:
    trajectory_preview = open(trajectory_file).read()[:5000]

    prompt = f"""审核以下任务的执行结果。

## 任务描述
{task['description']}

## 执行摘要
{json.dumps(summary, ensure_ascii=False, indent=2)}

## 轨迹（截取）
{trajectory_preview}

输出 JSON：
{{
  "verdict": "keep_high|keep_medium|keep_low|reject_*",
  "difficulty": 1-5,
  "reasoning": "判断理由",
  "final_task_card": {{
    "task_id": "...",
    "repo": "...",
    "base_commit": "...",
    "description": "...",
    "acceptance_criteria": [...],
    "test_cmd": "...",
    "hints": {{...}},
    "ground_truth_tag": "...",
    "difficulty": N,
    "rl_value": "high|medium|low"
  }}
}}"""

    result = None
    async for msg in harness.run(
        prompt=prompt,
        system_prompt=SYSTEM_PROMPT,
        provider="anthropic",
        model="claude-opus-4-20260514",
        max_turns=3,
        output_format="json",
    ):
        if isinstance(msg, harness.Result):
            result = msg.text

    return json.loads(result) if result else {}

主入口 `forge.py`

#!/usr/bin/env python3
"""HarnessForge — 多 Agent 代码任务工厂"""

import asyncio
import argparse
import json
import os
from pathlib import Path

from agents.task_designer import design_tasks
from agents.pm import generate_task
from agents.reviewer import review_task
from orchestrator import run_dev_task
from sandbox import RepoSandbox


async def forge_repo(repo: str, base_tag: str, next_tags: list[str]):
    """对一个 repo 执行完整的任务生成流程"""
    output = Path("output")
    slug = repo.replace("/", "_")

    print(f"\n{'='*60}")
    print(f"[Forge] {repo} @ {base_tag}")
    print(f"{'='*60}")

    # ── Step 1: Task Designer ──
    blueprint_file = output / "blueprints" / f"{slug}.json"
    if not blueprint_file.exists():
        print("[Step 1] Task Designer: 分析 repo，设计 feature 任务...")
        blueprint = await design_tasks(repo, base_tag, next_tags)
        blueprint_file.parent.mkdir(parents=True, exist_ok=True)
        blueprint_file.write_text(json.dumps(blueprint, ensure_ascii=False, indent=2))
    else:
        blueprint = json.loads(blueprint_file.read_text())

    features = blueprint.get("features", [])
    print(f"[Step 1] 设计了 {len(features)} 个 feature 任务")

    # ── Step 2: PM 生成需求 ──
    tasks = []
    for i, feature in enumerate(features):
        task_file = output / "tasks" / f"{slug}_f{i}.json"
        if not task_file.exists():
            print(f"[Step 2] PM: 生成任务 {i+1}/{len(features)}...")
            task = await generate_task(feature, repo)
            task["repo"] = repo
            task["dev_model"] = "claude-sonnet-4-20250514"
            task_file.parent.mkdir(parents=True, exist_ok=True)
            task_file.write_text(json.dumps(task, ensure_ascii=False, indent=2))
        else:
            task = json.loads(task_file.read_text())
        tasks.append(task)

    # ── Step 3: 构建 Docker 沙箱 ──
    sandbox = RepoSandbox(repo=repo, base_tag=base_tag, container_name=f"hf-{slug}")
    print("[Step 3] 构建 Docker 沙箱...")
    await sandbox.setup()

    # ── Step 4: Dev 执行（带 PM 提示） ──
    summaries = []
    for task in tasks:
        traj_file = output / "trajectories" / f"{task['task_id']}.jsonl"
        if not traj_file.exists():
            print(f"[Step 4] Dev: 执行 {task['task_id']}...")
            summary = await run_dev_task(task, sandbox)
            summaries.append(summary)
        else:
            summary_file = output / "trajectories" / f"{task['task_id']}_summary.json"
            summary = json.loads(summary_file.read_text()) if summary_file.exists() else {}
            summaries.append(summary)

    # ── Step 5: Reviewer 审核 ──
    kept = 0
    for task, summary in zip(tasks, summaries):
        verdict_file = output / "verified" / f"{task['task_id']}.json"
        if not verdict_file.exists():
            traj_file = str(output / "trajectories" / f"{task['task_id']}.jsonl")
            print(f"[Step 5] Reviewer: 审核 {task['task_id']}...")
            verdict = await review_task(task, traj_file, summary)
            verdict_file.parent.mkdir(parents=True, exist_ok=True)
            verdict_file.write_text(json.dumps(verdict, ensure_ascii=False, indent=2))
        else:
            verdict = json.loads(verdict_file.read_text())

        if verdict.get("verdict", "").startswith("keep"):
            kept += 1

    print(f"\n[Done] {repo}: {len(tasks)} 任务 → {kept} 个保留")
    return kept


async def main():
    parser = argparse.ArgumentParser(description="HarnessForge — 多 Agent 代码任务工厂")
    parser.add_argument("--repo", help="单个 repo (e.g. django/django)")
    parser.add_argument("--base-tag", help="基准版本 (e.g. v5.0)")
    parser.add_argument("--next-tags", help="后续版本，逗号分隔 (e.g. v5.1,v5.2)")
    parser.add_argument("--config", default="config.yaml", help="配置文件")
    args = parser.parse_args()

    if args.repo:
        next_tags = args.next_tags.split(",") if args.next_tags else []
        await forge_repo(args.repo, args.base_tag, next_tags)
    else:
        # 从 config.yaml 读取 repo 列表，批量执行
        import yaml
        config = yaml.safe_load(open(args.config))
        total_kept = 0
        for repo_cfg in config["repos"]:
            kept = await forge_repo(
                repo_cfg["name"],
                repo_cfg["base_tag"],
                repo_cfg["next_tags"],
            )
            total_kept += kept
        print(f"\n{'='*60}")
        print(f"[Total] {total_kept} 个任务通过审核")


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

使用方式

单个 repo 快速验证

cd tools/harnessforge

# 单个 repo
python forge.py \
  --repo django/django \
  --base-tag v5.0 \
  --next-tags v5.1,v5.2

# 查看产出
ls output/verified/
cat output/verified/django_django_f0.json | jq '.verdict, .difficulty'

批量执行

# config.yaml
repos:
  - name: django/django
    base_tag: v5.0
    next_tags: [v5.1, v5.2]

  - name: vercel/next.js
    base_tag: v14.0.0
    next_tags: [v14.1.0, v14.2.0]

  - name: gin-gonic/gin
    base_tag: v1.9.0
    next_tags: [v1.10.0]

python forge.py --config config.yaml

分阶段交付

Phase 1: 最小可用（3-5 天）

天	交付
Day 1	项目骨架 + `sandbox.py` + Docker 镜像构建
Day 2	`task_designer.py` + `pm.py` —— 能产出 task.json
Day 3	`orchestrator.py` —— Dev 执行 + PM 提示循环
Day 4	`reviewer.py` + `forge.py` —— 端到端跑通
Day 5	用 Django 跑 3 个 task，验证产出质量

Phase 1 成功标准：3 个 Django feature task，至少 1 个 difficulty ≥ 3 的 verified task。

Phase 2: 完善 + 多 repo（1 周）

支持 TypeScript 和 Go repo
轨迹格式对齐 TAO 标准
DPO 对自动生成
多模型 Dev 执行（Claude + 本地 vLLM）

Phase 3: 规模化（2 周）

30 repo 全量执行
RL 训练验证
消融实验

风险和 fallback

风险	概率	fallback
OpenHarness `harness.run()` API 不稳定	中	退回 Claude Code CLI 版本
Docker 沙箱构建失败（依赖问题）	高	提供预构建镜像 + `setup.py` 兼容处理
Dev Agent 在沙箱中工具不够用	低	OpenHarness 43+ 工具基本够用
PM 提示释放时机不准	中	参数可调（`COOLDOWN`/`L1_TRIGGER_FAILS`/`L2_TRIGGER_TURNS`）
API 费用超预算	低	Task Designer/PM 用 Sonnet，省钱

与其他版本的关系

文档	定位
HarnessForge v1	原始完整 proposal（理论和方法论）
HarnessForge v2 规格书	6 Agent 协议设计
HarnessForge v2 CLI 版	Claude Code CLI 实现（轻量 fallback）
本文（开发文档）	OpenHarness 实现的具体代码和开发计划