先让两份 main.py 同时留下来
思想篇里的问题可以直接实验：主目录提交 print(0)，alpha写 print(6 * 7)，beta写 print(40 + 3)。两位队友自主认领，再分别写入和运行。
源码在独立公开仓库第十八课。需要Python 3.10+和Git，默认只使用标准库；本次验证为2026-10-20，Git 2.50.1。先运行公开入口，再按core/README.md映射阅读三个函数：
bash
git clone https://github.com/Chengyunlai/learn-agent-harness.git
cd learn-agent-harness
python3 examples/s18-worktree-isolation/user_code/practice.py --decision allow
python3 examples/s18-worktree-isolation/user_code/practice.py模型是固定替身，Git仓库、两条工作线程、任务JSON、文件修改和获准后的Python进程真实执行，没有在线模型调用。允许时两项completed，各自输出42/43；主目录仍是0，两个有改动的目录进入保留待审。默认拒绝写入和运行，两项保留in_progress，未产生改动的目录可以移除。
默认实验使用临时目录，示例进程结束后整体清理。要在结束后继续看真实文件，给它一个空目录：
bash
S18_DEMO_DIR=$(mktemp -d)
python3 examples/s18-worktree-isolation/user_code/practice.py \
--decision allow --state-dir "$S18_DEMO_DIR"
git -C "$S18_DEMO_DIR/repo" worktree list --porcelain这样留下repo、worktrees和board供人工查看。会话内绑定表与成员历史没有恢复机制，不能拿同一个目录再次启动并假定已接回旧会话。
create_workspace：先等 Git，再登记绑定
输入包括主仓库、工作树容器、任务看板、会话绑定表、任务ID和共享RLock。函数先检查任务ID对应的受控路径，再读取任务；只有pending且尚无owner的任务可以准备目录。
我们没有扩展旧任务JSON。任务继续保存依赖、负责人和进度，bindings单独保存路径、分支、创建起点和生命周期。绑定一项目录不会顺便认领任务。
真实执行轨迹回放；原样函数与锁，Git和看板读取使用可展开的内存替身；不创建真实工作树或启动程序。
Pys18/core/runtime.py · create_workspace
未运行
def create_workspace(repo, parent, directory, bindings, task_id, lock):
"""先创建真实工作树，再登记绑定；不认领、不完成任务。"""
with lock:
path, branch = workspace_target(repo, parent, task_id)
task = load_board(directory)['tasks'][task_id]
if task['status'] != 'pending' or task['owner'] is not None:
raise ValueError('只为尚未认领的任务准备目录。')
if task_id in bindings or len(bindings) >= 4:
raise ValueError('已有绑定或达到四项目录额度。')
base = git(repo, 'rev-parse', 'HEAD')
git(repo, 'worktree', 'add', '-b', branch, str(path), base)
record = {'path': str(path), 'branch': branch, 'base': base,
'status': 'ready', 'active': False}
bindings[task_id] = record
return deepcopy(record)输出
点击行号设置断点，再点“调试”；也可以直接运行。
教学准备与调用入口
上方源码保持原样。以下准备输入和调用入口，在单独文件名中执行；断点只回放上方源码。
from copy import deepcopy
from pathlib import Path
from threading import RLock
import re
def workspace_target(repo, parent, task_id):
if not isinstance(task_id, str) or not re.fullmatch(r'[a-z][a-z0-9-]{0,31}', task_id):
raise ValueError('任务ID仅接受小写字母、数字和短横线。')
repo, parent = Path(repo).resolve(), Path(parent).resolve()
if parent == repo or parent.is_relative_to(repo):
raise ValueError('工作树容器必须在主目录外。')
path = parent / task_id
if path.is_symlink() or path.resolve().parent != parent:
raise ValueError('工作树路径不允许符号链接。')
return path, 'codex/' + task_id
def checked_workspace(repo, parent, record, task_id):
path, branch = workspace_target(repo, parent, task_id)
if record['path'] != str(path) or record['branch'] != branch or record['status'] == 'removed':
raise ValueError('绑定路径或生命周期不符。')
if git(path, 'rev-parse', '--show-toplevel') != str(path):
raise ValueError('目录不再是登记的工作树。')
common = lambda cwd: (Path(cwd) / git(cwd, 'rev-parse', '--git-common-dir')).resolve()
if common(path) != common(repo) or git(path, 'branch', '--show-current') != branch:
raise ValueError('仓库归属或当前分支已改变。')
return path
def create_workspace(repo, parent, directory, bindings, task_id, lock):
"""先创建真实工作树，再登记绑定；不认领、不完成任务。"""
with lock:
path, branch = workspace_target(repo, parent, task_id)
task = load_board(directory)['tasks'][task_id]
if task['status'] != 'pending' or task['owner'] is not None:
raise ValueError('只为尚未认领的任务准备目录。')
if task_id in bindings or len(bindings) >= 4:
raise ValueError('已有绑定或达到四项目录额度。')
base = git(repo, 'rev-parse', 'HEAD')
git(repo, 'worktree', 'add', '-b', branch, str(path), base)
record = {'path': str(path), 'branch': branch, 'base': base,
'status': 'ready', 'active': False}
bindings[task_id] = record
return deepcopy(record)
def workspace_for_task(repo, parent, directory, bindings, task_id, actor, lock):
"""核对最新owner，预留目录给这一轮；工具在锁外用返回路径。"""
with lock:
task = load_board(directory)['tasks'][task_id]
if task['status'] != 'in_progress' or task['owner'] != actor:
raise ValueError('当前成员没有持有这项任务。')
record = bindings[task_id]
if record['active']:
raise ValueError('这项目录已经有活跃轮次。')
path = checked_workspace(repo, parent, record, task_id)
record['active'] = True
return str(path)
def retire_workspace(repo, parent, bindings, task_id, action, lock):
"""先停工作再收尾；默认保留，清理不强删、不删分支、不改任务。"""
if action not in ('keep', 'remove'):
raise ValueError('收尾动作只接受keep或remove。')
with lock:
record = bindings[task_id]
path = checked_workspace(repo, parent, record, task_id)
if record['active']:
return {'status': 'busy'}
if action == 'keep':
record['status'] = 'kept'
return {'status': 'kept', 'branch': record['branch'], 'path': str(path)}
dirty = git(path, 'status', '--porcelain', '--untracked-files=all', '--ignored=matching')
head = git(path, 'rev-parse', 'HEAD')
if dirty or head != record['base']:
return {'status': 'needs_review', 'dirty': bool(dirty), 'changed_head': head != record['base']}
git(repo, 'worktree', 'remove', str(path))
record['status'] = 'removed'
return {'status': 'removed', 'branch': record['branch']}
scenario='new'
repo=Path('/teaching-s18/repo');parent=Path('/teaching-s18/worktrees');directory='board';lock=RLock();task_id='alpha';actor='reader'
base='a'*40;bindings={};commands=[];present={};changed=False;dirty_text='';fail_add=False;fail_remove=False
board={'tasks':{'alpha':{'id':'alpha','status':'pending','owner':None}}}
def load_board(_):return deepcopy(board)
def git(cwd,*args):
key=str(cwd);commands.append(list(args))
if args[:2]==('worktree','add'):
if fail_add:raise ValueError('教学Git创建失败')
present[args[4]]=args[3];return ''
if args[:2]==('worktree','remove'):
if fail_remove:raise ValueError('教学Git删除失败')
present.pop(args[2]);return ''
if args==('rev-parse','HEAD'):return ('b'*40 if changed and key!=str(repo) else base)
if args==('rev-parse','--show-toplevel'):
if key not in present:raise ValueError('工作树不在登记中')
return key
if args==('rev-parse','--git-common-dir'):return str(repo/'.git')
if args==('branch','--show-current'):return 'unexpected' if scenario=='branch_changed' else present[key]
if args[0]=='status':return dirty_text
raise AssertionError('未知替身命令：'+str(args))
if 'create_workspace'!='create_workspace' or scenario=='duplicate':
create_workspace(repo,parent,directory,bindings,task_id,lock)
if 'create_workspace'=='create_workspace':
if scenario=='claimed':board['tasks']['alpha'].update(status='in_progress',owner='reader')
if scenario=='invalid':task_id='../outside'
if scenario=='git_error':fail_add=True
if scenario=='limit':bindings.update({str(i):{} for i in range(4)})
if 'create_workspace'=='workspace_for_task':
if scenario!='pending':board['tasks']['alpha'].update(status='in_progress',owner='reviewer' if scenario=='wrong_owner' else 'reader')
if scenario=='busy':bindings['alpha']['active']=True
if scenario=='removed':bindings['alpha']['status']='removed'
if 'create_workspace'=='retire_workspace':
if scenario=='dirty':dirty_text=' M main.py'
if scenario=='ignored':dirty_text='!! local-cache'
if scenario=='committed':changed=True
if scenario=='busy':bindings['alpha']['active']=True
if scenario=='remove_error':fail_remove=True
commands.clear()
action='keep' if scenario=='keep' else 'remove'
# 调用入口
try:
observed=create_workspace(repo,parent,directory,bindings,task_id,lock)
except (ValueError,KeyError) as error:
observed={"error":str(error)}
print(observed)
- 检查路径、待领任务和额度。
- Git成功后登记绑定，任务仍pending。
在Git调用行暂停，base已经是这次实际读取的HEAD，但Git命令尚未执行。下一步返回后，才会构造record，再把它放入bindings。选择“Git创建失败”，表里不会冒出一个ready记录；选择“已有绑定”，不会再次创建同名分支。
helper将ID限制为小写字母、数字和短横线，最多32字符，拒绝路径穿越和目标符号链接，容器必须在主目录外。每会话最多四项绑定；分支固定为codex/<ID>。Git使用参数数组，不经过shell，也不使用覆盖分支的-B或force。
起点是实际HEAD提交，不是主目录当前可见的全部内容。测试会把主目录main.py改成99并留一个未跟踪文件，再创建worktree；新目录仍取得提交中的0，没有那个额外文件。
Git命令与Python内存不是一个事务。创建超时或进程在Git成功后中断，可能留下尚未登记的目录；这时应先用worktree list核查，不能依据缺少绑定就强删或重复创建。每条Git命令有10秒timeout，失败不会自动重试。
workspace_for_task：认领之后，预留这一轮的路径
第十七课的claim_next继续负责在锁内重新读取并认领任务。拿到任务以后，本函数再读最新看板，要求状态为in_progress，owner等于宿主绑定的成员名。
负责人通过后，还不能直接相信表里的路径。checked_workspace重新核对受控路径、实际Git顶层目录、公共Git目录和当前分支，确认工具没有被接到另一个仓库或已经切换的分支。
真实执行轨迹回放；原样函数与锁，Git和看板读取使用可展开的内存替身；不创建真实工作树或启动程序。
Pys18/core/runtime.py · workspace_for_task
未运行
def workspace_for_task(repo, parent, directory, bindings, task_id, actor, lock):
"""核对最新owner，预留目录给这一轮；工具在锁外用返回路径。"""
with lock:
task = load_board(directory)['tasks'][task_id]
if task['status'] != 'in_progress' or task['owner'] != actor:
raise ValueError('当前成员没有持有这项任务。')
record = bindings[task_id]
if record['active']:
raise ValueError('这项目录已经有活跃轮次。')
path = checked_workspace(repo, parent, record, task_id)
record['active'] = True
return str(path)输出
点击行号设置断点，再点“调试”；也可以直接运行。
教学准备与调用入口
上方源码保持原样。以下准备输入和调用入口，在单独文件名中执行；断点只回放上方源码。
from copy import deepcopy
from pathlib import Path
from threading import RLock
import re
def workspace_target(repo, parent, task_id):
if not isinstance(task_id, str) or not re.fullmatch(r'[a-z][a-z0-9-]{0,31}', task_id):
raise ValueError('任务ID仅接受小写字母、数字和短横线。')
repo, parent = Path(repo).resolve(), Path(parent).resolve()
if parent == repo or parent.is_relative_to(repo):
raise ValueError('工作树容器必须在主目录外。')
path = parent / task_id
if path.is_symlink() or path.resolve().parent != parent:
raise ValueError('工作树路径不允许符号链接。')
return path, 'codex/' + task_id
def checked_workspace(repo, parent, record, task_id):
path, branch = workspace_target(repo, parent, task_id)
if record['path'] != str(path) or record['branch'] != branch or record['status'] == 'removed':
raise ValueError('绑定路径或生命周期不符。')
if git(path, 'rev-parse', '--show-toplevel') != str(path):
raise ValueError('目录不再是登记的工作树。')
common = lambda cwd: (Path(cwd) / git(cwd, 'rev-parse', '--git-common-dir')).resolve()
if common(path) != common(repo) or git(path, 'branch', '--show-current') != branch:
raise ValueError('仓库归属或当前分支已改变。')
return path
def create_workspace(repo, parent, directory, bindings, task_id, lock):
"""先创建真实工作树，再登记绑定；不认领、不完成任务。"""
with lock:
path, branch = workspace_target(repo, parent, task_id)
task = load_board(directory)['tasks'][task_id]
if task['status'] != 'pending' or task['owner'] is not None:
raise ValueError('只为尚未认领的任务准备目录。')
if task_id in bindings or len(bindings) >= 4:
raise ValueError('已有绑定或达到四项目录额度。')
base = git(repo, 'rev-parse', 'HEAD')
git(repo, 'worktree', 'add', '-b', branch, str(path), base)
record = {'path': str(path), 'branch': branch, 'base': base,
'status': 'ready', 'active': False}
bindings[task_id] = record
return deepcopy(record)
def workspace_for_task(repo, parent, directory, bindings, task_id, actor, lock):
"""核对最新owner，预留目录给这一轮；工具在锁外用返回路径。"""
with lock:
task = load_board(directory)['tasks'][task_id]
if task['status'] != 'in_progress' or task['owner'] != actor:
raise ValueError('当前成员没有持有这项任务。')
record = bindings[task_id]
if record['active']:
raise ValueError('这项目录已经有活跃轮次。')
path = checked_workspace(repo, parent, record, task_id)
record['active'] = True
return str(path)
def retire_workspace(repo, parent, bindings, task_id, action, lock):
"""先停工作再收尾；默认保留，清理不强删、不删分支、不改任务。"""
if action not in ('keep', 'remove'):
raise ValueError('收尾动作只接受keep或remove。')
with lock:
record = bindings[task_id]
path = checked_workspace(repo, parent, record, task_id)
if record['active']:
return {'status': 'busy'}
if action == 'keep':
record['status'] = 'kept'
return {'status': 'kept', 'branch': record['branch'], 'path': str(path)}
dirty = git(path, 'status', '--porcelain', '--untracked-files=all', '--ignored=matching')
head = git(path, 'rev-parse', 'HEAD')
if dirty or head != record['base']:
return {'status': 'needs_review', 'dirty': bool(dirty), 'changed_head': head != record['base']}
git(repo, 'worktree', 'remove', str(path))
record['status'] = 'removed'
return {'status': 'removed', 'branch': record['branch']}
scenario='owned'
repo=Path('/teaching-s18/repo');parent=Path('/teaching-s18/worktrees');directory='board';lock=RLock();task_id='alpha';actor='reader'
base='a'*40;bindings={};commands=[];present={};changed=False;dirty_text='';fail_add=False;fail_remove=False
board={'tasks':{'alpha':{'id':'alpha','status':'pending','owner':None}}}
def load_board(_):return deepcopy(board)
def git(cwd,*args):
key=str(cwd);commands.append(list(args))
if args[:2]==('worktree','add'):
if fail_add:raise ValueError('教学Git创建失败')
present[args[4]]=args[3];return ''
if args[:2]==('worktree','remove'):
if fail_remove:raise ValueError('教学Git删除失败')
present.pop(args[2]);return ''
if args==('rev-parse','HEAD'):return ('b'*40 if changed and key!=str(repo) else base)
if args==('rev-parse','--show-toplevel'):
if key not in present:raise ValueError('工作树不在登记中')
return key
if args==('rev-parse','--git-common-dir'):return str(repo/'.git')
if args==('branch','--show-current'):return 'unexpected' if scenario=='branch_changed' else present[key]
if args[0]=='status':return dirty_text
raise AssertionError('未知替身命令：'+str(args))
if 'workspace_for_task'!='create_workspace' or scenario=='duplicate':
create_workspace(repo,parent,directory,bindings,task_id,lock)
if 'workspace_for_task'=='create_workspace':
if scenario=='claimed':board['tasks']['alpha'].update(status='in_progress',owner='reader')
if scenario=='invalid':task_id='../outside'
if scenario=='git_error':fail_add=True
if scenario=='limit':bindings.update({str(i):{} for i in range(4)})
if 'workspace_for_task'=='workspace_for_task':
if scenario!='pending':board['tasks']['alpha'].update(status='in_progress',owner='reviewer' if scenario=='wrong_owner' else 'reader')
if scenario=='busy':bindings['alpha']['active']=True
if scenario=='removed':bindings['alpha']['status']='removed'
if 'workspace_for_task'=='retire_workspace':
if scenario=='dirty':dirty_text=' M main.py'
if scenario=='ignored':dirty_text='!! local-cache'
if scenario=='committed':changed=True
if scenario=='busy':bindings['alpha']['active']=True
if scenario=='remove_error':fail_remove=True
commands.clear()
action='keep' if scenario=='keep' else 'remove'
# 调用入口
try:
observed=workspace_for_task(repo,parent,directory,bindings,task_id,actor,lock)
except (ValueError,KeyError) as error:
observed={"error":str(error)}
print(observed)
- 核对最新owner和Git归属。
- 预留active，工具锁外使用显式cwd。
选择“分支被切换”，路径字符串即使没变，也会停下来。选择“其他负责人”，在拿到目录之前就拒绝。这里的actor来自宿主线程，不是让模型自己填一个负责人字符串。
在 record['active'] = True 行停下，预留还未发生；单步后active变为True。它把“检查目录可用”和“预留给这一轮”放在同一把锁里，收尾不能插到二者之间。已有活跃轮次会被拒绝。
函数返回后，调用者在锁外运行模型和工具，finally释放active。路径绑定发生在工具装配时：
python
# user_code/practice.py 中的实际交接，省略工具注册细节
path = Path(workspace_for_task(
repo, parent, directory, bindings, task_id, actor, lock
))
try:
# 文件工具闭包捕获 path；旧 run_main 内部使用 subprocess.cwd=path
receipt = run_teammate(team, actor, turn, lock)
finally:
with lock:
bindings[task_id]["active"] = False闭包绑定绝对路径，不调用os.chdir。两条线程可以同时持有不同目录，彼此不会改进程级当前目录。前置Hook在实际工具调用前仍核对任务owner和Git归属；任务变化不会因为“刚才拿过路径”而被忽略。
本例新增write_variant，只允许将当前任务登记的固定版本写入main.py，拒绝该文件符号链接。它通过旧execute_hooked的guard扩展点接入单独的写入许可；run_main仍走原运行许可。两者都要求本次宿主允许，未知工具仍默认拒绝，模型不能传任意脚本或目标路径。
retire_workspace：文件干净，还要再看提交
收尾在两个工作线程join之后执行。函数接受keep或remove，不从任务completed推断目录可以删除，也不修改任务状态。
active仍为True时返回busy；keep保留目录与分支。remove才读取实际文件状态，覆盖跟踪、未跟踪和被忽略文件，再读取HEAD与创建起点比较。
真实执行轨迹回放；原样函数与锁，Git和看板读取使用可展开的内存替身；不创建真实工作树或启动程序。
Pys18/core/runtime.py · retire_workspace
未运行
def retire_workspace(repo, parent, bindings, task_id, action, lock):
"""先停工作再收尾；默认保留，清理不强删、不删分支、不改任务。"""
if action not in ('keep', 'remove'):
raise ValueError('收尾动作只接受keep或remove。')
with lock:
record = bindings[task_id]
path = checked_workspace(repo, parent, record, task_id)
if record['active']:
return {'status': 'busy'}
if action == 'keep':
record['status'] = 'kept'
return {'status': 'kept', 'branch': record['branch'], 'path': str(path)}
dirty = git(path, 'status', '--porcelain', '--untracked-files=all', '--ignored=matching')
head = git(path, 'rev-parse', 'HEAD')
if dirty or head != record['base']:
return {'status': 'needs_review', 'dirty': bool(dirty), 'changed_head': head != record['base']}
git(repo, 'worktree', 'remove', str(path))
record['status'] = 'removed'
return {'status': 'removed', 'branch': record['branch']}输出
点击行号设置断点，再点“调试”；也可以直接运行。
教学准备与调用入口
上方源码保持原样。以下准备输入和调用入口，在单独文件名中执行；断点只回放上方源码。
from copy import deepcopy
from pathlib import Path
from threading import RLock
import re
def workspace_target(repo, parent, task_id):
if not isinstance(task_id, str) or not re.fullmatch(r'[a-z][a-z0-9-]{0,31}', task_id):
raise ValueError('任务ID仅接受小写字母、数字和短横线。')
repo, parent = Path(repo).resolve(), Path(parent).resolve()
if parent == repo or parent.is_relative_to(repo):
raise ValueError('工作树容器必须在主目录外。')
path = parent / task_id
if path.is_symlink() or path.resolve().parent != parent:
raise ValueError('工作树路径不允许符号链接。')
return path, 'codex/' + task_id
def checked_workspace(repo, parent, record, task_id):
path, branch = workspace_target(repo, parent, task_id)
if record['path'] != str(path) or record['branch'] != branch or record['status'] == 'removed':
raise ValueError('绑定路径或生命周期不符。')
if git(path, 'rev-parse', '--show-toplevel') != str(path):
raise ValueError('目录不再是登记的工作树。')
common = lambda cwd: (Path(cwd) / git(cwd, 'rev-parse', '--git-common-dir')).resolve()
if common(path) != common(repo) or git(path, 'branch', '--show-current') != branch:
raise ValueError('仓库归属或当前分支已改变。')
return path
def create_workspace(repo, parent, directory, bindings, task_id, lock):
"""先创建真实工作树，再登记绑定；不认领、不完成任务。"""
with lock:
path, branch = workspace_target(repo, parent, task_id)
task = load_board(directory)['tasks'][task_id]
if task['status'] != 'pending' or task['owner'] is not None:
raise ValueError('只为尚未认领的任务准备目录。')
if task_id in bindings or len(bindings) >= 4:
raise ValueError('已有绑定或达到四项目录额度。')
base = git(repo, 'rev-parse', 'HEAD')
git(repo, 'worktree', 'add', '-b', branch, str(path), base)
record = {'path': str(path), 'branch': branch, 'base': base,
'status': 'ready', 'active': False}
bindings[task_id] = record
return deepcopy(record)
def workspace_for_task(repo, parent, directory, bindings, task_id, actor, lock):
"""核对最新owner，预留目录给这一轮；工具在锁外用返回路径。"""
with lock:
task = load_board(directory)['tasks'][task_id]
if task['status'] != 'in_progress' or task['owner'] != actor:
raise ValueError('当前成员没有持有这项任务。')
record = bindings[task_id]
if record['active']:
raise ValueError('这项目录已经有活跃轮次。')
path = checked_workspace(repo, parent, record, task_id)
record['active'] = True
return str(path)
def retire_workspace(repo, parent, bindings, task_id, action, lock):
"""先停工作再收尾；默认保留，清理不强删、不删分支、不改任务。"""
if action not in ('keep', 'remove'):
raise ValueError('收尾动作只接受keep或remove。')
with lock:
record = bindings[task_id]
path = checked_workspace(repo, parent, record, task_id)
if record['active']:
return {'status': 'busy'}
if action == 'keep':
record['status'] = 'kept'
return {'status': 'kept', 'branch': record['branch'], 'path': str(path)}
dirty = git(path, 'status', '--porcelain', '--untracked-files=all', '--ignored=matching')
head = git(path, 'rev-parse', 'HEAD')
if dirty or head != record['base']:
return {'status': 'needs_review', 'dirty': bool(dirty), 'changed_head': head != record['base']}
git(repo, 'worktree', 'remove', str(path))
record['status'] = 'removed'
return {'status': 'removed', 'branch': record['branch']}
scenario='keep'
repo=Path('/teaching-s18/repo');parent=Path('/teaching-s18/worktrees');directory='board';lock=RLock();task_id='alpha';actor='reader'
base='a'*40;bindings={};commands=[];present={};changed=False;dirty_text='';fail_add=False;fail_remove=False
board={'tasks':{'alpha':{'id':'alpha','status':'pending','owner':None}}}
def load_board(_):return deepcopy(board)
def git(cwd,*args):
key=str(cwd);commands.append(list(args))
if args[:2]==('worktree','add'):
if fail_add:raise ValueError('教学Git创建失败')
present[args[4]]=args[3];return ''
if args[:2]==('worktree','remove'):
if fail_remove:raise ValueError('教学Git删除失败')
present.pop(args[2]);return ''
if args==('rev-parse','HEAD'):return ('b'*40 if changed and key!=str(repo) else base)
if args==('rev-parse','--show-toplevel'):
if key not in present:raise ValueError('工作树不在登记中')
return key
if args==('rev-parse','--git-common-dir'):return str(repo/'.git')
if args==('branch','--show-current'):return 'unexpected' if scenario=='branch_changed' else present[key]
if args[0]=='status':return dirty_text
raise AssertionError('未知替身命令：'+str(args))
if 'retire_workspace'!='create_workspace' or scenario=='duplicate':
create_workspace(repo,parent,directory,bindings,task_id,lock)
if 'retire_workspace'=='create_workspace':
if scenario=='claimed':board['tasks']['alpha'].update(status='in_progress',owner='reader')
if scenario=='invalid':task_id='../outside'
if scenario=='git_error':fail_add=True
if scenario=='limit':bindings.update({str(i):{} for i in range(4)})
if 'retire_workspace'=='workspace_for_task':
if scenario!='pending':board['tasks']['alpha'].update(status='in_progress',owner='reviewer' if scenario=='wrong_owner' else 'reader')
if scenario=='busy':bindings['alpha']['active']=True
if scenario=='removed':bindings['alpha']['status']='removed'
if 'retire_workspace'=='retire_workspace':
if scenario=='dirty':dirty_text=' M main.py'
if scenario=='ignored':dirty_text='!! local-cache'
if scenario=='committed':changed=True
if scenario=='busy':bindings['alpha']['active']=True
if scenario=='remove_error':fail_remove=True
commands.clear()
action='keep' if scenario=='keep' else 'remove'
# 调用入口
try:
observed=retire_workspace(repo,parent,bindings,task_id,action,lock)
except (ValueError,KeyError) as error:
observed={"error":str(error)}
print(observed)
- 活跃时不收尾。
- 有文件或提交变化则保留；干净且HEAD未变才移除，分支仍在。
选择“已有新提交”：dirty是False，但HEAD已经变化，仍返回needs_review。只检查git status会把这种成果误判为“没有东西需要保留”。选择“仅有忽略文件”，即使普通status看不见缓存，这里也会保留目录。
停在Git remove行，绑定仍未标记removed；命令成功后才更新记录。删除失败保留旧状态。本实现不提供强制丢弃，不删除分支；干净、HEAD未变的目录移除后，分支引用仍然存在。
允许路径产生两份改动，demo的remove检查都会返回needs_review，随后改为keep。默认拒绝路径没有编辑，可以移除目录；但任务仍是in_progress。这组对照说明：任务未完成可以没有文件成果，任务完成也可以仍有目录需要保留。
三个函数如何接回原来的工作流？
宿主先创建两项任务和两个目录。两位成员通过旧claim_next分别拿到任务，用workspace_for_task预留路径，然后进入run_teammate与唯一原run_loop。每轮固定产生写入、运行、报告三次逻辑模型请求，两位成员共六次。
任务能否完成由本次工具证据决定：两次tool_result都成功，运行输出符合当前任务的42或43，文件也确实等于本项固定版本，宿主才调用complete_task。模型回答“完成”不能替代这些检查。失败保留owner和in_progress，不自动转交另一成员再试。
本课聚焦已有任务进入工作目录的交接；复用原自主认领与独立成员，不再增加空闲轮询、邮箱对话或计划协议的演示路径。代码仍复用按实际工具生成的system、模型请求有界恢复、原Hook与运行许可。成员本次各交付一轮，模型循环每轮上限四次，目录收尾不会额外调用模型。
绑定表只在当前进程内。应用读改写共用一把RLock，Git自己的锁不能替代owner的原子检查；工具执行放在锁外。多个进程同时调度、成员历史恢复、生命周期的崩溃补偿，需要另外设计，不能从一次正常执行推导出来。
用真实 Git 证明边界
bash
python3 examples/s18-worktree-isolation/user_code/verify_practice.py
python3 examples/s18-worktree-isolation/user_code/verify_fixtures.py十四组标准库测试创建真实临时仓库，用Barrier确保两个成员先各持目录再进入工具工作。测试检查两份同名文件、主目录保持、并发创建只有一次成功、owner和active门控、仅未跟踪或忽略文件的保护，以及创建/删除失败不标记成功。
还有一组会把两份成果分别提交，再尝试整合：先采用alpha，随后合并beta，Git实际报告main.py冲突。目录隔离保住了两份修改，冲突仍需在审查时解决。
三个教学区的十九种输入产生176个真实函数快照。浏览器原样执行展示函数与锁，Git和看板读取使用可展开的内存替身，不创建真实worktree、不运行程序；真实Git证据来自公开入口和测试。断点只保留下一行、少量变量及当前输出，图使用同一个游标。
可选verify_sdk.py使用Anthropic SDK 1.13.0与MockTransport，允许和拒绝路径各七次模拟HTTP，包含工具反馈后的一次529。恢复当前请求没有重复写入；Git与获准进程仍真实执行，没有在线模型调用。
最后回到两份main.py：工作目录可以分开，端口、数据库和宿主权限仍可能共享。这个示例的路径检查服务于可信固定工具，不能把Git worktree当作文件系统沙箱。成果保留之后，审查哪份、如何合并，仍然是一项明确的后续决定。