Harness实战：Worktree Isolation——让每个任务各干各的

Agent A: 改 config.py (删30行，重构)
Agent B: 改 config.py (在第25行加15行校验)
→ 最终结果: B的版本覆盖A，A的重构完全丢失

Lead: task_create("重构 config.py")
Lead: task_create("加校验逻辑")
Lead: worktree_create(task_id=1)     ← 创建 .worktrees/task-1/
Lead: worktree_create(task_id=2)     ← 创建 .worktrees/task-2/

Agent A: worktree_run(1, "vim config.py")   ← 在 task-1 目录操作
Agent B: worktree_run(2, "vim config.py")   ← 在 task-2 目录操作
→ 文件系统级隔离，不可能互相污染

原有 harness-agent-teams.py (保留全部)
├── BASE_TOOLS (4): bash, read_file, write_file, edit_file
├── TEAM_COMM_TOOLS (2): send_message, read_inbox
├── MATE_TOOLS (2): shutdown_response, plan_approval
├── LEAD_TOOLS (13): compact, task, parallel_tasks, background_task, check_task,
│                    spawn_teammate, list_teammates, broadcast,
│                    shutdown_request, shutdown_status, plan_review
└── Plan Mode + Context Compact

+ 新增 (塞入 LEAD_TOOLS)
├── 任务看板 (2): task_create, task_list
└── Worktree 隔离 (6): worktree_create, worktree_remove, worktree_run,
                        worktree_list, worktree_keep, worktree_events

TEAMMATE_TOOLS = BASE_TOOLS + TEAM_COMM_TOOLS + MATE_TOOLS  # 8
TOOLS = BASE_TOOLS + TEAM_COMM_TOOLS + LEAD_TOOLS            # 25

项目根目录/
├── .tasks/                    ← 控制面：调度和状态
│   ├── tasks.json             ← 任务列表 + 状态 + worktree 绑定
│   └── events.jsonl           ← 生命周期事件流
│
├── .worktrees/                ← 执行面：独立工作目录
│   ├── index.json             ← worktree 注册表
│   ├── task-1/                ← 任务1的完整项目副本
│   └── task-2/                ← 任务2的完整项目副本
│
├── config.py                  ← 主目录不受影响
└── ...

class TaskBoard:
    """任务看板: .tasks/tasks.json 管理任务全生命周期。

    Task FSM:
      [pending] → [in_progress] → [completed]
                        │
                        └→ [failed]
    """

    def __init__(self, tasks_dir: Path):
        self.dir = tasks_dir
        self.dir.mkdir(exist_ok=True)
        self.tasks_path = self.dir / "tasks.json"
        self.tasks: list[dict] = self._load()

    def create(self, subject: str, description: str = "",
               depends_on: list = None) -> str:
        next_id = max((t["id"] for t in self.tasks), default=0) + 1
        task = {
            "id": next_id,
            "subject": subject,
            "description": description,
            "status": "pending",
            "owner": None,
            "depends_on": depends_on or [],
            "worktree": None,            # ← 新增: worktree 绑定字段
        }
        self.tasks.append(task)
        self._save()
        return json.dumps(task, indent=2, ensure_ascii=False)

def update_status(self, task_id: int | str, status: str) -> str:
    """WorktreeManager 调用：推进任务状态。"""
    task = self.find(int(task_id))
    if not task:
        return f"Error: Task {task_id} not found"
    task["status"] = status
    self._save()
    return f"Task #{task_id} status → {status}"

def bind_worktree(self, task_id: int, worktree_name: str):
    """WorktreeManager 调用：绑定 worktree 名称。"""
    task = self.find(task_id)
    if task:
        task["worktree"] = worktree_name
        self._save()

def list_all(self) -> str:
    status_icon = {
        "pending": " ", "in_progress": ">",
        "completed": "x", "failed": "!",
    }
    lines = ["Task Board:"]
    for t in self.tasks:
        icon = status_icon.get(t["status"], "?")
        owner = f" @{t['owner']}" if t.get("owner") else ""
        wt = f" wt={t['worktree']}" if t.get("worktree") else ""
        deps = f" (depends on: {t['depends_on']})" if t.get("depends_on") else ""
        lines.append(f"  [{icon}] #{t['id']}: {t['subject']}{owner}{wt}{deps}")
    return "\n".join(lines)

Task Board:
  [>] #1: 重构 config.py wt=task-1
  [>] #2: 加校验逻辑 wt=task-2
  [ ] #3: 写集成测试 (depends on: [1, 2])

Task FSM:                          Worktree FSM:
  [pending] → [in_progress] → [completed]    [absent] → [active] → [removed]
                    │                                       │
                    └→ [failed]                             └→ [kept]

def _register(self, task_id: str, branch: str, path: str) -> None:
    index = self._load_index()
    key = f"task-{task_id}"
    index[key] = {
        "task_id": task_id,
        "branch": branch,
        "path": path,
        "status": "active",
        "created_at": datetime.now(timezone.utc).isoformat(),
    }
    self._save_index(index)

def _emit(self, event_type: str, **data) -> None:
    event = {
        "ts": datetime.now(timezone.utc).isoformat(),
        "type": event_type,
        **data,
    }
    with open(self._events_path(), "a") as f:
        f.write(json.dumps(event, ensure_ascii=False) + "\n")

def worktree_create(self, task_id: str, branch: str | None = None) -> str:
    """创建 worktree 并绑定到任务，自动推进 Task pending→in_progress。"""
    task = self.task_board.find(int(task_id))
    if not task:
        return f"Error: Task #{task_id} not found"
    if task["status"] != "pending":
        return f"Error: Task #{task_id} status is '{task['status']}', expected 'pending'"

    branch = branch or f"task-{task_id}"
    wt_path = self.worktrees_dir / f"task-{task_id}"

    self._emit("worktree.create.before", task_id=task_id, branch=branch)

    try:
        # 1. 创建 git worktree
        subprocess.run(
            ["git", "worktree", "add", str(wt_path), "-b", branch],
            cwd=self.project_root, check=True,
            capture_output=True, text=True,
        )
        # 2. 注册到 index.json
        self._register(task_id, branch, str(wt_path))
        # 3. 推进任务状态: pending → in_progress
        self.task_board.update_status(task_id, "in_progress")
        # 4. 绑定 worktree 名称到任务
        self.task_board.bind_worktree(int(task_id), f"task-{task_id}")
        self._emit("worktree.create.after", task_id=task_id, path=str(wt_path))

        return json.dumps({
            "task_id": task_id,
            "branch": branch,
            "path": str(wt_path),
            "status": "active",
        }, indent=2)

    except subprocess.CalledProcessError as e:
        self._emit("worktree.create.failed", task_id=task_id, error=e.stderr.strip())
        return f"Error: {e.stderr.strip()}"

def run_in_worktree(self, task_id: str, command: str) -> str:
    """在任务的 worktree 目录中执行命令。"""
    wt_path = self._get_worktree_path(task_id)
    if not wt_path or not wt_path.exists():
        return f"Error: No active worktree for task #{task_id}"

    try:
        r = subprocess.run(
            command, shell=True, cwd=wt_path,      # ← cwd 切到 worktree 路径
            capture_output=True, text=True, timeout=120,
        )
        out = (r.stdout + r.stderr).strip()
        return out[:50000] if out else "(no output)"
    except subprocess.TimeoutExpired:
        return "Error: Timeout (120s)"

def worktree_remove(self, task_id: str, complete_task: bool = True) -> str:
    """拆除 worktree。complete_task=True 同时标记任务完成。"""
    wt_path = self._get_worktree_path(task_id)
    if not wt_path:
        return f"Error: No worktree for task #{task_id}"

    self._emit("worktree.remove.before", task_id=task_id, complete_task=complete_task)

    # 1. 完成任务（可选）
    if complete_task:
        self.task_board.update_status(task_id, "completed")
        self._emit("task.completed", task_id=task_id)

    # 2. 移除 git worktree
    if wt_path.exists():
        subprocess.run(
            ["git", "worktree", "remove", str(wt_path), "--force"],
            cwd=self.project_root, capture_output=True, text=True,
        )

    # 3. 清理分支
    subprocess.run(
        ["git", "branch", "-d", f"task-{task_id}"],
        cwd=self.project_root, capture_output=True, text=True,
    )

    # 4. 从注册表注销
    self._unregister(task_id)
    self._emit("worktree.remove.after", task_id=task_id)
    return f"Removed worktree for task #{task_id}"

def worktree_keep(self, task_id: str) -> str:
    """标记 worktree 为 kept 状态，保留不拆，供调试。"""
    index = self._load_index()
    key = f"task-{task_id}"
    if key not in index:
        return f"Error: No worktree for task #{task_id}"
    index[key]["status"] = "kept"
    self._save_index(index)
    self._emit("worktree.keep", task_id=task_id)
    return json.dumps({"task_id": task_id, "status": "kept"})

def _find_incomplete_ops(self) -> list:
    """扫描事件流，找未闭合的 before 事件。"""
    before_events = {}
    for line in events_path.read_text().strip().split("\n"):
        event = json.loads(line)
        etype = event["type"]
        tid = event.get("task_id", "")

        if etype.endswith(".before"):
            before_events[f"{etype}:{tid}"] = event
        elif etype.endswith(".after") or etype.endswith(".failed"):
            base = etype.rsplit(".", 1)[0]
            before_events.pop(f"{base}.before:{tid}", None)

    return list(before_events.values())

def recover(self) -> str:
    """从 events + index.json + 磁盘交叉比对，恢复一致状态。"""
    issues = {"orphaned_worktrees": [], "incomplete_ops": [], "recovered": []}

    # 1. 扫描事件流，找未闭合的 before 事件
    pending_ops = self._find_incomplete_ops()
    for op in pending_ops:
        if op["type"] == "worktree.create.before":
            # 创建到一半崩溃了 → 清理残留目录
            partial_path = self.worktrees_dir / f"task-{op['task_id']}"
            if partial_path.exists():
                subprocess.run(
                    ["git", "worktree", "remove", str(partial_path), "--force"], ...
                )
            self._unregister(op["task_id"])

    # 2. 对比 index.json 和磁盘实际状态
    for wt_id, wt_info in self._load_index().items():
        if not Path(wt_info["path"]).exists():
            # 注册了但目录不存在 → 孤儿记录
            self._unregister(wt_info["task_id"])
            issues["orphaned_worktrees"].append(wt_id)

    # 3. 对比磁盘和 index.json
    for dir_path in self.worktrees_dir.iterdir():
        if dir_path.is_dir() and dir_path.name not in index:
            # 目录存在但没注册 → 孤儿目录
            issues["orphaned_worktrees"].append(dir_path.name)

    return json.dumps(issues, indent=2, ensure_ascii=False)

LEAD_TOOLS = [
    # ... 原有 13 个工具 ...

    # 任务看板 (+2)
    {"name": "task_create", "description": "Create a task on the board.",
     "input_schema": {"type": "object", "properties": {
         "subject": {"type": "string"},
         "description": {"type": "string"},
         "depends_on": {"type": "array", "items": {"type": "integer"}}
     }, "required": ["subject"]}},
    {"name": "task_list", "description": "List all tasks with status.",
     "input_schema": {"type": "object", "properties": {}}},

    # worktree 隔离 (+6)
    {"name": "worktree_create", ...},   # 创建并绑定
    {"name": "worktree_remove", ...},   # 拆除并完成
    {"name": "worktree_run", ...},      # 在隔离目录执行
    {"name": "worktree_list", ...},     # 列出所有 worktree
    {"name": "worktree_keep", ...},     # 标记保留
    {"name": "worktree_events", ...},   # 查看事件日志
]

TOOL_HANDLERS = {
    **WORKSPACE_HANDLERS,
    # ... 原有 handlers ...

    # task board
    "task_create":      lambda **kw: BOARD.create(kw["subject"],
                            kw.get("description", ""), kw.get("depends_on")),
    "task_list":        lambda **kw: BOARD.list_all(),

    # worktree isolation
    "worktree_create":  lambda **kw: WT.worktree_create(
                            kw["task_id"], kw.get("branch")),
    "worktree_remove":  lambda **kw: WT.worktree_remove(
                            kw["task_id"], kw.get("complete_task", True)),
    "worktree_run":     lambda **kw: WT.run_in_worktree(
                            kw["task_id"], kw["command"]),
    "worktree_list":    lambda **kw: WT.worktree_list(),
    "worktree_keep":    lambda **kw: WT.worktree_keep(kw["task_id"]),
    "worktree_events":  lambda **kw: WT.worktree_events(kw.get("limit", 30)),
}

elif query.strip() == "/tasks":
    print(BOARD.list_all())
    continue
elif query.strip() == "/worktrees":
    print(WT.worktree_list())
    continue
elif query.strip() == "/events":
    print(WT.worktree_events())
    continue
elif query.strip() == "/recover":
    print(WT.recover())
    continue

SYSTEM = f"""You are a team lead agent at {WORKDIR}. Use tools to solve tasks.

...（原有内容不变）

Task board:
- `task_create`: Create a task on the board with optional dependencies.
- `task_list`: List all tasks with status.

Worktree isolation:
- `worktree_create`: Create an isolated git worktree bound to a task.
- `worktree_remove`: Remove a worktree, optionally completing the bound task.
- `worktree_run`: Run a command inside a worktree's directory.
- `worktree_list`: List all worktrees with status.
- `worktree_keep`: Mark a worktree as kept (for debugging).
- `worktree_events`: Show worktree lifecycle event log."""

"Use `worktree_create` to isolate file changes per task. "