feat(sandbox): implement overlay filesystem sandbox strategy (#994)
CI / build (push) Successful in 17s
CI / lint (push) Successful in 3m41s
CI / quality (push) Successful in 3m47s
CI / unit_tests (push) Successful in 3m58s
CI / typecheck (push) Successful in 4m17s
CI / benchmark-regression (push) Has been skipped
CI / security (push) Successful in 4m22s
CI / docker (push) Successful in 1m22s
CI / e2e_tests (push) Successful in 8m57s
CI / coverage (push) Successful in 11m16s
CI / status-check (push) Successful in 3s
CI / benchmark-publish (push) Successful in 19m45s
CI / integration_tests (push) Failing after 20m53s

## Summary

Implement the overlay filesystem sandbox strategy with OverlayFS support and userspace fallback.

### Implementation

**`OverlaySandbox`** (`infrastructure/sandbox/overlay.py`, 497 lines):
- Detects OverlayFS availability at runtime via `/proc/filesystems` + `os.geteuid() == 0` check
- **Real OverlayFS mode** (requires root): creates upper/work/merged dirs, mounts overlay filesystem, captures writes in upper layer
- **Userspace fallback** (default in CI/containers): `shutil.copytree` the original into merged dir, tracks changes via `filecmp` diff on commit
- `create()`: sets up directory structure, mounts if available
- `commit()`: copies changed/added files from overlay to original, removes deleted files
- `rollback()`: unmounts (or removes) merged, recreates from scratch
- `cleanup()`: unmounts, removes all temp dirs, idempotent

### Domain Model Updates
- Added `OVERLAY = "overlay"` to `SandboxStrategy` enum in both `resource_type.py` and `resource.py`
- Added `STRATEGY_OVERLAY` to `SandboxFactory`, registered for `fs-mount`, `fs-directory`, `fs-file` resources

### Tests
- **22 Behave scenarios**: full lifecycle (create/commit/rollback/cleanup), status transitions, path traversal guard, fallback detection, error handling
- **6 Robot integration tests**: end-to-end overlay sandbox operations

### Quality Gates

| Session | Result |
|---|---|
| `nox -s lint` | PASS |
| `nox -s typecheck` | PASS (0 errors) |
| `nox -s unit_tests` | PASS (10,917 scenarios) |
| `nox -s coverage_report` | 97% (>= 97%) |

Closes #880

Reviewed-on: #994
Co-authored-by: Brent Edwards <brent.edwards@cleverthis.com>
Co-committed-by: Brent Edwards <brent.edwards@cleverthis.com>
This commit was merged in pull request #994.
This commit is contained in:
2026-03-21 03:14:40 +00:00
committed by Forgejo
parent 6531440431
commit 8a87262f86
15 changed files with 1681 additions and 19 deletions
+3
View File
@@ -2,6 +2,9 @@
## Unreleased
- Added overlay filesystem sandbox strategy with real OverlayFS support
and userspace copy-tree fallback. Includes lifecycle management, diff
computation, and sandbox factory registration. (#880)
- Added CLI polish infrastructure: shared constants.py (exit codes, format
constants), centralized errors.py (cli_error, cli_not_found, cli_warning),
and completion command for shell tab-completion generation. (#861)
+6 -5
View File
@@ -383,8 +383,8 @@ Feature: Consolidated Resource
# SandboxStrategy enum
Scenario: SandboxStrategy has exactly five values
Then the SandboxStrategy enum should have values "git_worktree, copy_on_write, transaction_rollback, snapshot, none"
Scenario: SandboxStrategy has exactly six values
Then the SandboxStrategy enum should have values "git_worktree, copy_on_write, transaction_rollback, snapshot, overlay, none"
Scenario Outline: SandboxStrategy accepts valid values
@@ -397,12 +397,13 @@ Feature: Consolidated Resource
| copy_on_write |
| transaction_rollback |
| snapshot |
| overlay |
| none |
Scenario: SandboxStrategy rejects overlay
When I try to create a SandboxStrategy with value "overlay"
Then a resource validation error should be raised
Scenario: SandboxStrategy accepts overlay
Given a SandboxStrategy value of "overlay"
Then the SandboxStrategy string value should be "overlay"
Scenario: SandboxStrategy rejects versioning
+9 -9
View File
@@ -758,10 +758,10 @@ Feature: Consolidated Sandbox
Then the factory should reject the request due to an unknown strategy
Scenario: The overlay strategy is now unknown and rejected
Scenario: The overlay strategy is supported
Given the sandbox factory is available
When a sandbox is requested for resource "fs-root" at "/mnt/data" using the overlay strategy
Then the factory should reject the request due to an unknown strategy
When the factory is asked whether the overlay strategy is supported
Then the factory should confirm the strategy is supported
Scenario: The versioning strategy is now unknown and rejected
@@ -810,22 +810,22 @@ Feature: Consolidated Sandbox
Then the factory should return only the none strategy as compatible
Scenario: A fs-mount resource supports copy-on-write and none
Scenario: A fs-mount resource supports copy-on-write, overlay, and none
Given the sandbox factory is available
When the compatible strategies for a "fs-mount" resource are queried
Then the factory should return copy-on-write and none as compatible
Then the factory should return copy-on-write, overlay, and none as compatible
Scenario: A fs-directory resource supports copy-on-write and none
Scenario: A fs-directory resource supports copy-on-write, overlay, and none
Given the sandbox factory is available
When the compatible strategies for a "fs-directory" resource are queried
Then the factory should return copy-on-write and none as compatible
Then the factory should return copy-on-write, overlay, and none as compatible
Scenario: A fs-file resource supports copy-on-write and none
Scenario: A fs-file resource supports copy-on-write, overlay, and none
Given the sandbox factory is available
When the compatible strategies for a "fs-file" resource are queried
Then the factory should return copy-on-write and none as compatible
Then the factory should return copy-on-write, overlay, and none as compatible
Scenario: An api_endpoint resource supports only the none strategy
+205
View File
@@ -0,0 +1,205 @@
Feature: Overlay filesystem sandbox lifecycle
As a developer
I want a sandbox that isolates plan changes using overlay filesystem semantics
So that changes are captured in an upper layer and merged on commit
# --- Creation ---
Scenario: Create an overlay sandbox
Given an ovl test directory is initialised
When an ovl sandbox is created for plan "plan-001"
Then the ovl sandbox should be in the "created" state
And the ovl sandbox context should reference plan "plan-001"
And the ovl sandbox context should have strategy metadata "overlay"
And the ovl sandbox merged path should exist
Scenario: Creating an overlay sandbox with empty plan_id raises ValueError
Given an ovl test directory is initialised
When an ovl sandbox is created with empty plan_id
Then an ovl ValueError should be raised with message "plan_id cannot be empty"
Scenario: Creating an overlay sandbox with empty resource_id raises ValueError
Given an ovl test directory is initialised
When an ovl sandbox is prepared with empty resource_id
Then an ovl ValueError should be raised with message "resource_id cannot be empty"
Scenario: Creating an overlay sandbox with empty original_path raises ValueError
Given an ovl test directory is initialised
When an ovl sandbox is prepared with empty original_path
Then an ovl ValueError should be raised with message "original_path cannot be empty"
Scenario: Creating an overlay sandbox on a non-existent directory raises SandboxCreationError
Given an ovl test directory is initialised
Given an ovl non-existent directory
When an ovl sandbox is created on the non-existent directory for plan "plan-001"
Then an ovl SandboxCreationError should be raised
# --- Directory structure ---
Scenario: Overlay sandbox creates correct directory structure
Given an ovl test directory is initialised
When an ovl sandbox is created for plan "plan-001"
Then the ovl sandbox should have upper, work, and merged directories
# --- Path resolution ---
Scenario: Resolve a path in the overlay merged directory
Given an ovl test directory is initialised
When an ovl sandbox is created for plan "plan-001"
And the ovl path "data/file.txt" is resolved
Then the ovl resolved path should be inside the merged directory
And the ovl sandbox should be in the "active" state
Scenario: Path traversal is rejected
Given an ovl test directory is initialised
When an ovl sandbox is created for plan "plan-001"
And the ovl path "../etc/passwd" is resolved
Then an ovl ValueError should be raised with message "Path traversal not allowed"
Scenario: Resolving a path on a cleaned-up sandbox raises SandboxStateError
Given an ovl test directory is initialised
When an ovl sandbox is created for plan "plan-001"
And the ovl sandbox is cleaned up
And the ovl path "file.txt" is resolved on a cleaned-up sandbox
Then an ovl SandboxStateError should be raised
# --- Commit ---
Scenario: Commit with no changes produces empty result
Given an ovl test directory is initialised
When an ovl sandbox is created for plan "plan-001"
And the ovl sandbox is committed
Then the ovl commit result should indicate success
And the ovl commit result should have 0 changed files
And the ovl commit result should have 0 added files
And the ovl commit result should have 0 deleted files
Scenario: Commit with a new file syncs it to original
Given an ovl test directory is initialised
When an ovl sandbox is created for plan "plan-001"
And an ovl file "new_file.txt" is created in the sandbox with content "hello world"
And the ovl sandbox is committed
Then the ovl commit result should indicate success
And the ovl commit result should have 1 added files
And the ovl file "new_file.txt" should exist in the original directory with content "hello world"
Scenario: Commit with a modified file syncs the change
Given an ovl test directory is initialised
When an ovl sandbox is created for plan "plan-001"
And the ovl existing file "existing.txt" is modified in the sandbox with content "updated"
And the ovl sandbox is committed
Then the ovl commit result should indicate success
And the ovl commit result should have 1 changed files
And the ovl file "existing.txt" in the original should have content "updated"
Scenario: Commit with a deleted file removes it from original
Given an ovl test directory is initialised
When an ovl sandbox is created for plan "plan-001"
And the ovl existing file "existing.txt" is deleted from the sandbox
And the ovl sandbox is committed
Then the ovl commit result should indicate success
And the ovl commit result should have 1 deleted files
And the ovl file "existing.txt" should not exist in the original directory
Scenario: Commit on a cleaned-up sandbox raises SandboxStateError
Given an ovl test directory is initialised
When an ovl sandbox is created for plan "plan-001"
And the ovl sandbox is cleaned up
And the ovl sandbox commit is attempted on cleaned-up sandbox
Then an ovl SandboxStateError should be raised
# --- Rollback ---
Scenario: Rollback restores the sandbox to original state
Given an ovl test directory is initialised
When an ovl sandbox is created for plan "plan-001"
And an ovl file "temp.txt" is created in the sandbox with content "temporary"
And the ovl path "temp.txt" is resolved
And the ovl sandbox is rolled back
Then the ovl sandbox should be in the "rolled_back" state
And the ovl file "temp.txt" should not exist in the sandbox
Scenario: Rollback on a non-active sandbox raises SandboxStateError
Given an ovl test directory is initialised
When an ovl sandbox is created for plan "plan-001"
And the ovl sandbox rollback is attempted on created sandbox
Then an ovl SandboxStateError should be raised
# --- Cleanup ---
Scenario: Cleanup removes the overlay directory
Given an ovl test directory is initialised
When an ovl sandbox is created for plan "plan-001"
And the ovl sandbox is cleaned up
Then the ovl sandbox should be in the "cleaned_up" state
And the ovl sandbox base path should not exist
Scenario: Cleanup is idempotent
Given an ovl test directory is initialised
When an ovl sandbox is created for plan "plan-001"
And the ovl sandbox is cleaned up
And the ovl sandbox is cleaned up again
Then the ovl sandbox should be in the "cleaned_up" state
# --- Protocol properties ---
Scenario: Overlay sandbox has a unique ULID identifier
Given an ovl test directory is initialised
When an ovl sandbox is instantiated
Then the ovl sandbox_id should be a valid ULID
And the ovl sandbox status should be "pending"
And the ovl sandbox context should be None
# --- Fallback mode ---
Scenario: Fallback mode works when OverlayFS is unavailable
Given an ovl test directory is initialised
When an ovl sandbox is created for plan "plan-fallback"
Then the ovl sandbox should use userspace fallback
# --- Coverage boost: edge cases ---
Scenario: Commit with a message stores it in metadata
Given an ovl test directory is initialised
When an ovl sandbox is created for plan "plan-msg"
And an ovl file "new.txt" is created in the sandbox with content "data"
And the ovl sandbox is committed with message "snapshot before deploy"
Then the ovl commit result should indicate success
And the ovl commit metadata should contain message "snapshot before deploy"
Scenario: get_path raises SandboxStateError when merged_dir is None
Given an ovl sandbox with merged_dir forced to None
When ovl get_path is called with "file.txt" expecting an error
Then an ovl SandboxStateError should be raised with message "merged directory not set"
Scenario: Commit wraps OSError into SandboxCommitError
Given an ovl test directory is initialised
And an ovl sandbox is created and activated for plan "plan-commiterr"
And ovl shutil.copy2 is patched to raise OSError
When the ovl sandbox commit is attempted
Then an ovl SandboxCommitError should be raised
And the ovl sandbox should be in the "errored" state
Scenario: Rollback wraps OSError into SandboxRollbackError
Given an ovl test directory is initialised
And an ovl sandbox is created and activated for plan "plan-rberr"
And ovl shutil.rmtree is patched to raise OSError for rollback
When the ovl sandbox rollback is attempted
Then an ovl SandboxRollbackError should be raised
And the ovl sandbox should be in the "errored" state
Scenario: Create wraps OSError into SandboxCreationError
Given an ovl test directory is initialised
And ovl shutil.copytree is patched to raise OSError
When an ovl sandbox create is attempted for plan "plan-createerr"
Then an ovl SandboxCreationError should be raised
And the ovl sandbox should be in the "errored" state
# --- Status transitions ---
Scenario: Status transitions follow protocol
Given an ovl test directory is initialised
When an ovl sandbox is created for plan "plan-001"
Then the ovl sandbox should be in the "created" state
When the ovl path "existing.txt" is resolved
Then the ovl sandbox should be in the "active" state
+615
View File
@@ -0,0 +1,615 @@
"""Step definitions for overlay filesystem sandbox lifecycle tests.
Covers the OverlaySandbox implementation including creation, path
resolution, commit, rollback, cleanup, and userspace fallback mode.
"""
from __future__ import annotations
import os
import re
import shutil
import tempfile
from unittest.mock import patch
from behave import given, then, when
from cleveragents.infrastructure.sandbox.overlay import OverlaySandbox
from cleveragents.infrastructure.sandbox.protocol import (
SandboxCommitError,
SandboxCreationError,
SandboxRollbackError,
SandboxStateError,
SandboxStatus,
)
# ---------------------------------------------------------------------------
# Givens
# ---------------------------------------------------------------------------
@given("an ovl test directory is initialised")
def given_ovl_test_directory(context):
"""Create a temporary directory with test files for overlay sandbox."""
context.ovl_tmpdir = tempfile.mkdtemp(prefix="ovl-test-")
context.add_cleanup(shutil.rmtree, context.ovl_tmpdir)
context.ovl_original = os.path.join(context.ovl_tmpdir, "original")
os.makedirs(context.ovl_original)
# Create test files
with open(os.path.join(context.ovl_original, "existing.txt"), "w") as fh:
fh.write("original content")
with open(os.path.join(context.ovl_original, "to_delete.txt"), "w") as fh:
fh.write("delete me")
context.ovl_error = None
context.ovl_sandbox = None
context.ovl_commit_result = None
context.ovl_resolved_path = None
@given("an ovl non-existent directory")
def given_ovl_nonexistent_dir(context):
"""Set up a path to a non-existent directory."""
context.ovl_nonexistent = os.path.join(context.ovl_tmpdir, "nonexistent")
# ---------------------------------------------------------------------------
# Whens - Creation
# ---------------------------------------------------------------------------
@when('an ovl sandbox is created for plan "{plan_id}"')
def when_ovl_sandbox_created(context, plan_id: str):
"""Create an OverlaySandbox and call create()."""
context.ovl_error = None
context.ovl_sandbox = OverlaySandbox(
resource_id="test-resource",
original_path=context.ovl_original,
)
try:
context.ovl_context = context.ovl_sandbox.create(plan_id)
except Exception as exc:
context.ovl_error = exc
@when("an ovl sandbox is created with empty plan_id")
def when_ovl_sandbox_empty_plan(context):
"""Attempt to create sandbox with empty plan_id."""
context.ovl_error = None
context.ovl_sandbox = OverlaySandbox(
resource_id="test-resource",
original_path=context.ovl_original,
)
try:
context.ovl_sandbox.create("")
except ValueError as exc:
context.ovl_error = exc
@when("an ovl sandbox is prepared with empty resource_id")
def when_ovl_sandbox_empty_resource_id(context):
"""Attempt to instantiate sandbox with empty resource_id."""
context.ovl_error = None
try:
OverlaySandbox(resource_id="", original_path=context.ovl_original)
except ValueError as exc:
context.ovl_error = exc
@when("an ovl sandbox is prepared with empty original_path")
def when_ovl_sandbox_empty_path(context):
"""Attempt to instantiate sandbox with empty original_path."""
context.ovl_error = None
try:
OverlaySandbox(resource_id="test-resource", original_path="")
except ValueError as exc:
context.ovl_error = exc
@when('an ovl sandbox is created on the non-existent directory for plan "{plan_id}"')
def when_ovl_sandbox_nonexistent(context, plan_id: str):
"""Create sandbox pointing to a non-existent directory."""
context.ovl_error = None
context.ovl_sandbox = OverlaySandbox(
resource_id="test-resource",
original_path=context.ovl_nonexistent,
)
try:
context.ovl_sandbox.create(plan_id)
except SandboxCreationError as exc:
context.ovl_error = exc
@when("an ovl sandbox is instantiated")
def when_ovl_sandbox_instantiated(context):
"""Instantiate sandbox without calling create()."""
context.ovl_sandbox = OverlaySandbox(
resource_id="test-resource",
original_path=context.ovl_original,
)
# ---------------------------------------------------------------------------
# Whens - Path resolution
# ---------------------------------------------------------------------------
@when('the ovl path "{path}" is resolved')
def when_ovl_path_resolved(context, path: str):
"""Resolve a path in the overlay sandbox."""
context.ovl_error = None
try:
context.ovl_resolved_path = context.ovl_sandbox.get_path(path)
except (ValueError, SandboxStateError) as exc:
context.ovl_error = exc
@when('the ovl path "{path}" is resolved on a cleaned-up sandbox')
def when_ovl_path_on_cleaned(context, path: str):
"""Resolve a path after sandbox cleanup."""
context.ovl_error = None
try:
context.ovl_sandbox.get_path(path)
except SandboxStateError as exc:
context.ovl_error = exc
# ---------------------------------------------------------------------------
# Whens - File operations in sandbox
# ---------------------------------------------------------------------------
@when('an ovl file "{filename}" is created in the sandbox with content "{content}"')
def when_ovl_file_created(context, filename: str, content: str):
"""Create a new file in the merged directory."""
merged = context.ovl_sandbox._merged_dir
filepath = os.path.join(merged, filename)
os.makedirs(os.path.dirname(filepath), exist_ok=True)
with open(filepath, "w") as fh:
fh.write(content)
@when(
'the ovl existing file "{filename}" is modified '
'in the sandbox with content "{content}"'
)
def when_ovl_file_modified(context, filename: str, content: str):
"""Modify an existing file in the merged directory."""
merged = context.ovl_sandbox._merged_dir
filepath = os.path.join(merged, filename)
with open(filepath, "w") as fh:
fh.write(content)
@when('the ovl existing file "{filename}" is deleted from the sandbox')
def when_ovl_file_deleted(context, filename: str):
"""Delete a file from the merged directory."""
merged = context.ovl_sandbox._merged_dir
filepath = os.path.join(merged, filename)
os.remove(filepath)
# ---------------------------------------------------------------------------
# Whens - Commit / Rollback / Cleanup
# ---------------------------------------------------------------------------
@when("the ovl sandbox is committed")
def when_ovl_sandbox_committed(context):
"""Commit sandbox changes."""
context.ovl_error = None
try:
context.ovl_commit_result = context.ovl_sandbox.commit()
except Exception as exc:
context.ovl_error = exc
@when("the ovl sandbox commit is attempted on cleaned-up sandbox")
def when_ovl_commit_on_cleaned(context):
"""Attempt to commit after cleanup."""
context.ovl_error = None
try:
context.ovl_sandbox.commit()
except SandboxStateError as exc:
context.ovl_error = exc
@when("the ovl sandbox is rolled back")
def when_ovl_sandbox_rolled_back(context):
"""Roll back the sandbox."""
context.ovl_error = None
try:
context.ovl_sandbox.rollback()
except Exception as exc:
context.ovl_error = exc
@when("the ovl sandbox rollback is attempted on created sandbox")
def when_ovl_rollback_on_created(context):
"""Attempt rollback from CREATED state (invalid)."""
context.ovl_error = None
try:
context.ovl_sandbox.rollback()
except SandboxStateError as exc:
context.ovl_error = exc
@when("the ovl sandbox is cleaned up")
def when_ovl_sandbox_cleaned_up(context):
"""Clean up the sandbox."""
context.ovl_sandbox.cleanup()
@when("the ovl sandbox is cleaned up again")
def when_ovl_sandbox_cleaned_up_again(context):
"""Clean up the sandbox again (idempotency check)."""
context.ovl_sandbox.cleanup()
# ---------------------------------------------------------------------------
# Thens - State
# ---------------------------------------------------------------------------
@then('the ovl sandbox should be in the "{state}" state')
def then_ovl_sandbox_state(context, state: str):
"""Assert sandbox is in the expected state."""
assert context.ovl_sandbox.status.value == state, (
f"Expected {state}, got {context.ovl_sandbox.status.value}"
)
@then('the ovl sandbox status should be "{status}"')
def then_ovl_sandbox_status(context, status: str):
"""Assert sandbox status value."""
assert context.ovl_sandbox.status.value == status
# ---------------------------------------------------------------------------
# Thens - Context
# ---------------------------------------------------------------------------
@then('the ovl sandbox context should reference plan "{plan_id}"')
def then_ovl_context_plan(context, plan_id: str):
"""Assert context references the correct plan."""
assert context.ovl_sandbox.context is not None
assert context.ovl_sandbox.context.plan_id == plan_id
@then('the ovl sandbox context should have strategy metadata "{strategy}"')
def then_ovl_context_strategy(context, strategy: str):
"""Assert context metadata includes strategy."""
assert context.ovl_sandbox.context is not None
assert context.ovl_sandbox.context.metadata["strategy"] == strategy
@then("the ovl sandbox context should be None")
def then_ovl_context_none(context):
"""Assert context is None before creation."""
assert context.ovl_sandbox.context is None
# ---------------------------------------------------------------------------
# Thens - Directory structure
# ---------------------------------------------------------------------------
@then("the ovl sandbox merged path should exist")
def then_ovl_merged_exists(context):
"""Assert the merged directory exists."""
assert context.ovl_sandbox._merged_dir is not None
assert os.path.isdir(context.ovl_sandbox._merged_dir)
@then("the ovl sandbox should have upper, work, and merged directories")
def then_ovl_directory_structure(context):
"""Assert all overlay layer directories exist."""
assert context.ovl_sandbox._upper_dir is not None
assert context.ovl_sandbox._work_dir is not None
assert context.ovl_sandbox._merged_dir is not None
assert os.path.isdir(context.ovl_sandbox._upper_dir)
assert os.path.isdir(context.ovl_sandbox._work_dir)
assert os.path.isdir(context.ovl_sandbox._merged_dir)
@then("the ovl sandbox base path should not exist")
def then_ovl_base_gone(context):
"""Assert the base directory has been removed."""
if context.ovl_sandbox._base_dir is not None:
assert not os.path.exists(context.ovl_sandbox._base_dir)
# ---------------------------------------------------------------------------
# Thens - Path resolution
# ---------------------------------------------------------------------------
@then("the ovl resolved path should be inside the merged directory")
def then_ovl_path_in_merged(context):
"""Assert resolved path is within the merged directory."""
assert context.ovl_resolved_path is not None
assert context.ovl_sandbox._merged_dir is not None
assert context.ovl_resolved_path.startswith(context.ovl_sandbox._merged_dir)
# ---------------------------------------------------------------------------
# Thens - Errors
# ---------------------------------------------------------------------------
@then('an ovl ValueError should be raised with message "{msg}"')
def then_ovl_value_error(context, msg: str):
"""Assert a ValueError was raised with expected message."""
assert context.ovl_error is not None, "Expected a ValueError"
assert isinstance(context.ovl_error, ValueError)
assert msg in str(context.ovl_error)
@then("an ovl SandboxCreationError should be raised")
def then_ovl_creation_error(context):
"""Assert a SandboxCreationError was raised."""
assert context.ovl_error is not None, "Expected a SandboxCreationError"
assert isinstance(context.ovl_error, SandboxCreationError)
@then("an ovl SandboxStateError should be raised")
def then_ovl_state_error(context):
"""Assert a SandboxStateError was raised."""
assert context.ovl_error is not None, "Expected a SandboxStateError"
assert isinstance(context.ovl_error, SandboxStateError)
# ---------------------------------------------------------------------------
# Thens - Commit results
# ---------------------------------------------------------------------------
@then("the ovl commit result should indicate success")
def then_ovl_commit_success(context):
"""Assert commit was successful."""
assert context.ovl_commit_result is not None
assert context.ovl_commit_result.success is True
@then("the ovl commit result should have {count:d} changed files")
def then_ovl_commit_changed(context, count: int):
"""Assert commit result has expected number of changed files."""
assert len(context.ovl_commit_result.changed_files) == count
@then("the ovl commit result should have {count:d} added files")
def then_ovl_commit_added(context, count: int):
"""Assert commit result has expected number of added files."""
assert len(context.ovl_commit_result.added_files) == count
@then("the ovl commit result should have {count:d} deleted files")
def then_ovl_commit_deleted(context, count: int):
"""Assert commit result has expected number of deleted files."""
assert len(context.ovl_commit_result.deleted_files) == count
# ---------------------------------------------------------------------------
# Thens - File assertions
# ---------------------------------------------------------------------------
@then(
'the ovl file "{filename}" should exist in the original '
'directory with content "{content}"'
)
def then_ovl_file_in_original(context, filename: str, content: str):
"""Assert file exists in original directory with expected content."""
filepath = os.path.join(context.ovl_original, filename)
assert os.path.isfile(filepath), f"File not found: {filepath}"
with open(filepath) as fh:
assert fh.read() == content
@then('the ovl file "{filename}" in the original should have content "{content}"')
def then_ovl_file_content(context, filename: str, content: str):
"""Assert file in original has expected content."""
filepath = os.path.join(context.ovl_original, filename)
with open(filepath) as fh:
assert fh.read() == content
@then('the ovl file "{filename}" should not exist in the original directory')
def then_ovl_file_not_in_original(context, filename: str):
"""Assert file does not exist in original directory."""
filepath = os.path.join(context.ovl_original, filename)
assert not os.path.exists(filepath)
@then('the ovl file "{filename}" should not exist in the sandbox')
def then_ovl_file_not_in_sandbox(context, filename: str):
"""Assert file does not exist in the merged directory."""
merged = context.ovl_sandbox._merged_dir
assert merged is not None
filepath = os.path.join(merged, filename)
assert not os.path.exists(filepath)
# ---------------------------------------------------------------------------
# Thens - ULID / Protocol
# ---------------------------------------------------------------------------
_ULID_RE = re.compile(r"^[0-9A-Z]{26}$")
@then("the ovl sandbox_id should be a valid ULID")
def then_ovl_valid_ulid(context):
"""Assert sandbox_id matches ULID format."""
assert _ULID_RE.match(context.ovl_sandbox.sandbox_id), (
f"Not a valid ULID: {context.ovl_sandbox.sandbox_id}"
)
# ---------------------------------------------------------------------------
# Thens - Fallback mode
# ---------------------------------------------------------------------------
@then("the ovl sandbox should use userspace fallback")
def then_ovl_fallback(context):
"""Assert sandbox is using userspace fallback (common in CI)."""
# In CI/containers, OverlayFS is typically not available
assert context.ovl_sandbox._use_real_overlay is False
# ---------------------------------------------------------------------------
# Coverage boost steps
# ---------------------------------------------------------------------------
@when('the ovl sandbox is committed with message "{msg}"')
def when_ovl_commit_with_message(context, msg: str):
"""Commit the overlay sandbox with a message."""
context.ovl_commit_result = context.ovl_sandbox.commit(message=msg)
@then('the ovl commit metadata should contain message "{msg}"')
def then_ovl_commit_metadata_message(context, msg: str):
"""Assert commit result metadata contains the message."""
assert context.ovl_commit_result.metadata.get("message") == msg
@given("an ovl sandbox with merged_dir forced to None")
def given_ovl_sandbox_merged_dir_none(context):
"""Create an OverlaySandbox with merged_dir forced to None."""
tmpdir = tempfile.mkdtemp(prefix="ovl-none-")
os.makedirs(os.path.join(tmpdir, "orig"))
context.ovl_sandbox = OverlaySandbox(
resource_id="res-none",
original_path=os.path.join(tmpdir, "orig"),
)
context.ovl_sandbox.create(plan_id="plan-none")
# Force merged_dir to None after creation
context.ovl_sandbox._merged_dir = None
context.ovl_sandbox._status = SandboxStatus.ACTIVE
context.ovl_error = None
context.ovl_tmpdir = tmpdir
@when('ovl get_path is called with "{path}" expecting an error')
def when_ovl_get_path_error(context, path: str):
"""Call get_path expecting an error."""
try:
context.ovl_sandbox.get_path(path)
context.ovl_error = None
except SandboxStateError as exc:
context.ovl_error = exc
@then('an ovl SandboxStateError should be raised with message "{msg}"')
def then_ovl_sandbox_state_error(context, msg: str):
"""Assert a SandboxStateError was raised with the expected message."""
assert context.ovl_error is not None, "Expected SandboxStateError but none raised"
assert msg in str(context.ovl_error), f"Expected '{msg}' in '{context.ovl_error}'"
@given('an ovl sandbox is created and activated for plan "{plan_id}"')
def given_ovl_sandbox_active(context, plan_id: str):
"""Create an overlay sandbox and activate it via get_path."""
context.ovl_sandbox = OverlaySandbox(
resource_id="res-active",
original_path=context.ovl_tmpdir,
)
context.ovl_sandbox.create(plan_id=plan_id)
context.ovl_sandbox.get_path("existing.txt")
context.ovl_error = None
@given("ovl shutil.copy2 is patched to raise OSError")
def given_ovl_copy2_raises(context):
"""Patch shutil.copy2 to raise OSError."""
context.ovl_copy2_patcher = patch(
"cleveragents.infrastructure.sandbox.overlay.shutil.copy2",
side_effect=OSError("disk full"),
)
context.ovl_copy2_patcher.start()
@when("the ovl sandbox commit is attempted")
def when_ovl_commit_attempted(context):
"""Attempt to commit, catching errors."""
# Ensure there's a changed file to trigger the copy2 path
merged = context.ovl_sandbox._merged_dir
if merged:
fpath = os.path.join(merged, "existing.txt")
with open(fpath, "w") as f:
f.write("modified for commit error test")
try:
context.ovl_sandbox.commit()
context.ovl_error = None
except SandboxCommitError as exc:
context.ovl_error = exc
finally:
if hasattr(context, "ovl_copy2_patcher"):
context.ovl_copy2_patcher.stop()
@then("an ovl SandboxCommitError should be raised")
def then_ovl_commit_error(context):
"""Assert a SandboxCommitError was raised."""
assert context.ovl_error is not None, "Expected SandboxCommitError"
assert isinstance(context.ovl_error, SandboxCommitError)
@given("ovl shutil.rmtree is patched to raise OSError for rollback")
def given_ovl_rmtree_raises(context):
"""Patch shutil.rmtree to raise OSError."""
context.ovl_rmtree_patcher = patch(
"cleveragents.infrastructure.sandbox.overlay.shutil.rmtree",
side_effect=OSError("permission denied"),
)
context.ovl_rmtree_patcher.start()
@when("the ovl sandbox rollback is attempted")
def when_ovl_rollback_attempted(context):
"""Attempt to rollback, catching errors."""
try:
context.ovl_sandbox.rollback()
context.ovl_error = None
except SandboxRollbackError as exc:
context.ovl_error = exc
finally:
if hasattr(context, "ovl_rmtree_patcher"):
context.ovl_rmtree_patcher.stop()
@then("an ovl SandboxRollbackError should be raised")
def then_ovl_rollback_error(context):
"""Assert a SandboxRollbackError was raised."""
assert context.ovl_error is not None, "Expected SandboxRollbackError"
assert isinstance(context.ovl_error, SandboxRollbackError)
@given("ovl shutil.copytree is patched to raise OSError")
def given_ovl_copytree_raises(context):
"""Patch shutil.copytree to raise OSError."""
context.ovl_copytree_patcher = patch(
"cleveragents.infrastructure.sandbox.overlay.shutil.copytree",
side_effect=OSError("I/O error"),
)
context.ovl_copytree_patcher.start()
@when('an ovl sandbox create is attempted for plan "{plan_id}"')
def when_ovl_create_attempted(context, plan_id: str):
"""Attempt to create a sandbox, catching errors."""
context.ovl_sandbox = OverlaySandbox(
resource_id="res-create-err",
original_path=context.ovl_tmpdir,
)
try:
context.ovl_sandbox.create(plan_id=plan_id)
context.ovl_error = None
except SandboxCreationError as exc:
context.ovl_error = exc
finally:
if hasattr(context, "ovl_copytree_patcher"):
context.ovl_copytree_patcher.stop()
@@ -214,6 +214,12 @@ def when_factory_is_supported_cow(context):
context.factory_supported = SandboxFactory.is_supported("copy_on_write")
@when("the factory is asked whether the overlay strategy is supported")
def when_factory_is_supported_overlay(context):
"""Check support for the 'overlay' strategy."""
context.factory_supported = SandboxFactory.is_supported("overlay")
@when("the factory is asked whether an unrecognised strategy is supported")
def when_factory_is_supported_unknown(context):
"""Check support for a made-up strategy."""
@@ -378,6 +384,12 @@ def then_factory_cow_none_strategies(context):
assert context.factory_strategies == ["copy_on_write", "none"]
@then("the factory should return copy-on-write, overlay, and none as compatible")
def then_factory_cow_overlay_none_strategies(context):
"""Assert the resource type supports copy_on_write, overlay, and none."""
assert context.factory_strategies == ["copy_on_write", "overlay", "none"]
@then("the factory should return only the none strategy as compatible")
def then_factory_none_only_strategies(context):
"""Assert only the 'none' strategy is returned."""
+254
View File
@@ -0,0 +1,254 @@
"""Helper utilities for overlay sandbox Robot integration tests.
Covers the OverlaySandbox implementation: creation, path resolution,
commit, rollback, cleanup, and userspace fallback mode.
"""
from __future__ import annotations
import os
import sys
import tempfile
from cleveragents.infrastructure.sandbox.overlay import OverlaySandbox
from cleveragents.infrastructure.sandbox.protocol import (
SandboxCreationError,
SandboxStateError,
SandboxStatus,
)
def _overlay_lifecycle() -> None:
"""Integration test: full overlay sandbox lifecycle."""
tmpdir = tempfile.mkdtemp(prefix="ovl-robot-")
original = os.path.join(tmpdir, "original")
os.makedirs(original)
with open(os.path.join(original, "readme.txt"), "w") as fh:
fh.write("original readme")
sandbox = OverlaySandbox(resource_id="robot-res", original_path=original)
assert sandbox.status == SandboxStatus.PENDING
assert sandbox.context is None
ctx = sandbox.create("plan-robot-001")
assert sandbox.status == SandboxStatus.CREATED
assert ctx.plan_id == "plan-robot-001"
assert ctx.metadata["strategy"] == "overlay"
# Resolve path -> transitions to ACTIVE
resolved = sandbox.get_path("readme.txt")
assert sandbox.status == SandboxStatus.ACTIVE
assert resolved.endswith("readme.txt")
# Modify file in merged directory
with open(resolved, "w") as fh:
fh.write("modified readme")
# Add a new file
new_path = sandbox.get_path("new_file.txt")
with open(new_path, "w") as fh:
fh.write("new content")
# Commit
result = sandbox.commit()
assert sandbox.status == SandboxStatus.COMMITTED
assert result.success is True
assert len(result.changed_files) == 1
assert len(result.added_files) == 1
assert len(result.deleted_files) == 0
# Verify changes in original
with open(os.path.join(original, "readme.txt")) as fh:
assert fh.read() == "modified readme"
with open(os.path.join(original, "new_file.txt")) as fh:
assert fh.read() == "new content"
# Cleanup
sandbox.cleanup()
assert sandbox.status == SandboxStatus.CLEANED_UP
# Idempotent cleanup
sandbox.cleanup()
assert sandbox.status == SandboxStatus.CLEANED_UP
print("overlay-lifecycle-ok")
def _overlay_rollback() -> None:
"""Integration test: overlay sandbox rollback."""
tmpdir = tempfile.mkdtemp(prefix="ovl-robot-rb-")
original = os.path.join(tmpdir, "original")
os.makedirs(original)
with open(os.path.join(original, "data.txt"), "w") as fh:
fh.write("original data")
sandbox = OverlaySandbox(resource_id="rb-res", original_path=original)
sandbox.create("plan-robot-rb")
# Make change and activate
merged = sandbox.get_path("data.txt")
with open(merged, "w") as fh:
fh.write("modified data")
# Rollback
sandbox.rollback()
assert sandbox.status == SandboxStatus.ROLLED_BACK
# Verify merged directory is restored by directly checking
# the merged dir (get_path not available in ROLLED_BACK state)
assert sandbox._merged_dir is not None
restored_path = os.path.join(sandbox._merged_dir, "data.txt")
with open(restored_path) as fh:
assert fh.read() == "original data"
sandbox.cleanup()
print("overlay-rollback-ok")
def _overlay_validation() -> None:
"""Integration test: overlay sandbox input validation."""
# Empty resource_id
try:
OverlaySandbox(resource_id="", original_path="/tmp")
print("FAIL: Expected ValueError for empty resource_id")
return
except ValueError:
pass
# Empty original_path
try:
OverlaySandbox(resource_id="res", original_path="")
print("FAIL: Expected ValueError for empty original_path")
return
except ValueError:
pass
# Non-existent directory
sandbox = OverlaySandbox(resource_id="res", original_path="/nonexistent/path")
try:
sandbox.create("plan-1")
print("FAIL: Expected SandboxCreationError")
return
except SandboxCreationError:
pass
# Empty plan_id
tmpdir = tempfile.mkdtemp(prefix="ovl-val-")
os.makedirs(os.path.join(tmpdir, "orig"))
sandbox2 = OverlaySandbox(
resource_id="res", original_path=os.path.join(tmpdir, "orig")
)
try:
sandbox2.create("")
print("FAIL: Expected ValueError for empty plan_id")
return
except ValueError:
pass
print("overlay-validation-ok")
def _overlay_directory_structure() -> None:
"""Integration test: overlay sandbox creates correct structure."""
tmpdir = tempfile.mkdtemp(prefix="ovl-robot-struct-")
original = os.path.join(tmpdir, "original")
os.makedirs(original)
with open(os.path.join(original, "file.txt"), "w") as fh:
fh.write("content")
sandbox = OverlaySandbox(resource_id="struct-res", original_path=original)
sandbox.create("plan-struct")
assert sandbox._upper_dir is not None
assert sandbox._work_dir is not None
assert sandbox._merged_dir is not None
assert os.path.isdir(sandbox._upper_dir)
assert os.path.isdir(sandbox._work_dir)
assert os.path.isdir(sandbox._merged_dir)
# Merged should contain original files
assert os.path.isfile(os.path.join(sandbox._merged_dir, "file.txt"))
sandbox.cleanup()
print("overlay-directory-structure-ok")
def _overlay_fallback_mode() -> None:
"""Integration test: overlay sandbox uses userspace fallback."""
tmpdir = tempfile.mkdtemp(prefix="ovl-robot-fb-")
original = os.path.join(tmpdir, "original")
os.makedirs(original)
sandbox = OverlaySandbox(resource_id="fb-res", original_path=original)
# In CI/containers, OverlayFS is typically not available
assert sandbox._use_real_overlay is False
sandbox.create("plan-fb")
sandbox.cleanup()
print("overlay-fallback-mode-ok")
def _overlay_state_errors() -> None:
"""Integration test: overlay sandbox state errors."""
tmpdir = tempfile.mkdtemp(prefix="ovl-robot-se-")
original = os.path.join(tmpdir, "original")
os.makedirs(original)
sandbox = OverlaySandbox(resource_id="se-res", original_path=original)
sandbox.create("plan-se")
# Rollback from CREATED should raise
try:
sandbox.rollback()
print("FAIL: Expected SandboxStateError for rollback from CREATED")
return
except SandboxStateError:
pass
# Commit then attempt commit again
sandbox.commit()
try:
sandbox.commit()
print("FAIL: Expected SandboxStateError for double commit")
return
except SandboxStateError:
pass
sandbox.cleanup()
# Get path on cleaned up sandbox should raise
try:
sandbox.get_path("file.txt")
print("FAIL: Expected SandboxStateError for get_path after cleanup")
return
except SandboxStateError:
pass
print("overlay-state-errors-ok")
_COMMANDS = {
"overlay-lifecycle": _overlay_lifecycle,
"overlay-rollback": _overlay_rollback,
"overlay-validation": _overlay_validation,
"overlay-directory-structure": _overlay_directory_structure,
"overlay-fallback-mode": _overlay_fallback_mode,
"overlay-state-errors": _overlay_state_errors,
}
def main() -> None:
"""Entry point for Robot helper."""
if len(sys.argv) < 2 or sys.argv[1] not in _COMMANDS:
valid = "|".join(sorted(_COMMANDS))
print(f"Usage: {sys.argv[0]} <{valid}>")
sys.exit(1)
_COMMANDS[sys.argv[1]]()
if __name__ == "__main__":
main()
+53
View File
@@ -0,0 +1,53 @@
*** Settings ***
Documentation Integration tests for overlay filesystem sandbox.
... Covers OverlaySandbox lifecycle, rollback, validation,
... directory structure, fallback mode, and state errors.
Resource ${CURDIR}/common.resource
Suite Setup Setup Test Environment
Suite Teardown Cleanup Test Environment
*** Variables ***
${HELPER_SCRIPT} robot/helper_overlay_sandbox.py
*** Test Cases ***
Overlay Sandbox Full Lifecycle
[Documentation] Verify create -> get_path -> modify -> commit -> cleanup
[Tags] sandbox overlay lifecycle
${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} overlay-lifecycle cwd=${WORKSPACE} on_timeout=kill timeout=60s
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} overlay-lifecycle-ok
Overlay Sandbox Rollback
[Documentation] Verify rollback discards changes and restores original state
[Tags] sandbox overlay rollback
${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} overlay-rollback cwd=${WORKSPACE} on_timeout=kill timeout=60s
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} overlay-rollback-ok
Overlay Sandbox Input Validation
[Documentation] Verify empty inputs and non-existent paths are rejected
[Tags] sandbox overlay validation
${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} overlay-validation cwd=${WORKSPACE} on_timeout=kill timeout=60s
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} overlay-validation-ok
Overlay Sandbox Directory Structure
[Documentation] Verify upper/work/merged directories are created
[Tags] sandbox overlay structure
${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} overlay-directory-structure cwd=${WORKSPACE} on_timeout=kill timeout=60s
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} overlay-directory-structure-ok
Overlay Sandbox Fallback Mode
[Documentation] Verify userspace fallback when OverlayFS unavailable
[Tags] sandbox overlay fallback
${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} overlay-fallback-mode cwd=${WORKSPACE} on_timeout=kill timeout=60s
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} overlay-fallback-mode-ok
Overlay Sandbox State Errors
[Documentation] Verify state machine rejects invalid transitions
[Tags] sandbox overlay state
${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} overlay-state-errors cwd=${WORKSPACE} on_timeout=kill timeout=60s
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} overlay-state-errors-ok
@@ -77,6 +77,7 @@ class SandboxStrategy(StrEnum):
COPY_ON_WRITE = "copy_on_write"
TRANSACTION_ROLLBACK = "transaction_rollback"
SNAPSHOT = "snapshot"
OVERLAY = "overlay"
NONE = "none"
@@ -81,6 +81,7 @@ class SandboxStrategy(StrEnum):
COPY_ON_WRITE = "copy_on_write"
TRANSACTION_ROLLBACK = "transaction_rollback"
SNAPSHOT = "snapshot"
OVERLAY = "overlay"
NONE = "none"
@@ -34,6 +34,7 @@ from cleveragents.infrastructure.sandbox.merge import (
SequentialMergeStrategy,
)
from cleveragents.infrastructure.sandbox.no_sandbox import NoSandbox
from cleveragents.infrastructure.sandbox.overlay import OverlaySandbox
from cleveragents.infrastructure.sandbox.protocol import (
CommitResult,
Sandbox,
@@ -65,6 +66,7 @@ __all__ = [
"MergeStrategy",
"NoSandbox",
"NoSandboxBoundaryError",
"OverlaySandbox",
"Sandbox",
"SandboxBoundaryError",
"SandboxCheckpoint",
@@ -17,6 +17,7 @@ from typing import TYPE_CHECKING, Literal
from cleveragents.infrastructure.sandbox.copy_on_write import CopyOnWriteSandbox
from cleveragents.infrastructure.sandbox.git_worktree import GitWorktreeSandbox
from cleveragents.infrastructure.sandbox.no_sandbox import NoSandbox
from cleveragents.infrastructure.sandbox.overlay import OverlaySandbox
from cleveragents.infrastructure.sandbox.protocol import (
Sandbox,
)
@@ -29,11 +30,12 @@ if TYPE_CHECKING:
logger = logging.getLogger(__name__)
# Strategy string constants aligned with spec SandboxStrategy enum (5 values)
# Strategy string constants aligned with spec SandboxStrategy enum (6 values)
STRATEGY_GIT_WORKTREE: Literal["git_worktree"] = "git_worktree"
STRATEGY_COPY_ON_WRITE: Literal["copy_on_write"] = "copy_on_write"
STRATEGY_TRANSACTION_ROLLBACK: Literal["transaction_rollback"] = "transaction_rollback"
STRATEGY_SNAPSHOT: Literal["snapshot"] = "snapshot"
STRATEGY_OVERLAY: Literal["overlay"] = "overlay"
STRATEGY_NONE: Literal["none"] = "none"
SandboxStrategyStr = Literal[
@@ -41,21 +43,22 @@ SandboxStrategyStr = Literal[
"copy_on_write",
"transaction_rollback",
"snapshot",
"overlay",
"none",
]
# Strategies that have concrete implementations
_IMPLEMENTED_STRATEGIES: frozenset[str] = frozenset(
{"none", "git_worktree", "copy_on_write", "transaction_rollback"}
{"none", "git_worktree", "copy_on_write", "transaction_rollback", "overlay"}
)
# Resource type to supported strategies mapping (spec-aligned)
_SUPPORTED_STRATEGIES: dict[str, list[SandboxStrategyStr]] = {
"git-checkout": ["git_worktree", "copy_on_write", "none"],
"git": ["none"],
"fs-mount": ["copy_on_write", "none"],
"fs-directory": ["copy_on_write", "none"],
"fs-file": ["copy_on_write", "none"],
"fs-mount": ["copy_on_write", "overlay", "none"],
"fs-directory": ["copy_on_write", "overlay", "none"],
"fs-file": ["copy_on_write", "overlay", "none"],
"api_endpoint": ["none"],
"postgres": ["transaction_rollback", "none"],
"mysql": ["transaction_rollback", "none"],
@@ -74,6 +77,7 @@ class SandboxFactory:
- ``"git_worktree"`` -> :class:`GitWorktreeSandbox`
- ``"copy_on_write"`` -> :class:`CopyOnWriteSandbox`
- ``"transaction_rollback"`` -> :class:`TransactionSandbox`
- ``"overlay"`` -> :class:`OverlaySandbox`
``"snapshot"`` raises ``NotImplementedError``.
@@ -146,6 +150,12 @@ class SandboxFactory:
original_path=original_path,
)
if sandbox_strategy == STRATEGY_OVERLAY:
return OverlaySandbox(
resource_id=resource_id,
original_path=original_path,
)
if sandbox_strategy == STRATEGY_SNAPSHOT:
raise NotImplementedError("Snapshot sandbox not yet implemented")
@@ -0,0 +1,498 @@
"""Overlay filesystem sandbox — OverlayFS with userspace fallback.
Creates upper/work/merged layers. Commit syncs to original; rollback
discards and recreates. For fs-mount, fs-directory, fs-file resources.
"""
from __future__ import annotations
import filecmp
import functools
import logging
import os
import shutil
import subprocess
import tempfile
from datetime import datetime
from ulid import ULID
from cleveragents.infrastructure.sandbox.protocol import (
CommitResult,
SandboxCommitError,
SandboxContext,
SandboxCreationError,
SandboxRollbackError,
SandboxStateError,
SandboxStatus,
)
logger = logging.getLogger(__name__)
@functools.lru_cache(maxsize=1)
def _is_overlayfs_available() -> bool:
"""Return True only if we can actually mount OverlayFS.
Uses a real mount probe ``mount --fake`` gives false positives
in Docker containers that lack CAP_SYS_ADMIN.
"""
try:
proc_fs = "/proc/filesystems"
if os.path.isfile(proc_fs):
with open(proc_fs) as fh:
if "overlay" not in fh.read():
return False
else:
return False
except OSError:
return False
if os.geteuid() != 0:
return False
# Real mount probe — attempt actual overlay mount in a temp dir.
probe_dir = None
try:
probe_dir = tempfile.mkdtemp(prefix="ca-overlay-probe-")
lower, upper = os.path.join(probe_dir, "l"), os.path.join(probe_dir, "u")
work, merged = os.path.join(probe_dir, "w"), os.path.join(probe_dir, "m")
for d in (lower, upper, work, merged):
os.makedirs(d)
opts = f"lowerdir={lower},upperdir={upper},workdir={work}"
rc = subprocess.run(
["mount", "-t", "overlay", "overlay", "-o", opts, merged],
capture_output=True,
timeout=5,
).returncode
if rc != 0:
return False
subprocess.run(["umount", merged], capture_output=True, timeout=5)
return True
except (OSError, subprocess.TimeoutExpired):
return False
finally:
if probe_dir is not None:
shutil.rmtree(probe_dir, ignore_errors=True)
class OverlaySandbox:
"""Sandbox that isolates changes using an overlay filesystem model.
Mounts a real OverlayFS union when available, otherwise falls back
to a userspace copy-based overlay.
Implements the
:class:`~cleveragents.infrastructure.sandbox.protocol.Sandbox`
protocol.
"""
def __init__(
self,
resource_id: str,
original_path: str,
) -> None:
"""Initialise an overlay sandbox.
Args:
resource_id: Identifier of the resource being sandboxed.
original_path: Path to the directory to sandbox.
Raises:
ValueError: If *resource_id* or *original_path* is empty.
"""
if not resource_id:
raise ValueError("resource_id cannot be empty")
if not original_path:
raise ValueError("original_path cannot be empty")
self._sandbox_id: str = str(ULID())
self._resource_id: str = resource_id
self._original_path: str = os.path.abspath(original_path)
self._status: SandboxStatus = SandboxStatus.PENDING
self._context: SandboxContext | None = None
# Layer directories — set after create()
self._base_dir: str | None = None
self._upper_dir: str | None = None
self._work_dir: str | None = None
self._merged_dir: str | None = None
self._use_real_overlay: bool = _is_overlayfs_available()
# -- protocol properties -------------------------------------------------
@property
def sandbox_id(self) -> str:
"""Unique identifier for this sandbox instance."""
return self._sandbox_id
@property
def status(self) -> SandboxStatus:
"""Current lifecycle status."""
return self._status
@property
def context(self) -> SandboxContext | None:
"""Context after creation, ``None`` before ``create``."""
return self._context
# -- protocol methods ----------------------------------------------------
def create(self, plan_id: str) -> SandboxContext:
"""Set up overlay directory layers and populate the merged view.
Args:
plan_id: The plan that owns this sandbox.
Returns:
A :class:`SandboxContext` with the merged layer path.
Raises:
ValueError: If *plan_id* is empty.
SandboxStateError: If not in ``PENDING`` status.
SandboxCreationError: If the directory setup fails.
"""
if not plan_id:
raise ValueError("plan_id cannot be empty")
SandboxStatus.assert_transition(self._status, SandboxStatus.CREATED)
try:
if not os.path.isdir(self._original_path):
raise SandboxCreationError(
f"Original path is not a directory: {self._original_path}"
)
self._base_dir = tempfile.mkdtemp(prefix="ca-overlay-sandbox-")
self._upper_dir = os.path.join(self._base_dir, "upper")
self._work_dir = os.path.join(self._base_dir, "work")
self._merged_dir = os.path.join(self._base_dir, "merged")
os.makedirs(self._upper_dir, exist_ok=True)
os.makedirs(self._work_dir, exist_ok=True)
os.makedirs(self._merged_dir, exist_ok=True)
if self._use_real_overlay:
self._mount_overlay()
else:
# Userspace fallback: copy original into merged
shutil.copytree(
self._original_path,
self._merged_dir,
symlinks=False,
dirs_exist_ok=True,
)
except SandboxCreationError:
self._status = SandboxStatus.ERRORED
raise
except OSError as exc:
self._status = SandboxStatus.ERRORED
raise SandboxCreationError(
f"Failed to create overlay sandbox for resource "
f"{self._resource_id}: {exc}"
) from exc
self._context = SandboxContext(
sandbox_id=self._sandbox_id,
sandbox_path=self._merged_dir,
original_path=self._original_path,
resource_id=self._resource_id,
plan_id=plan_id,
created_at=datetime.now(),
metadata={
"strategy": "overlay",
"sandbox_path": self._merged_dir,
"use_real_overlay": self._use_real_overlay,
},
)
self._status = SandboxStatus.CREATED
logger.info(
"Created overlay sandbox: plan=%s resource=%s merged=%s real_overlay=%s",
plan_id,
self._resource_id,
self._merged_dir,
self._use_real_overlay,
)
return self._context
def get_path(self, resource_path: str) -> str:
"""Translate a resource-relative path to the merged layer.
Args:
resource_path: Path relative to the resource root.
Returns:
Absolute path inside the merged directory.
Raises:
SandboxStateError: If sandbox is not in a usable status.
ValueError: If *resource_path* attempts directory traversal.
"""
if self._status not in (
SandboxStatus.CREATED,
SandboxStatus.ACTIVE,
SandboxStatus.ROLLED_BACK,
):
raise SandboxStateError(
f"Cannot resolve path in status {self._status.value}"
)
if ".." in resource_path.split("/"):
raise ValueError(f"Path traversal not allowed: {resource_path}")
if self._status == SandboxStatus.CREATED:
self._status = SandboxStatus.ACTIVE
if self._merged_dir is None:
raise SandboxStateError("Sandbox merged directory not set")
return os.path.join(self._merged_dir, resource_path)
def commit(self, message: str | None = None) -> CommitResult:
"""Apply overlay changes back to the original directory.
For real OverlayFS: copies the upper layer contents to original.
For userspace fallback: diffs merged against original.
Args:
message: Optional log message (stored in result metadata).
Returns:
A :class:`CommitResult` describing the outcome.
Raises:
SandboxCommitError: If the sync fails.
SandboxStateError: If sandbox is not in a committable status.
"""
if self._status not in (
SandboxStatus.CREATED,
SandboxStatus.ACTIVE,
):
raise SandboxStateError(f"Cannot commit from status {self._status.value}")
SandboxStatus.assert_transition(self._status, SandboxStatus.COMMITTED)
if self._merged_dir is None:
raise SandboxStateError("Sandbox merged directory not set")
try:
changed_files, added_files, deleted_files = self._compute_diff(
self._merged_dir, self._original_path
)
# Apply changes: copy modified and new files
for rel_path in changed_files + added_files:
src = os.path.join(self._merged_dir, rel_path)
dst = os.path.join(self._original_path, rel_path)
dst_dir = os.path.dirname(dst)
if dst_dir:
os.makedirs(dst_dir, exist_ok=True)
shutil.copy2(src, dst)
# Apply deletions
for rel_path in deleted_files:
dst = os.path.join(self._original_path, rel_path)
if os.path.exists(dst):
os.remove(dst)
except OSError as exc:
self._status = SandboxStatus.ERRORED
raise SandboxCommitError(
f"Failed to commit overlay sandbox {self._sandbox_id}: {exc}"
) from exc
self._status = SandboxStatus.COMMITTED
logger.info(
"Committed overlay sandbox: sandbox_id=%s changed=%d added=%d deleted=%d",
self._sandbox_id,
len(changed_files),
len(added_files),
len(deleted_files),
)
metadata: dict[str, object] = {}
if message:
metadata["message"] = message
return CommitResult(
sandbox_id=self._sandbox_id,
success=True,
commit_ref=None,
changed_files=changed_files,
added_files=added_files,
deleted_files=deleted_files,
error=None,
timestamp=datetime.now(),
metadata=metadata,
)
def rollback(self) -> None:
"""Discard overlay changes by resetting the merged view.
For real OverlayFS: unmounts, recreates upper/work, remounts.
For userspace fallback: deletes merged, re-copies original.
Raises:
SandboxRollbackError: If the rollback fails.
SandboxStateError: If called in an invalid status.
"""
if self._status != SandboxStatus.ACTIVE:
raise SandboxStateError(f"Cannot rollback from status {self._status.value}")
SandboxStatus.assert_transition(self._status, SandboxStatus.ROLLED_BACK)
if self._merged_dir is None:
raise SandboxStateError("Sandbox merged directory not set")
try:
if self._use_real_overlay:
self._unmount_overlay()
# Reset upper and work directories
if self._upper_dir is not None:
shutil.rmtree(self._upper_dir, ignore_errors=True)
os.makedirs(self._upper_dir, exist_ok=True)
if self._work_dir is not None:
shutil.rmtree(self._work_dir, ignore_errors=True)
os.makedirs(self._work_dir, exist_ok=True)
self._mount_overlay()
else:
# Userspace fallback: delete merged, re-copy
shutil.rmtree(self._merged_dir, ignore_errors=True)
shutil.copytree(
self._original_path,
self._merged_dir,
symlinks=False,
dirs_exist_ok=False,
)
except OSError as exc:
self._status = SandboxStatus.ERRORED
raise SandboxRollbackError(
f"Failed to rollback overlay sandbox {self._sandbox_id}: {exc}"
) from exc
self._status = SandboxStatus.ROLLED_BACK
logger.info(
"Rolled back overlay sandbox: sandbox_id=%s",
self._sandbox_id,
)
def cleanup(self) -> None:
"""Remove all overlay layers and mountpoints.
Idempotent safe to call multiple times.
Raises:
SandboxError: On unexpected errors during cleanup.
"""
if self._status == SandboxStatus.CLEANED_UP:
return
logger.debug(
"Cleaning up overlay sandbox: sandbox_id=%s base=%s",
self._sandbox_id,
self._base_dir,
)
# Unmount if real overlay was used
if self._use_real_overlay and self._merged_dir is not None:
self._unmount_overlay()
if self._base_dir is not None and os.path.exists(self._base_dir):
shutil.rmtree(self._base_dir, ignore_errors=True)
self._status = SandboxStatus.CLEANED_UP
logger.info(
"Cleaned up overlay sandbox: sandbox_id=%s",
self._sandbox_id,
)
# -- internal helpers ----------------------------------------------------
def _mount_overlay(self) -> None:
"""Mount an OverlayFS union at the merged directory.
Raises:
SandboxCreationError: If the mount command fails.
"""
opts = (
f"lowerdir={self._original_path},"
f"upperdir={self._upper_dir},"
f"workdir={self._work_dir}"
)
try:
subprocess.run(
[
"mount",
"-t",
"overlay",
"overlay",
"-o",
opts,
str(self._merged_dir),
],
capture_output=True,
check=True,
timeout=30,
)
except (subprocess.CalledProcessError, OSError) as exc:
raise SandboxCreationError(f"Failed to mount OverlayFS: {exc}") from exc
def _unmount_overlay(self) -> None:
"""Unmount the OverlayFS union. Errors are logged but not raised."""
if self._merged_dir is None:
return
try:
subprocess.run(
["umount", str(self._merged_dir)],
capture_output=True,
check=True,
timeout=30,
)
except (subprocess.CalledProcessError, OSError) as exc:
logger.warning(
"Failed to unmount overlay at %s: %s",
self._merged_dir,
exc,
)
@staticmethod
def _compute_diff(
sandbox_dir: str, original_dir: str
) -> tuple[list[str], list[str], list[str]]:
"""Compare sandbox to original and return changed/added/deleted.
Args:
sandbox_dir: Path to the merged/sandbox directory.
original_dir: Path to the original directory.
Returns:
Tuple of ``(changed, added, deleted)`` relative-path lists.
"""
# Collect all files in both trees
sandbox_files: set[str] = {
os.path.relpath(os.path.join(dp, f), sandbox_dir)
for dp, _, fnames in os.walk(sandbox_dir)
for f in fnames
}
original_files: set[str] = {
os.path.relpath(os.path.join(dp, f), original_dir)
for dp, _, fnames in os.walk(original_dir)
for f in fnames
}
added = sorted(sandbox_files - original_files)
deleted = sorted(original_files - sandbox_files)
changed = [
rel
for rel in sorted(sandbox_files & original_files)
if not filecmp.cmp(
os.path.join(sandbox_dir, rel),
os.path.join(original_dir, rel),
shallow=False,
)
]
return changed, added, deleted
@@ -190,6 +190,9 @@ class CommitResult:
timestamp: datetime = field(default_factory=datetime.now)
"""When the commit occurred."""
metadata: dict[str, Any] = field(default_factory=dict)
"""Arbitrary key-value metadata (e.g. commit message)."""
# ---------------------------------------------------------------------------
# Sandbox protocol (B3.1d)
+4
View File
@@ -1061,6 +1061,10 @@ validate_connection # noqa: B018, F821
# Transaction sandbox — public API (issue #342)
TransactionSandbox # noqa: B018, F821
# Overlay sandbox — public API (issue #880)
OverlaySandbox # noqa: B018, F821
_is_overlayfs_available # noqa: B018, F821
# ComponentResolver — pluggable scope chain resolution (#552)
ComponentResolver # noqa: B018, F821
ComponentNotFoundError # noqa: B018, F821