Agent Harness 工程：Worktree Isolation 与任务工作区

Backend Agent 正在修改搜索接口。

Frontend Agent 也在工作，它更新了双方共用的类型文件。几乎同一时刻，Backend 运行格式化工具，顺手重写了那个文件。

两边的 Task 都合法，两份租约也都有效，可后一次写入悄无声息地盖住了前一次修改。

更糟的是，QA 随后跑出的测试结果，已经没人说得清究竟验证了哪一版代码。

上一篇解决了团队里的请求、审批和退出；这一章处理更朴素的问题：多个 Agent 的手，不能同时伸进同一个工作台。

Task Store 能防止两个人认领同一个 Task，却不能阻止两个不同 Task 修改同一目录。

所以，每个 Task 都需要一份自己的工作现场。

一个仓库，可以有很多扇独立的门

Git 仓库可以拥有一个主 worktree 和多个 linked worktree。它们共享 Git 对象数据库与大部分 refs，却各自拥有独立的：

工作目录；
HEAD；
index；
tracked 和 untracked 文件。

Agent Team 因此可以这样展开：

repository
├── integration worktree
├── task-api worktree
├── task-ui worktree
└── task-test worktree

每个 Task 在独立目录中修改和测试，最后只把经过验证的 Commit 交给 Integration Worker。

Backend 在 task-api 中编辑，不会改变 Frontend 的 task-ui。未提交文件、构建目录和 index 都有了明确归属。

Git worktree 官方文档还提供了适合自动化的能力：

同一分支默认不能同时被两个 worktree checkout；
worktree list --porcelain -z 可以稳定解析；
活跃 worktree 可以锁定，避免被 prune；
正常清理使用 git worktree remove。

这些机制让 Git 不只保存结果，也参与管理任务现场。

Worktree 隔离协作，不负责安全

Worktree 能阻止文件互相覆盖，却不是完整沙箱。

它可以隔离：

工作目录
index 与 HEAD
未提交文件
目录内依赖与构建产物

它不能自动隔离：

网络和外部 API
数据库与云资源
固定端口
全局临时目录
用户缓存和凭据
共享 Git refs 与对象数据库

仅仅把 Shell 的 cwd 改到 worktree 还不够。命令仍然可以 cd ..，访问其它目录，甚至通过 Git 修改共享分支。

Harness 还要提供两层边界：

进程沙箱：只把当前 worktree 挂载为可写，限制网络、目录与系统能力；
Git Broker：创建分支、提交、集成和清理都走受控工具。

Worktree 解决“不互相干扰”，Sandbox 解决“不能越界”。两者缺一不可。

Worktree 本身也要有持久化状态

工作目录不是临时字符串，而是 Task 的运行资源：

from dataclasses import asdict, dataclass, field
from typing import Literal


WorktreeStatus = Literal[
    "allocating",
    "active",
    "ready_for_integration",
    "integrating",
    "conflicted",
    "integrated",
    "removing",
    "retained",
    "removed",
    "failed",
]


@dataclass
class WorktreeRecord:
    id: str
    task_id: str
    owner_agent_id: str
    owner_claim_token: str
    repo_root: str
    path: str
    branch: str
    base_commit: str
    status: WorktreeStatus

    result_commit: str | None = None
    integrated_commit: str | None = None
    conflict_paths: list[str] = field(
        default_factory=list
    )
    cleanup_error: str | None = None

    created_at: str = ""
    updated_at: str = ""
    removed_at: str | None = None

    def to_dict(self) -> dict:
        return asdict(self)

最小生命周期是：

allocating
-> active
-> ready_for_integration
-> integrating
-> integrated
-> removing
-> removed

发生冲突时：

integrating -> conflicted -> active

retained 表示 Harness 刻意保留现场，例如存在未提交文件、清理条件不满足或需要人工检查。它不是一个“删除失败但假装没看见”的状态。

Task 保存 workspace_id 以找到 Worktree Record，但真正的执行所有权仍然来自 claimed_by + claim_token。Record 中的 owner_claim_token 只用于恢复时核对这一轮所有权，对模型和普通日志都要隐藏。知道目录路径，不等于有权写入。

路径、分支和基线都由 Harness 决定

模型不能提交任意 worktree 路径和分支名。

import re
from pathlib import Path


TASK_ID_PATTERN = re.compile(
    r"^[a-z0-9][a-z0-9-]{0,63}$"
)


def task_branch_name(task_id: str) -> str:
    if not TASK_ID_PATTERN.fullmatch(task_id):
        raise ValueError("invalid task id")
    return f"agent/task/{task_id}"


def task_worktree_path(
    worktree_root: Path,
    task_id: str,
) -> Path:
    if not TASK_ID_PATTERN.fullmatch(task_id):
        raise ValueError("invalid task id")

    root = worktree_root.resolve()
    candidate = (root / task_id).resolve()
    if root not in candidate.parents:
        raise ValueError("path escapes worktree root")
    return candidate


def ensure_roots_are_disjoint(
    repo_root: Path,
    worktree_root: Path,
) -> None:
    repo = repo_root.resolve()
    runtime = worktree_root.resolve()
    if (
        runtime == repo
        or repo in runtime.parents
        or runtime in repo.parents
    ):
        raise ValueError(
            "repository and runtime roots overlap"
        )

实际 worktree 不要放进主仓库的 .agent/worktrees/。嵌套目录会污染 git status，也容易让清理边界变得含糊。

更清楚的布局是：

project/.agent/worktrees/records/
  保存 WorktreeRecord

.project-agent-runtime/worktrees/
  保存真实工作目录

Runtime Root 来自 Harness 配置和 allowlist，不来自模型。

Task 分配工作区时，还要先把 integration branch 解析成不可变 Commit OID。不要只记录一个会移动的 main：

base_commit = 48f13f...

恢复时，Harness 才能确认这个 worktree 当初究竟从哪一版代码出发。

Git 命令也要经过受控边界

不要拼接 Shell 字符串运行 Git。使用参数数组、固定 cwd、无交互环境和超时：

import os
import subprocess


class GitCommandError(RuntimeError):
    pass


def run_git(
    cwd: Path,
    *args: str,
    check: bool = True,
    timeout_seconds: int = 120,
) -> subprocess.CompletedProcess[bytes]:
    env = {
        "PATH": os.environ.get("PATH", ""),
        "HOME": os.environ.get("HOME", ""),
        "LANG": "C",
        "LC_ALL": "C",
        "GIT_CONFIG_NOSYSTEM": "1",
        "GIT_CONFIG_GLOBAL": os.devnull,
        "GIT_TERMINAL_PROMPT": "0",
    }
    result = subprocess.run(
        ["git", "-C", str(cwd), *args],
        stdin=subprocess.DEVNULL,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        env=env,
        timeout=timeout_seconds,
        check=False,
    )
    if check and result.returncode != 0:
        detail = result.stderr.decode(
            "utf-8",
            errors="replace",
        )[-2000:]
        raise GitCommandError(detail)
    return result


def resolve_commit(
    repo_root: Path,
    revision: str,
) -> str:
    result = run_git(
        repo_root,
        "rev-parse",
        "--verify",
        f"{revision}^{{commit}}",
    )
    return result.stdout.decode("ascii").strip()

不要把完整进程环境传给 Git。凭据、代理和自定义 GIT_* 变量都可能越过 Task 边界；确实需要的变量由 Harness 显式加入。

创建 Worktree 前，先保存“我要做什么”

创建工作区遵循和前几章相同的恢复原则：先写意图，再执行副作用。

def allocate_task_worktree(
    state: HarnessState,
    task: Task,
    agent_id: str,
    claim_token: str,
) -> WorktreeRecord:
    state.task_store.ensure_owner(
        task.id,
        agent_id,
        claim_token,
    )

    with state.repo_operation_lock:
        base_commit = resolve_commit(
            state.repo_root,
            state.integration_branch,
        )
        branch = task_branch_name(task.id)
        run_git(
            state.repo_root,
            "check-ref-format",
            "--branch",
            branch,
        )

        record = WorktreeRecord(
            id=f"worktree-{task.id}",
            task_id=task.id,
            owner_agent_id=agent_id,
            owner_claim_token=claim_token,
            repo_root=str(state.repo_root.resolve()),
            path=str(
                task_worktree_path(
                    state.worktree_root,
                    task.id,
                )
            ),
            branch=branch,
            base_commit=base_commit,
            status="allocating",
            created_at=utc_now_text(),
            updated_at=utc_now_text(),
        )
        stored = state.worktree_store.put_if_absent(
            record
        )
        return reconcile_worktree_allocation(
            state,
            stored,
        )

如果 git worktree add 已经成功、Record 还没更新时进程退出，重启后的 reconciliation 会读取 Git 的真实现场，再补写 active。

真实现场应通过：

git worktree list --porcelain -z

解析。-z 使用 NUL 分隔，即使路径包含空格或换行也不会被错误切开。

恢复分配时有三种主要情况：

path 已注册，branch 与 Record 一致：验证后进入 active；
branch 已存在，path 尚未注册：确认 branch 仍指向 base commit，再重新 worktree add；
path 已存在却不属于 Git worktree：保留并报告，不能自动删除。

创建时不使用 --force 或 -B。它们会绕过 Git 的保护，甚至重置已经存在的 Task 分支。

每个工具都自动落在当前 Task 的工作区

模型不需要在每次 read_file、edit_file 和 bash 时传 workspace_id。Harness 根据当前 Task Session 注入：

from dataclasses import dataclass


@dataclass(frozen=True)
class TaskWorkspace:
    task_id: str
    root: Path
    temp_dir: Path
    env: dict[str, str]


def resolve_workspace_path(
    workspace: TaskWorkspace,
    relative_path: str,
    must_exist: bool = False,
) -> Path:
    candidate = Path(relative_path)
    if candidate.is_absolute():
        raise PermissionError(
            "absolute paths are not allowed"
        )

    root = workspace.root.resolve(strict=True)
    resolved = (root / candidate).resolve(
        strict=must_exist
    )
    if (
        resolved != root
        and root not in resolved.parents
    ):
        raise PermissionError(
            "path escapes task worktree"
        )
    return resolved

对已有 symlink，resolve() 能阻止明显的链接越界；创建新文件时，还要检查最近的已存在父目录。

Shell 至少绑定：

cwd     = Task worktree
TMPDIR  = Task 专属临时目录
PORT    = Task 分配到的端口
TASK_ID = 当前 Task ID

但 cwd 仍然不是 OS 安全边界。Shell 要运行在沙箱或容器中，只挂载允许目录。

端口、测试数据库、浏览器 Profile 和覆盖率文件也要按 Task 隔离。两个工作目录不同，不代表它们不会同时争抢 localhost:3000。

可交接结果不是一堆文件，而是一份经过验证的 Commit

Agent 说“文件改好了”时，Worktree 里可能还有未跟踪文件、脏 index、后台进程和未经验证的改动。

Harness 应把结果固定为 Git tree，再验证这个 tree，最后创建 Commit：

def prepare_task_candidate(
    state: HarnessState,
    task: Task,
    record: WorktreeRecord,
) -> str:
    worktree = Path(record.path)
    if state.job_manager.has_running_jobs(
        record.id
    ):
        raise ValueError(
            "workspace still has running jobs"
        )

    paths = changed_paths(worktree)
    if not paths:
        raise ValueError("task has no changes")

    ensure_task_paths_allowed(task, paths)
    scan_for_secrets(worktree, paths)
    run_git(worktree, "diff", "--check")
    run_git(worktree, "add", "--all", "--", ".")
    run_git(
        worktree,
        "diff",
        "--cached",
        "--check",
    )

    candidate_tree = run_git(
        worktree,
        "write-tree",
    ).stdout.decode("ascii").strip()
    head_tree = run_git(
        worktree,
        "show",
        "-s",
        "--format=%T",
        "HEAD",
    ).stdout.decode("ascii").strip()
    if candidate_tree == head_tree:
        raise ValueError("candidate has no changes")
    return candidate_tree

接下来暂停这个工作区的写工具，让验证 Job 只读取候选内容。每条验证记录同时保存：

job_id
candidate_tree
command
exit_status

测试通过后，提交前再次确认 index、未暂存文件和验证结果仍属于同一棵 tree：

def create_task_commit(
    state: HarnessState,
    task: Task,
    record: WorktreeRecord,
    candidate_tree: str,
    verification_ids: list[str],
) -> str:
    worktree = Path(record.path)
    current_tree = run_git(
        worktree,
        "write-tree",
    ).stdout.decode("ascii").strip()
    if current_tree != candidate_tree:
        raise ValueError(
            "candidate changed after verification"
        )

    unstaged = run_git(
        worktree,
        "diff",
        "--quiet",
        check=False,
    )
    if unstaged.returncode not in {0, 1}:
        raise GitCommandError(
            "cannot inspect unstaged changes"
        )
    untracked = run_git(
        worktree,
        "ls-files",
        "--others",
        "--exclude-standard",
        "-z",
    ).stdout
    if unstaged.returncode != 0 or untracked:
        raise ValueError(
            "workspace changed after verification"
        )

    ensure_verification_passed(
        state,
        verification_ids,
        candidate_tree,
    )
    run_git(
        worktree,
        "-c",
        "user.name=Agent Harness",
        "-c",
        "user.email=agent@local.invalid",
        "-c",
        "core.hooksPath=/dev/null",
        "commit",
        "-m",
        f"agent({task.id}): {task.title}",
    )

    result_commit = resolve_commit(
        worktree,
        "HEAD",
    )
    committed_tree = run_git(
        worktree,
        "show",
        "-s",
        "--format=%T",
        result_commit,
    ).stdout.decode("ascii").strip()
    if committed_tree != candidate_tree:
        raise ValueError(
            "commit does not match verified candidate"
        )
    if run_git(
        worktree,
        "status",
        "--porcelain",
        "-z",
    ).stdout:
        raise ValueError(
            "workspace is dirty after commit"
        )
    return result_commit

验证不能相信模型写的一句“tests passed”。ensure_verification_passed() 只接受 Harness 生成的 Job ID，并从 Job Store 读取真实退出状态与 candidate tree。

提交成功后，Worktree Record 写入 result_commit 并进入 ready_for_integration。Task 仍然保持 in_progress。

为什么还不能完成？

因为下游 Task 会从 integration branch 创建新 worktree。如果上游 Commit 尚未集成，下游即使解锁，也看不到它的结果。

Integration Worker 是唯一可以推开主门的人

普通成员只修改自己的 task branch。唯一的 Integration Worker 在独立 integration worktree 中串行合并。

Task Commit 只是候选结果。通过合并和集成测试后，Task 才能完成并解锁下游。

集成前必须确认：

integration worktree 干净；
result_commit 以原 base_commit 为祖先；
task branch 的 HEAD 仍等于 result_commit；
结果尚未包含在 integration branch 中。

然后在全局 integration_lock 中执行：

git merge --no-ff --no-commit <result_commit>

不立即提交，是为了先运行集成测试。

结果分成三类：

already_integrated
  Commit 已经是 integration HEAD 的祖先
  说明合并成功后 Record 更新曾中断，补写状态即可

ready_for_tests
  无冲突，MERGE_HEAD 存在
  运行集成测试，通过后创建 merge commit

conflicted
  保存冲突路径，git merge --abort
  把问题送回原 Task worktree

集成测试失败时也要 merge --abort，保存日志并让 Task 继续保持未完成。只有 merge commit 和集成验证全部成功，Harness 才写入 integrated_commit 并调用 task_complete。

这条顺序确保下游 Task 的 base commit 一定包含已验收的上游结果。

冲突应该回到独立现场解决

两个 Task 修改同一文件时，冲突不是 Worktree Manager 崩溃，而是正常协作结果。

Integration Worker 不应该占着共享 integration worktree 等模型慢慢修。它应该：

记录冲突文件和当前 integration commit；
执行 git merge --abort 恢复干净状态；
把 Record 标记为 conflicted；
向原成员或合适的接手者发送 blocker。

责任成员回到自己的 task worktree，合并最新 integration branch，解决冲突，重新运行验证，再产生新的 result_commit。

共享集成现场因此始终短暂、串行、可恢复，不会被一个冲突长期占住。

清理现场时，宁可保留，也不要强删

只有满足这些条件，Task worktree 才能回收：

结果已经集成；
没有后台 Job 仍以它为 cwd；
git status --porcelain -z 为空；
Artifact 已复制到持久化位置；
Worktree Record 与 Git 现场一致。

清理顺序是：

Record -> removing
git worktree unlock
git worktree remove <path>
git branch -d <task-branch>
Record -> removed

不要直接 rm -rf，也不要默认使用：

git worktree remove --force
git branch -D

普通删除失败，往往说明仍有未提交文件、submodule 或 Git 元数据不一致。Harness 应把 Record 标记为 retained，保存错误，等待检查。

removing 是持久化的清理意图。如果目录已经删除、Record 还没更新时进程退出，恢复器会继续清理分支并补写 removed。

git worktree prune 也不是正常回收流程。它适合清理已经丢失路径的 stale administrative metadata，应先用 prune --dry-run 生成审计结果。

重启后，要同时相信记录和现场

Harness 启动时，对照 Worktree Store 与 git worktree list --porcelain -z：

Record allocating，Git path 已存在
  校验 branch 与 base，补写 active

Record active，path 与 branch 一致
  恢复 Task Workspace 与关联 Job

Record 非 removing，Git path 消失
  标记 failed，保留 Task checkpoint

Git path 存在，却没有 Record
  隔离并通知 Lead，不自动删除

Record integrating，存在 MERGE_HEAD
  保存现场，abort 后重新排队

Record integrated，目录干净且没有 Job
  进入受控清理

Record removing，path 已消失
  完成分支清理并补写 removed

还要校验 active Record 对应的 Task 是否仍由同一 Agent 和 claim token 持有。

租约过期时，worktree 不会立刻删除。新的所有者先读取 checkpoint、Commit 和真实目录状态，再决定接手还是保留；确认接手后，再在 Store 锁内同时更新 owner_agent_id 与 owner_claim_token。

Worktree 是可恢复资源，Task Store 仍然是工作状态的最终事实来源。

模型只需要高层工作区工具

Agent 不需要直接看到 git worktree add、unlock、remove 或共享 refs 操作。

最小工具面可以是：

workspace_status
workspace_changed_files
workspace_prepare_commit
workspace_submit
workspace_integration_status

文件工具和 Bash 甚至不接收 workspace_id。Harness 从当前 Task Session 注入，避免模型把命令切换到另一个成员的目录。

当底层状态复杂时，给模型的操作面反而应该更小。

小结

这一章终于把团队的并行落实到文件系统：

每个 Task 使用独立 branch、固定 base commit 和 linked worktree；
Worktree Record 保存创建、集成、冲突与回收状态；
Git Broker 通过参数数组执行受控命令；
文件、Shell、临时目录、端口和测试资源绑定 Task；
候选 Git tree 与验证 Job 一一对应；
普通成员只交付 Commit，Integration Worker 串行合并；
Task 只在结果完成集成后进入 completed；
冲突回到独立 worktree 解决；
清理前检查 Job、Artifact 和 Git 现场，不默认强制删除；
重启后通过 Store 与真实 Git 状态 reconciliation。

现在，多个 Agent 可以在同一个仓库里并行工作，却不会直接踩进彼此尚未完成的现场。

不过，这些文件、Shell 和 Git 工具仍然和 Harness 写在同一个进程里。每接入一个独立服务，我们都要重新处理工具发现、参数 schema、连接和错误。

下一章实现 MCP Server 与 Client：让能力跨过进程边界，同时继续服从已经建立的权限、Task 与 Worktree 约束。