S18: Worktree Isolation — 代码解析
相对于 S17 新增：git worktree 隔离 + Task.worktree 字段 + 名称校验 + 事件日志 + 3 个 Lead 工具 + Teammate cwd 绑定
1. 整体功能概览
01
解决什么问题
多个 teammate 并发编辑同一分支会产生冲突。S18 给每个任务分配独立的 git worktree（独立工作目录 + 独立分支 wt/name），彼此完全隔离，互不干扰。
02
系统中的角色
Worktree 是任务的「沙盒」：创建 → 绑定任务 → teammate claim 时自动切换 cwd → 完成后 remove（有改动时保留或强制删除）。整个生命周期都写入 events.jsonl 可审计。
03
新增模块
Task.worktree 字段、validate_worktree_name()、run_git()、log_event()、create/remove/keep_worktree()、bind_task_to_worktree()、_count_worktree_changes()、Teammate wt_ctx + cwd 绑定逻辑。
2. Task.worktree — 新增字段
变更S17 Task dataclass 无 worktree 字段。S18 追加一个带默认值的可选字段。
| @dataclass
class Task:
id: str
subject: str
description: str
status: str
owner: str | None
blockedBy: list[str]
worktree: str | None = None # s18: bound worktree name |
新字段必须放在末尾（有默认值的字段不能在无默认值字段之前）；= None 使其可选；str | None 联合类型（Python 3.10+）；序列化到 JSON 时 null |
字段位置规则：
dataclass 字段顺序 = __init__ 参数顺序。有默认值的字段必须排在无默认值字段之后，否则报 TypeError。
JSON 序列化： asdict(task) 包含 worktree 字段；None → JSON null
反序列化： Task(**json.loads(...)) — 旧的 JSON（无 worktree 字段）会 TypeError，需数据迁移或 .get()
3. validate_worktree_name — 安全校验
新增Worktree 名称会成为文件系统路径的一部分，必须在执行 git 命令前校验，防止路径遍历和注入。
| VALID_WT_NAME = re.compile(r'^[A-Za-z0-9._-]{1,64}$') |
模块级正则；re.compile() 预编译，性能好；^ $ 锚定完整字符串；{1,64} 长度限制 |
| def validate_worktree_name(name: str) -> str | None: |
返回错误信息字符串或 None（无错）；调用方 if err: return f"Error: {err}" |
| if not name: |
空字符串为 falsy；覆盖 "" 的情况 |
| if name == "." or name == "..": |
路径遍历保护；"." 和 ".." 在文件系统中有特殊含义 |
| if not VALID_WT_NAME.match(name): |
Pattern.match() 从头开始匹配；配合 ^ 等价于 fullmatch；字符白名单：字母/数字/点/下划线/连字符 |
| return None |
无错误时返回 None；调用方用 if err: 判断 |
示例 — 校验案例：
validate_worktree_name("auth") → None (OK)
validate_worktree_name("../etc") → "Invalid worktree name..." (路径遍历)
validate_worktree_name("auth feature") → "Invalid..." (含空格)
validate_worktree_name("") → "Worktree name cannot be empty"
validate_worktree_name("a" * 65) → "Invalid..." (超过64字符)
4. run_git — 统一 git 命令执行器
新增S17 直接用 subprocess.run 执行 bash。S18 将 git 操作封装为返回 (bool, str) 的函数，让调用方无需解析 returncode。
| def run_git(args: list[str]) -> tuple[bool, str]: |
返回值类型注解：tuple[bool, str]；Python 3.9+ 语法（不需要 Tuple 大写） |
| try:
r = subprocess.run(["git"] + args, cwd=WORKDIR, |
["git"] + args 列表拼接；避免 shell=True（更安全）；cwd=WORKDIR 在主仓库执行 |
| capture_output=True, text=True, timeout=30) |
capture_output=True 等价于 stdout=PIPE stderr=PIPE；text=True 返回字符串；timeout=30（git 命令通常快，30s 足够） |
| out = (r.stdout + r.stderr).strip()
out = out[:5000] if out else "(no output)" |
合并 stdout+stderr；限制 5000 字符防止大输出；三元表达式 |
| return r.returncode == 0, out |
returncode == 0 为 bool；Python 元组无需括号；调用方可 ok, result = run_git(...) |
| except subprocess.TimeoutExpired:
return False, "Error: git timeout" |
超时异常处理；返回 (False, 错误信息)，与正常失败格式一致 |
示例 — 调用方模式：
ok, result = run_git(["worktree", "add", str(path), "-b", "wt/auth", "HEAD"])
if not ok:
return f"Git error: {result}"
事件仅在成功后记录：
run_git 返回 (True, ...) → log_event("create", name, task_id)
run_git 返回 (False, ...) → 直接返回错误，不记录事件
5. log_event — Worktree 事件日志
新增所有 worktree 生命周期事件（create/remove/keep）都追加到 .worktrees/events.jsonl，供审计和调试。
| def log_event(event_type: str, worktree_name: str, task_id: str = ""): |
task_id 可选（默认空字符串）；remove/keep 时不一定有关联任务 |
| event = {"type": event_type, "worktree": worktree_name,
"task_id": task_id, "ts": time.time()} |
dict 字面量；time.time() Unix 时间戳（浮点秒） |
| events_file = WORKTREES_DIR / "events.jsonl" |
/ 运算符拼接 Path；JSONL = JSON Lines，每行一个 JSON 对象 |
| with open(events_file, "a") as f: |
mode "a" = append；文件不存在时自动创建 |
| f.write(json.dumps(event) + "\n") |
json.dumps() 序列化为单行 JSON；加 "\n" 分隔各条记录 |
events.jsonl 示例内容：
{"type":"create","worktree":"auth","task_id":"task_001","ts":1718755200.123}
{"type":"create","worktree":"ui","task_id":"task_002","ts":1718755201.456}
{"type":"remove","worktree":"auth","task_id":"","ts":1718755300.789}
{"type":"keep","worktree":"ui","task_id":"","ts":1718755310.012}
6. create_worktree — 创建隔离工作区
| def create_worktree(name: str, task_id: str = "") -> str: |
task_id 可选；不绑定任务时纯粹创建 worktree |
| err = validate_worktree_name(name)
if err:
return f"Error: {err}" |
先校验；校验失败早返回（early return 模式），避免嵌套 |
| path = WORKTREES_DIR / name |
pathlib / 运算符；构造 .worktrees/auth 路径 |
| if path.exists():
return f"Worktree '{name}' already exists at {path}" |
幂等保护；path.exists() 检查文件系统 |
| ok, result = run_git(["worktree", "add",
str(path), "-b", f"wt/{name}", "HEAD"]) |
str(path) 将 Path 转字符串；-b wt/auth 创建新分支；HEAD 从当前提交开始 |
| if not ok:
return f"Git error: {result}" |
git 失败时不继续；不调用 bind/log |
| if task_id:
bind_task_to_worktree(task_id, name) |
字符串 truthy 检查；有 task_id 才绑定 |
| log_event("create", name, task_id) |
记录事件；仅在 git 成功后才记录 |
示例：
create_worktree("auth", "task_001")
1. validate_worktree_name("auth") → None (OK)
2. path = .worktrees/auth
3. path.exists() → False
4. run_git(["worktree","add",".worktrees/auth","-b","wt/auth","HEAD"])
→ (True, "Preparing worktree...")
5. bind_task_to_worktree("task_001", "auth")
→ task.worktree = "auth" → save_task(task)
6. log_event("create", "auth", "task_001")
→ events.jsonl 追加一行
返回: "Worktree 'auth' created at .worktrees/auth"
bind_task_to_worktree
| def bind_task_to_worktree(task_id: str, worktree_name: str):
task = load_task(task_id)
task.worktree = worktree_name
save_task(task) |
只修改 worktree 字段；status 保持 pending（让 idle_poll 的 scan_unclaimed_tasks 能找到并 auto-claim） |
7. remove_worktree & _count_worktree_changes
_count_worktree_changes — 安全检查
| def _count_worktree_changes(path: Path) -> tuple[int, int]: |
返回 (未提交文件数, 未推送提交数)；-1 表示检查失败 |
| r1 = subprocess.run(["git","status","--porcelain"],
cwd=path, ...) |
--porcelain 机器友好输出；每行一个文件；cwd=path 在 worktree 目录执行 |
| files = len([l for l in r1.stdout.strip().splitlines() if l.strip()]) |
列表推导 + 过滤空行；len() 计数 |
| r2 = subprocess.run(["git","log","@{push}..HEAD","--oneline"],
cwd=path, ...) |
@{push}..HEAD = 本地有但远端没有的提交；--oneline 每行一个提交 |
| commits = len([l for l in r2.stdout.strip().splitlines() if l.strip()]) |
同上模式统计提交数 |
| except Exception:
return -1, -1 |
任何异常（超时、git 错误）返回 (-1,-1)；调用方用 files < 0 判断 |
remove_worktree — 带保护的删除
| def remove_worktree(name: str,
discard_changes: bool = False) -> str: |
discard_changes=False 默认拒绝有改动的 worktree |
| if not discard_changes:
files, commits = _count_worktree_changes(path)
if files < 0:
return "Cannot verify..."
if files > 0 or commits > 0:
return f"Worktree '{name}' has {files} file(s)..." |
先检查后删除；files < 0 表示检查失败也拒绝；files > 0 or commits > 0 有任何改动就拒绝 |
| ok1, _ = run_git(["worktree","remove",str(path),"--force"]) |
_ 忽略第二个返回值；--force 强制移除（即使有未提交改动，因为我们已经在上面检查过了） |
| run_git(["branch","-D",f"wt/{name}"]) |
-D 强制删除分支（不需要 merge）；不检查返回值（分支可能不存在） |
| log_event("remove", name) |
记录删除事件；task_id="" 默认值 |
示例 — 有改动时的保护：
remove_worktree("auth")
→ _count_worktree_changes(.worktrees/auth)
→ files=2 (2个未提交文件), commits=1 (1个未推送提交)
→ "Worktree 'auth' has 2 file(s) and 1 commit(s). Use discard_changes=true or keep_worktree."
remove_worktree("auth", discard_changes=True)
→ 跳过检查
→ run_git(["worktree","remove",".worktrees/auth","--force"]) → (True, ...)
→ run_git(["branch","-D","wt/auth"]) → (True, ...)
→ log_event("remove","auth")
返回: "Worktree 'auth' removed"
8. keep_worktree — 保留用于人工审查
| def keep_worktree(name: str) -> str:
err = validate_worktree_name(name)
if err:
return err
log_event("keep", name)
print(f" [worktree] kept: {name}")
return f"Worktree '{name}' kept for review (branch: wt/{name})" |
不执行任何 git 操作；仅记录 "keep" 事件，表示有人知道这个 worktree 有改动但选择保留。 分支 wt/name 继续存在，可以手动 merge 或 cherry-pick。 |
keep vs remove 的设计意图：
remove(discard_changes=True) → 丢弃所有工作，删除分支
keep → "我知道有未保存工作，先不管它，手动处理"
典型场景：teammate 编写了有价值但有 bug 的代码 → Lead 决定 keep → 人工 review → 手动 merge 好的部分
9. Teammate cwd 绑定 — 工具自动指向 Worktree 目录
新增Teammate claim 一个有 worktree 的任务时，后续所有 bash/read/write 工具自动在该 worktree 目录下执行，不需要 LLM 手动 cd。
| # Track current worktree for this teammate's cwd
wt_ctx = {"path": None} |
dict 作为可变容器；闭包内可修改（list/dict 可就地 mutate，不需要 nonlocal） |
| def _wt_cwd() -> Path | None:
p = wt_ctx["path"]
return Path(p) if p else None |
辅助函数；p 有值时转 Path，None 时返回 None（让工具用 WORKDIR 默认值） |
| def _run_bash(command: str) -> str:
return run_bash(command, cwd=_wt_cwd()) |
闭包捕获 _wt_cwd；每次调用时动态获取当前 cwd |
| def _run_claim_task(task_id: str):
result = claim_task(task_id, owner=name)
if "Claimed" in result:
# Set worktree cwd if task has one
task = load_task(task_id)
if task.worktree:
wt_ctx["path"] = str(WORKTREES_DIR / task.worktree)
else:
wt_ctx["path"] = None
return result |
claim 成功后检查 task.worktree；有 worktree 时更新 wt_ctx["path"]；后续 bash/read/write 自动用新 cwd |
| def _run_complete_task(task_id: str):
result = complete_task(task_id)
wt_ctx["path"] = None
return result |
完成任务后重置 cwd = None；下一个任务可能没有 worktree |
示例 — cwd 自动切换：
wt_ctx = {"path": None}
_run_bash("pwd") → 在 WORKDIR/. 执行
_run_claim_task("task_001")
→ claim_task("task_001", owner="alice") → "Claimed task_001"
→ load_task("task_001").worktree = "auth"
→ wt_ctx["path"] = ".worktrees/auth"
_run_bash("git status") → 在 .worktrees/auth/ 执行
_run_read("src/main.py") → 读取 .worktrees/auth/src/main.py
_run_write("README.md", "...") → 写到 .worktrees/auth/README.md
_run_complete_task("task_001") → wt_ctx["path"] = None
10. 目录拓扑 & 完整流程
目录结构
主仓库 (WORKDIR = ./)
├── .git/ (主仓库 git 数据)
├── .tasks/
│ ├── task_001.json {worktree: "auth"}
│ └── task_002.json {worktree: "ui"}
├── .worktrees/
│ ├── auth/ (独立工作目录, branch: wt/auth)
│ │ ├── .git (指向主 .git 的文件)
│ │ └── src/ (teammate alice 在此工作)
│ ├── ui/ (独立工作目录, branch: wt/ui)
│ │ └── ... (teammate bob 在此工作)
│ └── events.jsonl (生命周期事件日志)
└── .mailboxes/
├── lead.jsonl
├── alice.jsonl
└── bob.jsonl
完整操作流程
Lead:
1. create_task("Build auth") → task_001
2. create_worktree("auth", "task_001")
→ git worktree add .worktrees/auth -b wt/auth HEAD
→ bind_task_to_worktree(task_001, "auth")
→ log_event("create","auth","task_001")
3. spawn_teammate("alice","backend","Help with tasks")
alice (auto-IDLE):
4. scan_unclaimed_tasks() → [task_001 (pending,worktree:"auth")]
5. claim_task(task_001, owner="alice")
→ wt_ctx["path"] = ".worktrees/auth"
6. WORK: bash("git branch") → 在 .worktrees/auth 执行
7. WORK: write_file("src/auth.py",...) → .worktrees/auth/src/auth.py
8. complete_task(task_001) → wt_ctx["path"] = None
Lead (收到 result):
9. check_inbox() → "alice: Done."
10. remove_worktree("auth") → (无改动时) 成功
或 keep_worktree("auth") → 保留分支 wt/auth