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

上一篇为 Agent Team 增加了审批与优雅退出协议。成员可以安全接收高风险操作审批，也能在关机前停止认领、处理后台 Job 并保存 Task 现场。

但团队仍然存在一个很现实的问题：多个 Agent 可能同时写入同一个项目目录。

假设 Backend Agent 正在修改 API，Frontend Agent 同时更新共享类型：

Backend 尚未完成的文件可能被 Frontend 读到；
两个 Agent 可能同时修改同一个文件，后写入的一方覆盖前者；
一个 Agent 执行格式化，可能改动另一个 Agent 正在编辑的文件；
两组测试共享构建目录、临时文件或依赖安装状态；
某个 Task 执行 git reset，可能破坏整个团队的现场。

Task Store 可以阻止两个成员同时认领同一个 Task，却无法阻止两个不同 Task 修改同一个工作区。

本章为每个 Task 分配独立的 Git linked worktree，解决五个问题：

worktree 能隔离什么，又不能隔离什么；
如何从稳定的 base commit 创建任务分支和工作目录；
如何把文件工具、Shell、依赖与端口绑定到任务工作区；
Task 完成后如何提交、集成、处理冲突；
Harness 崩溃后如何恢复并安全清理 worktree。

一个仓库，多份独立工作目录

Git 仓库可以拥有一个 main worktree 和多个 linked worktree。它们共享对象数据库与大部分 refs，但各自拥有独立的 HEAD、index 和工作目录。

这正适合 Agent Team：

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

Backend Agent 在 task-api 中编辑时，不会改变 Frontend Agent 的 task-ui 目录；每个 Task 也有自己的 index、未跟踪文件和当前分支。

任务工作区彼此独立，但 Git 对象和 refs 仍然共享。只有受控的 Git Broker 可以创建分支、提交和集成。

Git worktree 官方文档还说明了几个重要行为：

同一个本地分支默认不能同时被两个 worktree checkout；
worktree list --porcelain -z 提供适合脚本解析的稳定输出；
活跃 worktree 可以锁定，避免被移动、删除或自动 prune；
清理应该使用 git worktree remove，不要直接删除目录。

Worktree 不是安全沙箱

Worktree 提供的是文件与 index 隔离，不是完整的权限隔离。

能隔离	不能自动隔离
tracked 与 untracked 文件	Git 对象数据库和大部分 refs
每个工作区的 index 与 `HEAD`	网络、数据库、云资源和外部 API
工作目录内的依赖与构建产物	固定端口、全局临时目录和系统进程
不同任务尚未提交的文件修改	用户目录下的缓存、凭据和配置

这意味着仅仅把 Shell 的 cwd 改到 worktree 还不够。命令仍然可以执行 cd ..，访问主项目目录，或者通过 Git 修改共享分支。

Harness 还需要两层约束：

进程沙箱：只把当前 worktree 挂载为可写，主工作区、其它 worktree 和 .agent Runtime 状态不可写；
Git Broker：Agent 不能随意执行修改 refs 的 Git 命令，提交、创建和清理 worktree 由受控工具完成。

Worktree 负责减少协作干扰，权限系统负责阻止越界。两者不能互相替代。

一条 Worktree Record 保存什么

Worktree 是 Task 的运行资源，不应该只存在于内存。先定义持久化记录：

from dataclasses import asdict, dataclass
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
    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] | None = None
    branch_cleanup_error: str | None = None
    created_at: str = ""
    updated_at: str = ""
    removed_at: str | None = None

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

Task 增加一个可选的 workspace_id，用于找到自己的 Worktree Record。真正的 Task 所有权仍由 claimed_by + claim_token 管理；知道 worktree 路径不等于有权修改它。

最小生命周期是：

allocating -> active -> ready_for_integration
                              |
                              +-> integrating -> integrated -> removing -> removed
                                      |
                                      +-> conflicted -> active

retained 表示 Harness 刻意保留现场，例如存在未提交修改、清理条件不满足或需要人工检查。它不是失败后偷偷遗留的未知目录。

路径和分支名由 Harness 生成

模型不能传入任意 worktree 路径或分支名。Harness 从经过校验的 Task ID 派生：

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("worktree path escapes configured root")
    return candidate


def ensure_worktree_root_is_external(
    repo_root: Path,
    worktree_root: Path,
) -> None:
    repo = repo_root.resolve()
    root = worktree_root.resolve()
    if root == repo or repo in root.parents or root in repo.parents:
        raise ValueError("repository and worktree root must be disjoint")

实际 worktree 不建议放在主仓库的 .agent/worktrees/ 下。嵌套目录容易出现在主工作区的 git status 中，也会让路径与清理边界更难理解。

可以把 Worktree Record 保存在项目 .agent 目录，把实际工作目录放在独立 Runtime Root：

project/
└── .agent/worktrees/records/task-api.json

.project-agent-runtime/
└── worktrees/task-api/

Runtime Root 必须来自 Harness 配置和 allowlist，不能由模型决定。

Git 命令使用参数数组

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

import os
import subprocess
from pathlib import Path


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:
        message = result.stderr.decode("utf-8", errors="replace")[-2000:]
        raise GitCommandError(message)
    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_* 变量。需要签名、远程访问或 hooks 时，由 Harness 显式加入允许项。

创建前先固定 base commit

任务工作区不应该隐式跟随不断移动的 main 或 integration branch。Task 被分配工作区时，Harness 先把基线解析成不可变 commit OID。

如果 task-api 和 task-ui 可以并行，它们可以从同一个 integration commit 开始。依赖任务只有在上游结果完成集成后才会 ready，因此后续 worktree 会自然基于包含上游结果的新 commit。

创建过程遵循“先写意图，再执行副作用”：

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,
            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)

put_if_absent() 比较 Task、owner、repo、path、branch 和 base commit。相同请求返回原记录，冲突内容则拒绝。

repo_operation_lock 只保护 worktree 创建、删除和关键 ref 操作。不同 Agent 在各自工作区编辑、安装依赖和测试时不持有这把锁，否则并行执行就失去意义。

解析 Git 的真实 Worktree 列表

不能只相信 JSON Record。进程可能在 git worktree add 成功后、Record 更新前退出。

Git 为脚本提供 worktree list --porcelain -z。-z 使用 NUL 分隔，即使路径包含空格或换行也可以安全解析：

def list_git_worktrees(repo_root: Path) -> list[dict[str, str | bool]]:
    result = run_git(
        repo_root,
        "worktree",
        "list",
        "--porcelain",
        "-z",
    )

    records: list[dict[str, str | bool]] = []
    current: dict[str, str | bool] = {}
    for field in result.stdout.split(b"\0"):
        if not field:
            if current:
                records.append(current)
                current = {}
            continue

        key, separator, value = field.partition(b" ")
        name = key.decode("ascii")
        current[name] = os.fsdecode(value) if separator else True

    if current:
        records.append(current)
    return records

branch 字段使用完整 ref，例如 refs/heads/agent/task/task-api。恢复逻辑比较分支时，要把 Record 中的短分支名补成 refs/heads/{record.branch}，不能直接比较两个字符串。

恢复分配时分三种情况：

path 已注册，branch 与 Record 一致：验证后进入 active；
branch 已创建但 path 未注册：从该 branch 重新执行 worktree add；
path 已存在却不属于 Git worktree：保留现场并报错，不能直接删除。

真正创建时使用唯一任务分支，并在同一个命令里锁定：

def create_linked_worktree(
    state: HarnessState,
    record: WorktreeRecord,
    branch_exists: bool,
) -> None:
    path = Path(record.path)
    if path.exists():
        raise GitCommandError("unregistered worktree path already exists")
    if (
        branch_exists
        and resolve_commit(
            state.repo_root,
            f"refs/heads/{record.branch}",
        )
        != record.base_commit
    ):
        raise GitCommandError("existing branch does not match base commit")

    arguments = [
        "worktree",
        "add",
        "--lock",
        "--reason",
        f"active agent task {record.task_id}",
    ]
    if branch_exists:
        arguments.extend([str(path), record.branch])
    else:
        arguments.extend(
            ["-b", record.branch, str(path), record.base_commit]
        )
    run_git(state.repo_root, *arguments)

这里不使用 --force 或 -B。它们会绕过 Git 对重复 checkout、缺失 worktree 和已有分支的保护，自动化恢复不应该用“强制重置”掩盖冲突。

每个工具都绑定 Task Workspace

Agent 不应该在每次调用 read_file、edit_file 或 bash 时自行传入 workspace。Harness 根据当前 Task 注入：

@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

resolve() 也会处理已有 symlink，避免通过链接跳出工作区。创建新文件时，还要对最近的已存在父目录执行同样检查。

Shell 工具至少注入：

cwd     = task worktree
TMPDIR  = task-specific temp directory
PORT    = port allocated to this task
TASK_ID = current task id

但 cwd 和路径检查仍不是 OS 安全边界。Shell 进程要运行在沙箱或容器中，只挂载允许目录。主工作区、其它 worktree、Protocol Store 和 Task Store 不应该对任务进程可写。

依赖和外部资源也要隔离

独立工作目录会让 node_modules、虚拟环境、构建目录和测试快照自然分开，但它们可能很大。

可以共享只读或内容寻址的下载缓存，例如包管理器全局 cache；不能共享会被任务直接修改的构建输出。

Worktree 之外的资源还需要单独命名：

为每个 Task 分配不同服务端口；
测试数据库使用独立 schema 或容器；
临时目录放在 task-specific Runtime Root；
浏览器 profile、日志目录和覆盖率文件按 Task 隔离；
后台 Job 记录 workspace_id，恢复时回到同一个目录。

如果测试脚本写死端口或数据库名称，worktree 再独立也会互相干扰。

Task 结果是一份经过验证的 Commit

Agent 完成修改后，不要把“worktree 里有文件”当成可交接结果。Harness 要生成稳定 commit：

def changed_paths(worktree: Path) -> set[str]:
    tracked = run_git(
        worktree,
        "diff",
        "--name-only",
        "-z",
    ).stdout
    staged = run_git(
        worktree,
        "diff",
        "--cached",
        "--name-only",
        "-z",
    ).stdout
    untracked = run_git(
        worktree,
        "ls-files",
        "--others",
        "--exclude-standard",
        "-z",
    ).stdout
    return {
        os.fsdecode(path)
        for path in (
            tracked.split(b"\0")
            + staged.split(b"\0")
            + untracked.split(b"\0")
        )
        if path
    }


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("wait for background jobs before preparing")

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

    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 from HEAD")
    return candidate_tree


def create_task_commit(
    state: HarnessState,
    task: Task,
    record: WorktreeRecord,
    candidate_tree: str,
    verification: 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 index 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 candidate was prepared")

    ensure_verification_passed(verification, candidate_tree)
    run_git(
        worktree,
        "-c",
        "user.name=Agent Harness",
        "-c",
        "user.email=agent@local.invalid",
        "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 hooks changed the candidate tree")
    if run_git(
        worktree,
        "status",
        "--porcelain",
        "-z",
    ).stdout:
        raise ValueError("task worktree is not clean after commit")
    return result_commit

这里分成准备与提交两步：

prepare_task_candidate() 暂存当前修改，得到唯一的 Git tree OID；
Harness 暂停该工作区的编辑工具，只允许验证 Job 读取候选内容；
每个验证结果同时记录 Job ID 和 candidate_tree；
create_task_commit() 确认 index、工作目录和验证结果仍属于同一棵 tree，再创建 commit。

测试过程可以生成被 .gitignore 排除的缓存或报告，但不能修改候选源码。只要候选 tree、未暂存文件或未跟踪文件发生变化，已有验证就失效，必须重新准备并测试。

ensure_task_paths_allowed() 可以根据 Task 描述、目录所有权或显式 allowlist 拒绝无关修改。scan_for_secrets() 在 staging 前检查未跟踪文件，避免把 .env、私钥或令牌提交进 Git。verification 只接受 Harness 生成的 Job ID，并由 ensure_verification_passed() 查询真实退出状态与 tree OID，不能相信模型提交的“测试已通过”文本。

是否运行 commit hooks 由 Harness Policy 决定，不能让模型临时加 --no-verify。如果 hooks 来自不受信任仓库，它们也必须运行在任务沙箱中。

提交成功后：

Worktree Record 写入 result_commit；
状态进入 ready_for_integration；
Task checkpoint 保存 commit、验证命令和风险；
Task 暂时保持 in_progress，等待集成。

为什么不立刻 task_complete？因为下游 Task 从 integration branch 创建 worktree。如果上游 commit 还没有进入该分支，下游即使依赖已解锁，也看不到上游结果。

Integration Worker 串行合并

普通成员只修改自己的 task branch。唯一的 Integration Worker 在独立 integration worktree 中串行处理结果：

任务提交只是候选结果。通过合并与集成测试后，Task 才完成并解锁下游工作。

集成前再次固定四件事：

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

def is_ancestor(repo: Path, ancestor: str, descendant: str) -> bool:
    result = run_git(
        repo,
        "merge-base",
        "--is-ancestor",
        ancestor,
        descendant,
        check=False,
    )
    if result.returncode not in {0, 1}:
        raise GitCommandError("cannot compare commits")
    return result.returncode == 0


def worktree_is_clean(path: Path) -> bool:
    return not run_git(
        path,
        "status",
        "--porcelain",
        "-z",
    ).stdout

如果 commit 已经是 integration HEAD 的祖先，说明此前集成成功但 Record 更新中断。在 integration branch 只允许 Integration Worker 修改的前提下，恢复逻辑可以直接补写 integrated，不能再次 merge。

否则，在全局 integration_lock 内执行一次不提交的 merge：

@dataclass(frozen=True)
class MergePreparation:
    status: Literal[
        "already_integrated",
        "ready_for_tests",
        "conflicted",
    ]
    conflict_paths: tuple[str, ...] = ()


def merge_task_for_verification(
    state: HarnessState,
    record: WorktreeRecord,
) -> MergePreparation:
    integration_root = state.integration_worktree
    if not worktree_is_clean(integration_root):
        raise RuntimeError("integration worktree is not clean")
    if record.result_commit is None:
        raise RuntimeError("task has no submitted commit")
    if not is_ancestor(
        state.repo_root,
        record.base_commit,
        record.result_commit,
    ):
        raise RuntimeError("submitted commit does not descend from base")

    integration_head = resolve_commit(integration_root, "HEAD")
    if is_ancestor(
        state.repo_root,
        record.result_commit,
        integration_head,
    ):
        return MergePreparation(status="already_integrated")

    branch_head = resolve_commit(
        state.repo_root,
        f"refs/heads/{record.branch}",
    )
    if branch_head != record.result_commit:
        raise RuntimeError("task branch changed after submission")

    merge = run_git(
        integration_root,
        "-c",
        "user.name=Agent Harness",
        "-c",
        "user.email=agent@local.invalid",
        "merge",
        "--no-ff",
        "--no-commit",
        record.result_commit,
        check=False,
    )
    if merge.returncode == 0:
        merge_head = run_git(
            integration_root,
            "rev-parse",
            "--quiet",
            "--verify",
            "MERGE_HEAD",
            check=False,
        )
        if merge_head.returncode != 0:
            raise GitCommandError("merge has no pending MERGE_HEAD")
        return MergePreparation(status="ready_for_tests")

    conflicts = run_git(
        integration_root,
        "diff",
        "--name-only",
        "--diff-filter=U",
        "-z",
    ).stdout
    paths = [
        os.fsdecode(path)
        for path in conflicts.split(b"\0")
        if path
    ]
    merge_head = run_git(
        integration_root,
        "rev-parse",
        "--quiet",
        "--verify",
        "MERGE_HEAD",
        check=False,
    )
    if merge_head.returncode != 0:
        message = merge.stderr.decode("utf-8", errors="replace")[-2000:]
        raise GitCommandError(message)

    run_git(integration_root, "merge", "--abort")
    if not paths:
        message = merge.stderr.decode("utf-8", errors="replace")[-2000:]
        raise GitCommandError(message)
    return MergePreparation(
        status="conflicted",
        conflict_paths=tuple(paths),
    )

already_integrated 直接进入状态修复，conflicted 保存路径并回传原 Task。只有 ready_for_tests 表示 integration worktree 中存在待提交的 merge。

没有冲突时，Integration Worker 在尚未提交的合并结果上运行集成测试：

测试通过：使用固定的 Harness identity 创建 merge commit，记录 integrated_commit；
测试失败：保存日志，执行 git merge --abort；
测试修改 tracked 文件：视为不干净测试，拒绝提交并 abort；
Harness 中断：恢复时检测 MERGE_HEAD，保存现场后 abort，再安全重试。

只有 merge commit 和集成测试都成功后，Harness 才调用 task_complete。这样下游 Task 的 base commit 一定包含已经验收的上游结果。

冲突回到任务工作区解决

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

Integration Worker 不应该在共享 integration worktree 中长时间等待模型解决。它先：

记录冲突文件和当前 integration commit；
git merge --abort 恢复干净状态；
把 Worktree Record 标为 conflicted；
向原成员或具备对应 capability 的成员发送 blocker。

责任成员在自己的 task worktree 中合并最新 integration branch、解决冲突、重新测试并提交。新的 result_commit 再次进入集成队列。

这种方式让 integration worktree 始终是短时、串行的受控资源，不会因为一个冲突阻塞所有已经完成的 Task。

安全回收 Worktree

只有满足下面条件，Task worktree 才能回收：

结果已经集成；
没有仍以该目录为 cwd 的后台 Job；
git status --porcelain -z 为空；
重要日志和 artifact 已经复制到持久化位置；
Worktree Record 与 Git 列表一致。

清理顺序：

def find_git_worktree(
    repo_root: Path,
    path: Path,
) -> dict[str, str | bool] | None:
    expected = str(path.resolve(strict=False))
    return next(
        (
            entry
            for entry in list_git_worktrees(repo_root)
            if entry.get("worktree") == expected
        ),
        None,
    )


def remove_integrated_worktree(
    state: HarnessState,
    record: WorktreeRecord,
) -> None:
    path = Path(record.path)
    if record.status not in {"integrated", "removing"}:
        raise ValueError("worktree is not ready for removal")
    if state.job_manager.has_running_jobs(record.id):
        raise ValueError("worktree still has running jobs")
    if not state.artifact_store.is_persisted(record.task_id):
        raise ValueError("task artifacts are not persisted")

    with state.repo_operation_lock:
        entry = find_git_worktree(state.repo_root, path)
        if entry is not None:
            if not worktree_is_clean(path):
                state.worktree_store.mark_retained(
                    record.id,
                    "worktree contains uncommitted files",
                )
                return
            if (
                record.result_commit is None
                or resolve_commit(path, "HEAD") != record.result_commit
            ):
                state.worktree_store.mark_retained(
                    record.id,
                    "worktree HEAD differs from submitted commit",
                )
                return

            if record.status == "integrated":
                state.worktree_store.mark_removing(record.id)
            if "locked" in entry:
                run_git(
                    state.repo_root,
                    "worktree",
                    "unlock",
                    str(path),
                )
            removal = run_git(
                state.repo_root,
                "worktree",
                "remove",
                str(path),
                check=False,
            )
            if removal.returncode != 0:
                message = removal.stderr.decode(
                    "utf-8",
                    errors="replace",
                )[-2000:]
                state.worktree_store.mark_retained(
                    record.id,
                    message,
                )
                return
        elif record.status != "removing":
            raise GitCommandError("registered worktree unexpectedly missing")

        branch_ref = f"refs/heads/{record.branch}"
        branch_exists = run_git(
            state.repo_root,
            "show-ref",
            "--verify",
            "--quiet",
            branch_ref,
            check=False,
        )
        branch_error: str | None = None
        if branch_exists.returncode == 0:
            branch_delete = run_git(
                state.integration_worktree,
                "branch",
                "-d",
                record.branch,
                check=False,
            )
            if branch_delete.returncode != 0:
                branch_error = branch_delete.stderr.decode(
                    "utf-8",
                    errors="replace",
                )[-2000:]
        elif branch_exists.returncode != 1:
            raise GitCommandError("cannot inspect task branch")

        state.worktree_store.mark_removed(
            record.id,
            removed_at=utc_now_text(),
            branch_cleanup_error=branch_error,
        )

不使用 worktree remove --force，也不使用 git branch -D。如果普通删除失败，说明还有未提交文件、submodule 或 Git 元数据不一致，Harness 应保留现场并要求检查。包含 submodule 的仓库需要单独的回收策略，不能因为 Git 要求 --force 就自动放宽删除条件。

removing 是清理意图的持久化状态。如果进程在目录删除后、Record 更新前退出，恢复逻辑会跳过已经消失的 worktree，继续尝试删除任务分支并补写 removed。分支普通删除失败时保留 ref 和错误记录；worktree 已经回收，不需要为了掩盖这项残留而执行 branch -D。

git worktree prune 只清理路径已经丢失的 stale administrative metadata，不是正常回收流程。可以定期先运行 prune --dry-run 生成审计报告，再由维护任务决定是否清理。

Harness 重启后的 Reconciliation

启动时同时读取 Worktree Store 与 git worktree list --porcelain -z：

Record 与 Git 现场	恢复动作
`allocating`，Git path 已注册	验证 branch/base，补写 `active`
`active`，path 与 branch 一致	恢复 Task Workspace 与后台 Job
非 `removing` Record，Git path 消失	标记 `failed`，保留 Task checkpoint
Git path 存在，没有对应 Record	隔离并通知 Lead，不自动删除
`integrating`，存在 `MERGE_HEAD`	保存冲突或测试现场，abort 后重新排队
`integrated`，目录干净且没有 Job	进入受控清理
`removing`，Git path 已消失	完成 branch 清理并补写 `removed`

还要校验每个 active Record 对应的 Task 是否仍由同一 Agent 和 claim token 持有。如果 Task 租约已经过期，worktree 不会立即删除；新的所有者先读取 checkpoint、commit 和工作目录状态，再决定接管或保留。

Worktree 路径只是可恢复资源，Task Store 仍然是工作状态的唯一事实来源。

暴露给 Agent 的工具

Agent 只需要看到少量高层工具：

工具	作用
`workspace_status`	查看当前 Task 的路径、branch 与状态
`workspace_changed_files`	查看允许范围内的修改文件
`workspace_prepare_commit`	校验并暂存候选 tree
`workspace_submit`	验证 tree、创建 commit 并进入集成队列
`workspace_integration_status`	查看集成、冲突与测试结果

创建、移动、解锁、删除 worktree，以及修改共享 refs 的 Git 命令不直接暴露给模型。

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

常见坑

第一，把 cwd 当成安全边界。 Shell 可以离开当前目录，还需要 OS 沙箱和挂载限制。

第二，把 worktree 放进主工作区。 嵌套目录会污染状态并增加误删风险，实际目录应位于独立 Runtime Root。

第三，创建时使用移动分支名作为隐式基线。 先解析并保存 base commit OID，保证恢复时语义不变。

第四，遇到已有 branch 就使用 -B。 这可能重置已经存在的任务提交，应该核对 Record 与 Git 现场。

第五，允许模型自由执行 Git 管理命令。 Worktree 共享 refs 和对象库，需要 Git Broker 控制危险操作。

第六，只隔离源码，不隔离端口和数据库。 外部资源冲突不会因为目录不同而消失。

第七，Task commit 后立即标记完成。 下游工作区可能看不到尚未集成的上游结果。

第八，让多个成员同时写 integration worktree。 集成必须串行，并且每次开始前检查工作区干净。

第九，在 integration worktree 中等待 Agent 慢慢解冲突。 保存冲突后 abort，把问题送回独立 task worktree。

第十，直接删除 worktree 目录。 使用 git worktree remove，否则会遗留 shared Git metadata。

第十一，默认使用 --force 清理。 未提交文件和 submodule 需要保留检查，不能被自动抹掉。

第十二，清理 worktree 时忽略后台 Job。 进程仍以该目录为 cwd 时删除文件，会破坏日志、输出和恢复能力。

小结

本文为 Agent Team 增加了 Task 级 Worktree Isolation：

每个 Task 使用唯一 branch、稳定 base commit 和独立 linked worktree；
Worktree Store 在 Harness 重启后恢复工作目录；
Git 的 porcelain NUL 格式用于可靠解析真实现场；
活跃 worktree 在创建时锁定，避免被自动 prune；
文件、Shell、临时目录、端口和测试资源都绑定当前 Task；
Worktree 只隔离工作目录，权限与外部资源仍由沙箱控制；
Agent 通过受控 Git Broker 创建经过验证的 result commit；
Integration Worker 串行 merge，并在提交前执行集成测试；
冲突保存为 blocker，回到任务 worktree 解决；
Task 只在结果完成集成后进入 completed；
清理前检查状态、Job 和 artifact，不使用强制删除；
reconciliation 对照 Record 与 Git 现场修复中断窗口。

现在，不同 Agent 可以并行修改同一个仓库，却不会直接踩进彼此尚未完成的工作目录。它们交付的也不再是一堆散落文件，而是可以验证、集成和回收的 commit。

到目前为止，文件、Shell、Task 和 Worktree 工具都直接注册在 Harness 进程中。这样的实现足以验证 Agent Loop，却很难让独立服务、第三方系统和其它 Agent 复用同一套能力。每接入一种外部工具都编写专用适配器，工具发现、参数校验、连接管理和错误处理也会逐渐散落在主循环里。

下一章讲 MCP 的实现：从一个最小的 MCP Server 与 MCP Client 开始，打通连接生命周期、能力发现、工具列表和工具调用，再把现有 Tool Registry 接入协议层。这样 Agent 不需要知道工具运行在哪个进程，只需要通过统一 schema 发现并调用能力，同时继续复用前面实现的审批、Job、Task 与 Worktree 边界。