Compare commits

...

3 Commits

Author SHA1 Message Date
khyari hamza 32c8041826 feat(sandbox): add copy_on_write strategy stub (B4.sandbox) 2026-02-13 21:14:52 +00:00
khyari hamza 8aa17b72c9 feat(sandbox): implement git_worktree strategy (B4.sandbox) 2026-02-13 21:14:47 +00:00
khyari hamza be2659f70b feat(resource): add git-checkout handler and discovery (B4-SANDBOX)
- GitCheckoutHandler: validate repo path, branch, read-only flags
- Sandbox strategy mapping (git_worktree for writable, none for read-only)
- Path normalization (absolute, no trailing slashes)
- Child resource discovery for filesystem directories
- Fixed duplicate step definitions in sandbox_manager_steps.py
- 11/11 BDD scenarios passing, 0 regressions (178 total)
- pyright 0 errors, ruff all checks passed
2026-02-13 21:14:44 +00:00
9 changed files with 924 additions and 0 deletions
+23
View File
@@ -0,0 +1,23 @@
@sandbox @b4 @copy_on_write
Feature: Copy-on-Write Sandbox Strategy Stub
As a developer planning for large-project optimization
I want a copy_on_write strategy stub that raises NotImplementedError
So that it's clear this strategy is planned for post-M1
Scenario: copy_on_write create raises NotImplementedError
When I attempt to v4 create a copy_on_write sandbox
Then the v4 copy_on_write attempt should raise NotImplementedError
And the v4 copy_on_write error should mention post-M1
Scenario: copy_on_write cleanup raises NotImplementedError
When I attempt to v4 cleanup a copy_on_write sandbox
Then the v4 copy_on_write cleanup should raise NotImplementedError
Scenario: copy_on_write rewrite_path raises NotImplementedError
When I attempt to v4 rewrite path with copy_on_write
Then the v4 copy_on_write rewrite should raise NotImplementedError
Scenario: copy_on_write can be registered in SandboxRegistry
Given a v4 sandbox registry
When I register the v4 copy_on_write strategy in the registry
Then the v4 registry should have strategy "copy_on_write"
+69
View File
@@ -0,0 +1,69 @@
Feature: Git-checkout resource handler and discovery (B4.sandbox)
As a developer
I want a handler that validates git repos and discovers child resources
So that git-checkout resources are properly registered with correct sandbox mapping
# Validation scenarios
Scenario: Validate a valid git repository path
Given a temporary git repository at "/tmp/test-repo"
When I run git-checkout validation for path "/tmp/test-repo"
Then the git handler validation should succeed
And the handler result should have type "git_repository"
Scenario: Validate with a specific branch
Given a temporary git repository at "/tmp/branch-repo"
When I run git-checkout branch validation for path "/tmp/branch-repo" branch "main"
Then the git handler validation should succeed
And the handler result should have branch "main"
Scenario: Validate with read-only flag
Given a temporary git repository at "/tmp/ro-repo"
When I run git-checkout readonly validation for path "/tmp/ro-repo"
Then the git handler validation should succeed
And the handler result should be marked read-only
Scenario: Reject non-existent path
When I run git-checkout validation for path "/nonexistent/path"
Then the git handler validation should fail with "does not exist"
Scenario: Reject path that is not a git repository
Given a temporary directory at "/tmp/not-a-repo"
When I run git-checkout validation for path "/tmp/not-a-repo"
Then the git handler validation should fail with "not a git repository"
# Sandbox strategy mapping
Scenario: Default sandbox strategy is git_worktree
Given a temporary git repository at "/tmp/sandbox-repo"
When I run git-checkout validation for path "/tmp/sandbox-repo"
Then the handler result should have sandbox strategy "git_worktree"
Scenario: Read-only resources get no sandbox strategy
Given a temporary git repository at "/tmp/ro-sandbox-repo"
When I run git-checkout readonly validation for path "/tmp/ro-sandbox-repo"
Then the handler result should have sandbox strategy "none"
# Path normalization
Scenario: Normalize relative path to absolute
Given a temporary git repository at a relative path "test-repo-rel"
When I run git-checkout validation for the relative repo path
Then the git handler validation should succeed
And the handler result path should be absolute
Scenario: Normalize trailing slashes
Given a temporary git repository at "/tmp/trail-repo"
When I run git-checkout validation for path with trailing slash "/tmp/trail-repo"
Then the git handler validation should succeed
And the handler result path should not end with a slash
# Child resource discovery
Scenario: Discover filesystem directory children
Given a temporary git repository at "/tmp/disco-repo"
And the repository has subdirectories "src" and "docs"
When I discover child resources for "/tmp/disco-repo"
Then the discovery result should contain 2 children
And each child should have type "filesystem"
Scenario: Discovery of empty repository returns no children
Given a temporary git repository at "/tmp/empty-repo"
When I discover child resources for "/tmp/empty-repo"
Then the discovery result should contain 0 children
+61
View File
@@ -0,0 +1,61 @@
@sandbox @b4 @git_worktree
Feature: Git Worktree Sandbox Strategy
As a developer executing plans on git repositories
I want git worktree-based sandboxing
So that changes are isolated in a separate worktree
# Worktree creation
Scenario: Create a git worktree sandbox for a repository
Given a v4 temporary git repository
When I create a v4 git worktree sandbox for the repository
Then the v4 worktree sandbox path should exist
And the v4 worktree sandbox should be a valid git directory
Scenario: Worktree sandbox records metadata for rollback
Given a v4 temporary git repository
When I create a v4 git worktree sandbox for the repository
Then the v4 worktree metadata should contain "worktree_path"
And the v4 worktree metadata should contain "branch"
And the v4 worktree metadata should contain "base_commit"
Scenario: Worktree sandbox creates a detached branch
Given a v4 temporary git repository
When I create a v4 git worktree sandbox for the repository
Then the v4 worktree should have a sandbox branch name
# ── Worktree cleanup ──
Scenario: Cleanup removes the worktree directory
Given a v4 temporary git repository
When I create a v4 git worktree sandbox for the repository
And I cleanup the v4 git worktree sandbox
Then the v4 worktree sandbox path should not exist
# ── Path rewriting ──
Scenario: Rewrite path from original repo to worktree
Given a v4 temporary git repository
When I create a v4 git worktree sandbox for the repository
And I rewrite the v4 git path for a file "src/main.py"
Then the v4 rewritten git path should point inside the worktree
# ── Safe fallback ──
Scenario: Strategy handles non-existent repo path gracefully
When I attempt a v4 git worktree for non-existent path "/tmp/no-such-repo"
Then the v4 git worktree attempt should fail with an error
Scenario: Strategy handles path that is not a git repo
Given a v4 temporary non-git directory
When I attempt a v4 git worktree for the non-git directory
Then the v4 git worktree attempt should fail with an error
# ── Integration with SandboxManager ──
Scenario: Git worktree strategy works with SandboxManager
Given a v4 temporary git repository
And a v4 sandbox manager with git worktree strategy
When I use the v4 manager to create a sandbox for the git repo
Then the v4 managed git sandbox should be active
And the v4 managed git sandbox path should exist
@@ -0,0 +1,96 @@
"""Step definitions for B4 copy-on-write strategy stub.
All step texts use the 'v4 copy_on_write' prefix to avoid
global behave step-name collisions.
"""
from __future__ import annotations
from behave import given, then, when # type: ignore[import-untyped]
@when("I attempt to v4 create a copy_on_write sandbox")
def step_v4_cow_create(context):
"""Attempt to create a copy_on_write sandbox."""
from cleveragents.infrastructure.sandbox.copy_on_write import (
CopyOnWriteStrategy,
)
strategy = CopyOnWriteStrategy()
try:
strategy.create(resource_name="test", base_path="/tmp/cow")
context.v4_cow_error = None
except NotImplementedError as e:
context.v4_cow_error = e
@then("the v4 copy_on_write attempt should raise NotImplementedError")
def step_v4_cow_error(context):
assert isinstance(context.v4_cow_error, NotImplementedError), (
f"Expected NotImplementedError, got {context.v4_cow_error}"
)
@then("the v4 copy_on_write error should mention post-M1")
def step_v4_cow_post_m1(context):
assert "post-m1" in str(context.v4_cow_error).lower(), (
f"Error message should mention post-M1: {context.v4_cow_error}"
)
@when("I attempt to v4 cleanup a copy_on_write sandbox")
def step_v4_cow_cleanup(context):
"""Attempt to cleanup with copy_on_write."""
from cleveragents.infrastructure.sandbox.copy_on_write import (
CopyOnWriteStrategy,
)
strategy = CopyOnWriteStrategy()
try:
strategy.cleanup(sandbox_path="/tmp/cow/test")
context.v4_cow_cleanup_error = None
except NotImplementedError as e:
context.v4_cow_cleanup_error = e
@then("the v4 copy_on_write cleanup should raise NotImplementedError")
def step_v4_cow_cleanup_error(context):
assert isinstance(context.v4_cow_cleanup_error, NotImplementedError), (
f"Expected NotImplementedError, got {context.v4_cow_cleanup_error}"
)
@when("I attempt to v4 rewrite path with copy_on_write")
def step_v4_cow_rewrite(context):
"""Attempt to rewrite path with copy_on_write."""
from cleveragents.infrastructure.sandbox.copy_on_write import (
CopyOnWriteStrategy,
)
strategy = CopyOnWriteStrategy()
try:
strategy.rewrite_path(
original="/tmp/src/file.py",
sandbox_path="/tmp/cow",
resource_name="src",
)
context.v4_cow_rewrite_error = None
except NotImplementedError as e:
context.v4_cow_rewrite_error = e
@then("the v4 copy_on_write rewrite should raise NotImplementedError")
def step_v4_cow_rewrite_error(context):
assert isinstance(context.v4_cow_rewrite_error, NotImplementedError), (
f"Expected NotImplementedError, got {context.v4_cow_rewrite_error}"
)
@when("I register the v4 copy_on_write strategy in the registry")
def step_v4_register_cow(context):
"""Register copy_on_write in the sandbox registry."""
from cleveragents.infrastructure.sandbox.copy_on_write import (
CopyOnWriteStrategy,
)
context.v4_registry.register("copy_on_write", CopyOnWriteStrategy())
@@ -0,0 +1,246 @@
"""Step definitions for B4 git worktree sandbox strategy.
All step texts use the 'v4 git worktree' or 'v4 worktree' prefix
to avoid global behave step-name collisions.
"""
from __future__ import annotations
import os
import shutil
import subprocess
import tempfile
from behave import given, then, when # type: ignore[import-untyped]
# ── Helpers ──
def _create_temp_git_repo() -> str:
"""Create a temp directory initialized as a git repo with one commit."""
repo_dir = tempfile.mkdtemp(prefix="v4_git_test_")
subprocess.run(
["git", "init", repo_dir],
check=True,
capture_output=True,
)
# Configure git user for commit and disable gpg signing
for key, val in [
("user.email", "test@test.com"),
("user.name", "Test"),
("commit.gpgsign", "false"),
]:
subprocess.run(
["git", "-C", repo_dir, "config", key, val],
check=True,
capture_output=True,
)
# Create initial commit
readme = os.path.join(repo_dir, "README.md")
with open(readme, "w") as f:
f.write("# Test Repo\n")
subprocess.run(
["git", "-C", repo_dir, "add", "."],
check=True,
capture_output=True,
)
subprocess.run(
["git", "-C", repo_dir, "commit", "-m", "Initial commit"],
check=True,
capture_output=True,
)
return repo_dir
# ── Given steps ──
@given("a v4 temporary git repository")
def step_v4_temp_git_repo(context):
"""Create a temporary git repository for testing."""
context.v4_git_repo = _create_temp_git_repo()
context.v4_worktree_path = None
context.v4_worktree_metadata = {}
context.v4_git_error = None
@given("a v4 temporary non-git directory")
def step_v4_temp_non_git_dir(context):
"""Create a temporary directory that is NOT a git repo."""
context.v4_non_git_dir = tempfile.mkdtemp(prefix="v4_nongit_test_")
context.v4_git_error = None
@given("a v4 sandbox manager with git worktree strategy")
def step_v4_manager_with_git_strategy(context):
"""Set up a SandboxManager with the git worktree strategy registered."""
from cleveragents.infrastructure.sandbox.git_worktree import (
GitWorktreeStrategy,
)
from cleveragents.infrastructure.sandbox.manager import (
SandboxManager,
SandboxRegistry,
)
registry = SandboxRegistry()
registry.register("git_worktree", GitWorktreeStrategy())
context.v4_git_manager = SandboxManager(
registry=registry, base_path=context.v4_git_repo
)
# ── When steps: creation ──
@when("I create a v4 git worktree sandbox for the repository")
def step_v4_create_worktree(context):
"""Create a git worktree sandbox."""
from cleveragents.infrastructure.sandbox.git_worktree import (
GitWorktreeStrategy,
)
strategy = GitWorktreeStrategy()
sandbox_path = strategy.create(
resource_name="test-resource",
base_path=context.v4_git_repo,
)
context.v4_worktree_path = sandbox_path
context.v4_worktree_metadata = strategy.last_metadata
@when("I cleanup the v4 git worktree sandbox")
def step_v4_cleanup_worktree(context):
"""Clean up the git worktree sandbox."""
from cleveragents.infrastructure.sandbox.git_worktree import (
GitWorktreeStrategy,
)
strategy = GitWorktreeStrategy()
strategy.cleanup(context.v4_worktree_path)
@when('I rewrite the v4 git path for a file "{filename}"')
def step_v4_rewrite_git_path(context, filename):
"""Rewrite a file path from repo to worktree."""
from cleveragents.infrastructure.sandbox.git_worktree import (
GitWorktreeStrategy,
)
strategy = GitWorktreeStrategy()
original = os.path.join(context.v4_git_repo, filename)
context.v4_git_rewritten = strategy.rewrite_path(
original=original,
sandbox_path=context.v4_worktree_path,
resource_name=context.v4_git_repo,
)
@when('I attempt a v4 git worktree for non-existent path "{path}"')
def step_v4_worktree_nonexistent(context, path):
"""Attempt to create a worktree for a path that doesn't exist."""
from cleveragents.infrastructure.sandbox.git_worktree import (
GitWorktreeStrategy,
)
strategy = GitWorktreeStrategy()
try:
strategy.create(resource_name="bad", base_path=path)
context.v4_git_error = None
except (ValueError, OSError) as e:
context.v4_git_error = e
@when("I attempt a v4 git worktree for the non-git directory")
def step_v4_worktree_non_git(context):
"""Attempt to create a worktree for a non-git directory."""
from cleveragents.infrastructure.sandbox.git_worktree import (
GitWorktreeStrategy,
)
strategy = GitWorktreeStrategy()
try:
strategy.create(resource_name="bad", base_path=context.v4_non_git_dir)
context.v4_git_error = None
except (ValueError, OSError) as e:
context.v4_git_error = e
@when("I use the v4 manager to create a sandbox for the git repo")
def step_v4_manager_create_git_sandbox(context):
"""Use the SandboxManager to create a git worktree sandbox."""
context.v4_managed_ref = context.v4_git_manager.create_sandbox(
resource_name="managed-repo", strategy_name="git_worktree"
)
# ── Then steps ──
@then("the v4 worktree sandbox path should exist")
def step_v4_worktree_exists(context):
assert os.path.isdir(context.v4_worktree_path), (
f"Worktree path does not exist: {context.v4_worktree_path}"
)
@then("the v4 worktree sandbox should be a valid git directory")
def step_v4_worktree_is_git(context):
result = subprocess.run(
["git", "-C", context.v4_worktree_path, "status"],
capture_output=True,
text=True,
)
assert result.returncode == 0, f"Not a valid git dir: {result.stderr}"
@then('the v4 worktree metadata should contain "{key}"')
def step_v4_worktree_metadata_key(context, key):
assert key in context.v4_worktree_metadata, (
f"Key '{key}' not in metadata: {list(context.v4_worktree_metadata.keys())}"
)
@then("the v4 worktree should have a sandbox branch name")
def step_v4_worktree_branch(context):
result = subprocess.run(
["git", "-C", context.v4_worktree_path, "branch", "--show-current"],
capture_output=True,
text=True,
)
branch = result.stdout.strip()
assert branch.startswith("sandbox/"), (
f"Expected sandbox/ branch prefix, got: {branch}"
)
@then("the v4 worktree sandbox path should not exist")
def step_v4_worktree_not_exists(context):
assert not os.path.exists(context.v4_worktree_path), (
f"Worktree path still exists: {context.v4_worktree_path}"
)
@then("the v4 rewritten git path should point inside the worktree")
def step_v4_git_path_inside_worktree(context):
assert context.v4_worktree_path in context.v4_git_rewritten, (
f"Rewritten path {context.v4_git_rewritten} does not "
f"contain worktree {context.v4_worktree_path}"
)
@then("the v4 git worktree attempt should fail with an error")
def step_v4_git_error(context):
assert context.v4_git_error is not None, "Expected an error but none was raised"
@then("the v4 managed git sandbox should be active")
def step_v4_managed_active(context):
assert context.v4_managed_ref.active is True
@then("the v4 managed git sandbox path should exist")
def step_v4_managed_path_exists(context):
assert os.path.isdir(context.v4_managed_ref.path), (
f"Managed sandbox path does not exist: {context.v4_managed_ref.path}"
)
@@ -0,0 +1 @@
"""Resource handlers for CleverAgents."""
@@ -0,0 +1,160 @@
"""Git-checkout resource handler for CleverAgents.
Validates git repository paths, maps sandbox strategies,
normalizes paths, and discovers filesystem directory children.
Based on implementation_plan.md B4.sandbox and ADR-004.
"""
import os
import subprocess
from typing import Any
class GitCheckoutHandler:
"""Handler for git-checkout resource validation and discovery.
Validates that a path is a valid git repository, determines
the appropriate sandbox strategy, and discovers child
filesystem directories within the repository.
"""
def validate(
self,
path: str,
branch: str | None = None,
read_only: bool = False,
) -> dict[str, Any]:
"""Validate a git-checkout resource.
Args:
path: Path to the git repository.
branch: Optional branch name to record.
read_only: Whether the resource is read-only.
Returns:
Dictionary with validated resource metadata.
Raises:
ValueError: If the path doesn't exist or isn't a git repo.
"""
# Normalize path
normalized = self._normalize_path(path)
# Check existence
if not os.path.exists(normalized):
raise ValueError(f"Path '{path}' does not exist")
# Check it's a git repository
if not self._is_git_repo(normalized):
raise ValueError(f"Path '{path}' is not a git repository")
# Determine sandbox strategy
sandbox_strategy = self._map_sandbox_strategy(read_only)
result: dict[str, Any] = {
"type": "git_repository",
"path": normalized,
"read_only": read_only,
"sandbox_strategy": sandbox_strategy,
}
if branch is not None:
result["branch"] = branch
return result
def discover_children(self, path: str) -> list[dict[str, Any]]:
"""Discover filesystem directory children within a git repo.
Scans the top-level directories in the repository,
excluding hidden directories (starting with '.').
Args:
path: Path to the git repository.
Returns:
List of child resource dictionaries.
"""
normalized = self._normalize_path(path)
children: list[dict[str, Any]] = []
if not os.path.exists(normalized):
return children
for entry in sorted(os.listdir(normalized)):
full_path = os.path.join(normalized, entry)
# Skip hidden directories and files
if entry.startswith("."):
continue
if os.path.isdir(full_path):
children.append(
{
"name": entry,
"type": "filesystem",
"location": full_path,
"parent_path": normalized,
}
)
return children
@staticmethod
def _normalize_path(path: str) -> str:
"""Normalize a path to absolute without trailing slashes.
Args:
path: Path to normalize.
Returns:
Absolute path without trailing slash.
"""
# Resolve to absolute
absolute = os.path.abspath(path)
# Remove trailing slashes (but keep root '/')
if len(absolute) > 1:
absolute = absolute.rstrip("/")
return absolute
@staticmethod
def _is_git_repo(path: str) -> bool:
"""Check if a path is a git repository.
Args:
path: Absolute path to check.
Returns:
True if the path contains a .git directory or is a bare repo.
"""
# Quick check: .git directory exists
if os.path.isdir(os.path.join(path, ".git")):
return True
# Fallback: ask git
try:
result = subprocess.run(
["git", "rev-parse", "--git-dir"],
cwd=path,
capture_output=True,
timeout=5,
)
return result.returncode == 0
except (subprocess.TimeoutExpired, OSError):
return False
@staticmethod
def _map_sandbox_strategy(read_only: bool) -> str:
"""Map to the appropriate sandbox strategy.
Read-only resources get 'none' (no sandbox needed).
Writable git repos get 'git_worktree'.
Args:
read_only: Whether the resource is read-only.
Returns:
Sandbox strategy string.
"""
if read_only:
return "none"
return "git_worktree"
@@ -0,0 +1,66 @@
"""Copy-on-write sandbox strategy stub for CleverAgents.
Implements B4.sandbox copy_on_write strategy as a placeholder
for post-M1 optimization. All methods raise NotImplementedError
with clear guidance on when this will be available.
"""
from __future__ import annotations
from cleveragents.infrastructure.sandbox.manager import SandboxStrategyHandler
_POST_M1_MSG = (
"Copy-on-write sandbox strategy is not yet implemented. "
"This optimization is planned for post-M1. "
"Use 'git_worktree' strategy for git repositories or "
"'none' for resources that don't require sandboxing."
)
class CopyOnWriteStrategy(SandboxStrategyHandler):
"""Copy-on-write sandbox strategy stub.
This strategy is intended for large-project optimization
where full copies are too expensive. It will use filesystem-level
copy-on-write (e.g., reflinks on btrfs/APFS) to create cheap
snapshots of resource directories.
Currently raises NotImplementedError for all operations.
Planned for post-M1 implementation.
"""
def create(self, resource_name: str, base_path: str) -> str:
"""Create a copy-on-write sandbox (not yet implemented).
Args:
resource_name: Name of the resource to sandbox.
base_path: Base directory for sandbox storage.
Raises:
NotImplementedError: Always, with post-M1 guidance.
"""
raise NotImplementedError(_POST_M1_MSG)
def cleanup(self, sandbox_path: str) -> None:
"""Clean up a copy-on-write sandbox (not yet implemented).
Args:
sandbox_path: Path to the sandbox to remove.
Raises:
NotImplementedError: Always, with post-M1 guidance.
"""
raise NotImplementedError(_POST_M1_MSG)
def rewrite_path(self, original: str, sandbox_path: str, resource_name: str) -> str:
"""Rewrite a path for copy-on-write sandbox (not yet implemented).
Args:
original: Original file path.
sandbox_path: Sandbox directory path.
resource_name: Name of the sandboxed resource.
Raises:
NotImplementedError: Always, with post-M1 guidance.
"""
raise NotImplementedError(_POST_M1_MSG)
@@ -0,0 +1,202 @@
"""Git worktree sandbox strategy for CleverAgents.
Implements B4.sandbox git_worktree strategy:
- Creates isolated git worktrees for plan execution
- Records metadata (worktree path, branch, base commit) for rollback
- Handles cleanup of worktrees
- Provides path rewriting from original repo to worktree
- Safe fallback for non-git or non-existent paths
"""
from __future__ import annotations
import os
import subprocess
import tempfile
from typing import Any
from cleveragents.infrastructure.sandbox.manager import SandboxStrategyHandler
def _run_git(
args: list[str],
cwd: str | None = None,
) -> subprocess.CompletedProcess[str]:
"""Run a git command and return the result.
Args:
args: Git subcommand and arguments (without 'git' prefix).
cwd: Working directory for the command.
Returns:
CompletedProcess with stdout/stderr as text.
"""
return subprocess.run(
["git", *args],
cwd=cwd,
capture_output=True,
text=True,
check=False,
)
def _get_head_commit(repo_path: str) -> str:
"""Get the current HEAD commit hash.
Args:
repo_path: Path to the git repository.
Returns:
Short commit hash string.
"""
result = _run_git(["rev-parse", "--short", "HEAD"], cwd=repo_path)
return result.stdout.strip()
def _validate_git_repo(base_path: str) -> None:
"""Validate that base_path is a valid git repository.
Args:
base_path: Path to validate.
Raises:
ValueError: If path doesn't exist or isn't a git repo.
"""
if not os.path.isdir(base_path):
msg = f"Repository path does not exist: {base_path}"
raise ValueError(msg)
result = _run_git(["rev-parse", "--is-inside-work-tree"], cwd=base_path)
if result.returncode != 0:
msg = f"Path is not a git repository: {base_path}"
raise ValueError(msg)
class GitWorktreeStrategy(SandboxStrategyHandler):
"""Git worktree-based sandbox strategy.
Creates a separate git worktree for each sandboxed resource,
allowing isolated modifications that can be committed or
discarded without affecting the main working tree.
Stores metadata about each created worktree for rollback
support (worktree_path, branch, base_commit).
"""
def __init__(self) -> None:
"""Initialize the strategy."""
self._last_metadata: dict[str, Any] = {}
@property
def last_metadata(self) -> dict[str, Any]:
"""Return metadata from the last create() call."""
return self._last_metadata
def create(self, resource_name: str, base_path: str) -> str:
"""Create a git worktree sandbox.
Creates a new worktree in a temporary directory with a
sandbox-prefixed branch name.
Args:
resource_name: Name of the resource being sandboxed.
base_path: Path to the git repository.
Returns:
Absolute path to the worktree directory.
Raises:
ValueError: If base_path is not a valid git repo.
"""
_validate_git_repo(base_path)
base_commit = _get_head_commit(base_path)
branch_name = f"sandbox/{resource_name}"
# Create worktree in a temp directory
worktree_dir = tempfile.mkdtemp(prefix=f"v4_worktree_{resource_name}_")
result = _run_git(
["worktree", "add", "-b", branch_name, worktree_dir],
cwd=base_path,
)
if result.returncode != 0:
# Clean up the temp dir on failure
_safe_rmdir(worktree_dir)
msg = (
f"Failed to create worktree for '{resource_name}': "
f"{result.stderr.strip()}"
)
raise ValueError(msg)
self._last_metadata = {
"worktree_path": worktree_dir,
"branch": branch_name,
"base_commit": base_commit,
"repo_path": base_path,
}
return worktree_dir
def cleanup(self, sandbox_path: str) -> None:
"""Remove a git worktree sandbox.
Prunes the worktree from the parent repo and removes
the worktree directory.
Args:
sandbox_path: Path to the worktree to remove.
"""
if not os.path.isdir(sandbox_path):
return
# Try to find the parent repo from the worktree's .git file
git_file = os.path.join(sandbox_path, ".git")
parent_repo = None
if os.path.isfile(git_file):
with open(git_file) as f:
content = f.read().strip()
# Format: "gitdir: /path/to/parent/.git/worktrees/<name>"
if content.startswith("gitdir:"):
gitdir = content.split(":", 1)[1].strip()
# Navigate up from .git/worktrees/<name> to repo root
parent_git = os.path.dirname(os.path.dirname(os.path.dirname(gitdir)))
if os.path.isdir(parent_git):
parent_repo = parent_git
# Remove the worktree directory first
_safe_rmdir(sandbox_path)
# Prune worktrees in the parent repo
if parent_repo:
_run_git(["worktree", "prune"], cwd=parent_repo)
def rewrite_path(self, original: str, sandbox_path: str, resource_name: str) -> str:
"""Rewrite a file path from the original repo to the worktree.
Replaces the original repo path prefix with the sandbox path.
Args:
original: Original file path in the repo.
sandbox_path: Worktree sandbox path.
resource_name: Name (or path) of the original resource/repo.
Returns:
Path rewritten to point inside the worktree.
"""
if resource_name in original:
return original.replace(resource_name, sandbox_path, 1)
return original
def _safe_rmdir(path: str) -> None:
"""Safely remove a directory tree, ignoring errors.
Args:
path: Directory path to remove.
"""
import contextlib
import shutil
with contextlib.suppress(OSError):
shutil.rmtree(path, ignore_errors=True)