我给编程 agent 做了一个共享 worktree 池

		
#!/usr/bin/env python3
"""A shared pool of git worktrees for coding agents (and their sub-agents).
 
Agents borrow a slot instead of creating a fresh worktree each time. Slots are
reused between tasks, so the expensive parts of a checkout are paid for once:
 
  symlink               ignored folders every slot reads but never writes
                        (vendored tools, reference material, big assets)
  clone                 ignored folders every slot gets its own copy of,
                        refreshed when the main checkout's copy changes
                        (build or import caches)
  symlink_unless_owned  symlinked by default; a task that needs to write to
                        one asks for its own copy with --own PATH
 
Copies are copy-on-write where the filesystem allows it (APFS clones on macOS,
reflinks on Btrfs/XFS on Linux) and plain copies elsewhere.
 
The pool starts at `start` slots and grows by itself up to `max`. Past `max`
it refuses and tells the agent to ask a human, then rerun with
--approved-by-user. The script does not check who gave that approval; it is a
speed bump, not an access control.
 
The config is committed with the project, in the main checkout. Leases live
in the git common dir (usually .git/), which every worktree shares and git
never tracks:
 
  worktree-pool.json              config (optional; `init` writes an example)
  .git/worktree-pool-leases.json  who holds which slot
  .git/worktree-pool.lock         flock that serialises concurrent agents
 
Every configured path must be gitignored, and the pattern must match a
symlink too: write `assets`, not `assets/` (a trailing slash matches folders only).
 
    worktree_pool.py init
    worktree_pool.py status
    worktree_pool.py acquire --owner claude --branch feat/x --task "..."
    worktree_pool.py acquire ... --own assets/raw   # this task writes to assets/raw
    worktree_pool.py release pool-b                 # branch merged into base
    worktree_pool.py release pool-b --abandon       # drop unmerged work on purpose
 
Requires Python 3.8+, git 2.23+ and a Unix-like OS (it uses fcntl).
"""
import argparse
import fcntl
import json
import os
import shutil
import string
import subprocess
import sys
import time
from contextlib import contextmanager
from pathlib import Path
 
DEFAULTS = {
    'base': None,          # None: origin's default branch, else main, else master
    'pool_dir': None,      # None: <repo>.worktrees next to the main checkout
    'start': 5,
    'max': 10,
    'stale_hours': 12,
    'symlink': [],
    'clone': [],
    'symlink_unless_owned': [],
}
 
EXAMPLE = dict(DEFAULTS, symlink=['node_modules'], clone=['.cache/build'],
               symlink_unless_owned=['assets/raw'])
 
 
def git(*args, cwd=None, check=True):
    result = subprocess.run(['git', *args], cwd=cwd, capture_output=True, text=True)
    if check and result.returncode != 0:
        raise SystemExit('git %s failed: %s' % (' '.join(args), result.stderr.strip()))
    return result.stdout.strip()
 
 
def git_ok(*args, cwd=None) -> bool:
    return subprocess.run(['git', *args], cwd=cwd, capture_output=True).returncode == 0
 
 
def main_checkout() -> Path:
    """The first entry of `git worktree list` is always the main working tree."""
    first = git('worktree', 'list', '--porcelain').splitlines()[0]
    return Path(first[len('worktree '):])
 
 
COMMON = Path(git('rev-parse', '--path-format=absolute', '--git-common-dir'))
MAIN = main_checkout()
CONFIG = MAIN / 'worktree-pool.json'
STATE = COMMON / 'worktree-pool-leases.json'
LOCK = COMMON / 'worktree-pool.lock'
 
 
def load_config(path: Path, resolve_base=True) -> dict:
    config = dict(DEFAULTS)
    if path.exists():
        config.update(json.loads(path.read_text()))
    if not config['base'] and resolve_base:
        config['base'] = default_base()
    config['pool_dir'] = (MAIN / config['pool_dir']) if config['pool_dir'] else MAIN.parent / (MAIN.name + '.worktrees')
    return config
 
 
def default_base() -> str:
    remote = git('symbolic-ref', '--short', 'refs/remotes/origin/HEAD', cwd=MAIN, check=False)
    if remote.startswith('origin/'):
        return remote[len('origin/'):]
    for name in ('main', 'master'):
        if git_ok('show-ref', '--verify', '--quiet', 'refs/heads/' + name, cwd=MAIN):
            return name
    raise SystemExit('could not guess the base branch; set "base" in %s' % CONFIG)
 
 
def check_ignored(config: dict) -> None:
    """Symlinks or copies of tracked paths would show up as changes in every slot."""
    for key in ('symlink', 'clone', 'symlink_unless_owned'):
        for rel in config[key]:
            if not git_ok('check-ignore', '-q', rel, cwd=MAIN):
                raise SystemExit('%s (from "%s") is not gitignored in %s; ignore it first.' % (rel, key, MAIN))
 
 
@contextmanager
def locked():
    with open(LOCK, 'w') as handle:
        fcntl.flock(handle, fcntl.LOCK_EX)
        try:
            yield
        finally:
            fcntl.flock(handle, fcntl.LOCK_UN)
 
 
def load() -> dict:
    if STATE.exists():
        return json.loads(STATE.read_text())
    return {'created': [], 'slots': {}, 'clones': {}}
 
 
def save(state: dict) -> None:
    tmp = STATE.with_suffix('.tmp')
    tmp.write_text(json.dumps(state, ensure_ascii=False, indent=2) + '\n')
    tmp.replace(STATE)
 
 
def slot_names(count: int) -> list:
    return ['pool-' + string.ascii_lowercase[i] for i in range(count)]
 
 
def registered_worktrees() -> set:
    paths = set()
    for line in git('worktree', 'list', '--porcelain').splitlines():
        if line.startswith('worktree '):
            paths.add(str(Path(line[len('worktree '):]).resolve()))
    return paths
 
 
def link(path: Path, target: Path) -> None:
    """Points path at target, replacing a stale link or an old copy."""
    if path.is_symlink():
        if Path(os.readlink(path)) == target:
            return
        path.unlink()
    elif path.is_dir():
        shutil.rmtree(path)
    elif path.exists():
        path.unlink()
    path.parent.mkdir(parents=True, exist_ok=True)
    path.symlink_to(target)
 
 
def clone(source: Path, dest: Path) -> None:
    """Copy-on-write copy where the filesystem supports it, a plain copy otherwise."""
    if dest.is_symlink() or dest.is_file():
        dest.unlink()
    elif dest.exists():
        shutil.rmtree(dest)
    dest.parent.mkdir(parents=True, exist_ok=True)
    flag = '-c' if sys.platform == 'darwin' else '--reflink=auto'
    if subprocess.run(['cp', flag, '-R', str(source), str(dest)], capture_output=True).returncode == 0:
        return
    if dest.exists():
        shutil.rmtree(dest)
    shutil.copytree(source, dest, symlinks=True)
 
 
def fingerprint(folder: Path) -> float:
    """Cheap change marker: newest mtime of the folder and its direct children."""
    times = [folder.stat().st_mtime]
    times += [child.lstat().st_mtime for child in folder.iterdir()]
    return max(times)
 
 
def prepare(config: dict, state: dict, name: str, owned: list) -> None:
    """Makes the slot's ignored folders match the config."""
    path = config['pool_dir'] / name
    stamps = state.setdefault('clones', {}).setdefault(name, {})
    for rel in config['symlink'] + [r for r in config['symlink_unless_owned'] if r not in owned]:
        if (MAIN / rel).exists():
            link(path / rel, MAIN / rel)
            stamps.pop(rel, None)
    for rel in config['clone'] + owned:
        source = MAIN / rel
        if not source.exists():
            continue
        dest = path / rel
        current = fingerprint(source)
        if dest.is_symlink() or not dest.exists() or stamps.get(rel, -1.0) < current:
            clone(source, dest)
            stamps[rel] = current
 
 
def dirty(path: Path) -> list:
    out = git('status', '--porcelain', cwd=path)
    return [line for line in out.splitlines() if line.strip()]
 
 
def cmd_init(args, _config) -> None:
    if args.config.exists():
        raise SystemExit('%s already exists.' % args.config)
    args.config.write_text(json.dumps(EXAMPLE, indent=2) + '\n')
    print('wrote %s; edit the folder lists to match your project, then commit it.' % args.config)
 
 
def cmd_status(_args, config) -> None:
    state = load()
    created = state.get('created', [])
    now = time.time()
    for name in slot_names(max(config['start'], len(created))):
        lease = state['slots'].get(name)
        if not lease:
            print('%-7s free%s' % (name, '' if name in created else ' (created on first use)'))
            continue
        hours = (now - lease['since']) / 3600
        stale = '  STALE?' if hours > config['stale_hours'] else ''
        print('%-7s %-14s %-24s %5.1fh  %s%s' % (name, lease['owner'], lease['branch'], hours, lease.get('task', ''), stale))
    print('pool: %d created, %d leased; grows by itself up to %d, more needs a human' % (len(created), len(state['slots']), config['max']))
 
 
def cmd_acquire(args, config) -> None:
    owned = [str(Path(rel)) for rel in args.own]
    for rel in owned:
        if rel not in config['symlink_unless_owned']:
            raise SystemExit('--own %s: only paths listed in "symlink_unless_owned" can be owned.' % rel)
    check_ignored(config)
    base = args.base or config['base']
    with locked():
        state = load()
        created = state.setdefault('created', [])
        free = [name for name in created if name not in state['slots']]
        if free:
            chosen = free[0]
        else:
            if len(created) >= config['max'] and not args.approved_by_user:
                cmd_status(args, config)
                raise SystemExit('pool is full (%d slots, all leased). Growing past %d needs a human: ask them, then rerun with --approved-by-user.' % (len(created), config['max']))
            if len(created) >= len(string.ascii_lowercase):
                raise SystemExit('pool is at %d slots, the most this script names.' % len(created))
            chosen = slot_names(len(created) + 1)[-1]
        path = config['pool_dir'] / chosen
        if str(path.resolve()) not in registered_worktrees():
            if path.exists():
                raise SystemExit('%s exists but is not a registered worktree; inspect it before reusing.' % path)
            config['pool_dir'].mkdir(parents=True, exist_ok=True)
            git('worktree', 'add', '--detach', str(path), config['base'], cwd=MAIN)
        if chosen not in created:
            created.append(chosen)
            save(state)
        leftover = dirty(path)
        if leftover:
            raise SystemExit('%s has uncommitted changes from an earlier lease:\n%s' % (chosen, '\n'.join(leftover[:20])))
        if git_ok('show-ref', '--verify', '--quiet', 'refs/heads/' + args.branch, cwd=MAIN):
            git('switch', args.branch, cwd=path)
        else:
            git('switch', '-c', args.branch, base, cwd=path)
        prepare(config, state, chosen, owned)
        state['slots'][chosen] = {'owner': args.owner, 'branch': args.branch, 'base': base,
                                  'task': args.task, 'since': time.time(), 'owned': owned}
        save(state)
    unignored = dirty(path)
    if unignored:
        print('warning: git sees the pool\'s links as changes in %s:\n%s\n'
              'A pattern with a trailing slash ("assets/") matches folders, not symlinks; drop the slash.'
              % (chosen, '\n'.join(unignored[:20])), file=sys.stderr)
    print(path)
 
 
def resolve_slot(text: str, state: dict) -> str:
    name = Path(text).name
    if name not in state['slots']:
        raise SystemExit('%s is not leased (see status).' % name)
    return name
 
 
def cmd_release(args, config) -> None:
    with locked():
        state = load()
        name = resolve_slot(args.slot, state)
        lease = state['slots'][name]
        base = lease.get('base', config['base'])
        path = config['pool_dir'] / name
        if path.exists():
            leftover = dirty(path)
            if leftover and not args.abandon:
                raise SystemExit('%s has uncommitted changes; commit them, or rerun with --abandon to drop them:\n%s' % (name, '\n'.join(leftover[:20])))
            merged = git_ok('merge-base', '--is-ancestor', lease['branch'], base, cwd=MAIN)
            if not merged and not args.abandon:
                raise SystemExit('%s is not merged into %s; merge it first, or rerun with --abandon (the branch and its commits stay).' % (lease['branch'], base))
            if leftover:
                git('reset', '--hard', cwd=path)
                git('clean', '-fd', cwd=path)
            git('switch', '--detach', config['base'], cwd=path)
            prepare(config, state, name, [])
        del state['slots'][name]
        save(state)
    print('released %s (branch %s kept)' % (name, lease['branch']))
 
 
def main() -> None:
    parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
    parser.add_argument('--config', type=Path, default=CONFIG, help='config file (default: %(default)s)')
    sub = parser.add_subparsers(dest='command', required=True)
    sub.add_parser('init', help='write an example config')
    sub.add_parser('status', help='show slots and leases')
    acquire = sub.add_parser('acquire', help='lease a slot and switch it to a branch')
    acquire.add_argument('--owner', required=True, help='agent name, e.g. codex-1')
    acquire.add_argument('--branch', required=True)
    acquire.add_argument('--base', help='branch to start a new branch from (default: config base)')
    acquire.add_argument('--task', default='', help='one line saying what the lease is for')
    acquire.add_argument('--own', action='append', default=[], metavar='PATH',
                         help='copy this "symlink_unless_owned" path instead of linking it; repeatable')
    acquire.add_argument('--approved-by-user', action='store_true',
                         help='grow past the max; only after a human said yes (not verified)')
    release = sub.add_parser('release', help='return a slot to the pool')
    release.add_argument('slot', help='pool-<letter> or its path')
    release.add_argument('--abandon', action='store_true', help='release although unmerged or dirty (drops uncommitted changes)')
    args = parser.parse_args()
    config = load_config(args.config, resolve_base=args.command != 'init')
    {'init': cmd_init, 'status': cmd_status, 'acquire': cmd_acquire, 'release': cmd_release}[args.command](args, config)
 
 
if __name__ == '__main__':
    main()
 
	

Expand