Compare commits
3 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 32c8041826 | |||
| 8aa17b72c9 | |||
| be2659f70b |
@@ -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"
|
||||
@@ -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
|
||||
@@ -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)
|
||||
Reference in New Issue
Block a user