Function create_agent_worktree

src/bridge/worktree.py:130–198

Create `` /.claude/worktrees/agent- /`` as a detached worktree of ``base_dir``. Returns a :class:`WorktreePaths` with ``created=True`` on success. On any failure — bad session_id, not a git repo, mkdir error, git command error, subprocess timeout — falls back to ``ba

(
    base_dir: str, session_id: str,
)

async def create_agent_worktree(
    base_dir: str, session_id: str,
) -> WorktreePaths:
    """Create ``<base>/.claude/worktrees/agent-<session_id>/`` as a
    detached worktree of ``base_dir``.

    Returns a :class:`WorktreePaths` with ``created=True`` on success.
    On any failure — bad session_id, not a git repo, mkdir error, git
    command error, subprocess timeout — falls back to ``base_dir``
    with ``created=False`` and a warning log. Never raises.
    """
    # Validate session_id before it touches the filesystem. The
    # work-secret payload type is loose (server-provided string), so
    # nothing else stops a malicious or buggy ``id`` value from
    # turning into a path-traversal write target.
    if not _SESSION_ID_RE.match(session_id):
        logger.warning(
            '[worktree] session_id %r is not allowlist-safe — '
            'spawning in base dir', session_id,
        )
        return WorktreePaths(
            base_dir=base_dir, working_dir=base_dir, created=False,
        )

    if not await _is_git_repo(base_dir):
        logger.warning(
            '[worktree] %s is not a git repo — spawning in base dir',
            base_dir,
        )
        return WorktreePaths(
            base_dir=base_dir, working_dir=base_dir, created=False,
        )

    target = os.path.join(
        base_dir, '.claude', 'worktrees', f'agent-{session_id}',
    )
    parent = os.path.dirname(target)
    try:
        os.makedirs(parent, exist_ok=True)
    except OSError as err:
        logger.warning(
            '[worktree] mkdir(%s) failed: %s — spawning in base dir',
            parent, err,
        )
        return WorktreePaths(
            base_dir=base_dir, working_dir=base_dir, created=False,
        )

    # ``--detach`` avoids polluting the branch namespace; the worktree
    # starts on whatever commit the base dir's HEAD pointed at when
    # we ran ``git worktree add`` (concurrent ``git reset`` is racy
    # but acceptable for an ephemeral agent worktree). It has no
    # branch attached, so ``--force`` removal stays clean later.
    rc, stderr = await _run_git(
        'worktree', 'add', '--detach', target, 'HEAD', cwd=base_dir,
    )
    if rc != 0:
        logger.warning(