feat(resource): implement DevcontainerHandler missing protocol methods (#1286)
CI / benchmark-publish (push) Has been cancelled
CI / integration_tests (push) Has been cancelled
CI / lint (push) Has been cancelled
CI / typecheck (push) Has been cancelled
CI / security (push) Has been cancelled
CI / build (push) Has been cancelled
CI / unit_tests (push) Has been cancelled
CI / quality (push) Has been cancelled
CI / helm (push) Has been cancelled
CI / e2e_tests (push) Has been cancelled
CI / status-check (push) Has been cancelled
CI / coverage (push) Has been cancelled
CI / benchmark-regression (push) Has been cancelled
CI / docker (push) Has been cancelled

Implements the four missing protocol methods on DevcontainerHandler required by Epic #825 (ResourceHandler Protocol Completion):

- delete(): uses 'devcontainer exec rm -rf' to delete files/dirs inside the container; returns DeleteResult(success=False) for missing/stopped containers rather than raising.
- list_children(): uses 'devcontainer exec ls -1' to enumerate workspace entries; returns an empty list on container failure.
- diff(): compares content hashes between the devcontainer workspace and another filesystem location; uses EMPTY_CONTENT_HASH sentinel to treat both-absent as no-change.
- create_sandbox(): delegates to BaseResourceHandler.create_sandbox with lazy activation (same DETECTED/STOPPED/FAILED guard as resolve()).

All methods raise ValueError for resources with no location.

Adds 18 Behave BDD scenarios covering success paths, failure/stopped-container paths, empty-path edge cases, and ValueError guards.

ISSUES CLOSED: #1242

Co-authored-by: Jeffrey Phillips Freeman <the@jeffreyfreeman.me>
Co-committed-by: Jeffrey Phillips Freeman <the@jeffreyfreeman.me>
This commit was merged in pull request #1286.
This commit is contained in:
2026-04-02 17:08:52 +00:00
committed by Forgejo
parent c2097ee2ed
commit ee980ee117
3 changed files with 930 additions and 1 deletions
@@ -0,0 +1,150 @@
Feature: DevcontainerHandler missing protocol methods
As a developer using DevcontainerHandler
I want delete(), list_children(), diff(), and create_sandbox() to be implemented
So that the handler satisfies the full ResourceHandler protocol (issue #1242)
# delete() method
Scenario: dcproto delete file successfully from devcontainer
Given dcproto a devcontainer handler
And dcproto a devcontainer-instance resource with location "/ws/project"
And dcproto subprocess returns successful rm output
When dcproto I delete path "src/old_file.py" from the resource
Then dcproto the delete should succeed
And dcproto the delete message should contain "src/old_file.py"
And dcproto the subprocess should have been called with "rm" and "src/old_file.py"
Scenario: dcproto delete with empty path targets workspace root
Given dcproto a devcontainer handler
And dcproto a devcontainer-instance resource with location "/ws/project"
And dcproto subprocess returns successful rm output
When dcproto I delete with empty path from the resource
Then dcproto the delete should succeed
And dcproto the subprocess should have been called with "rm" and "."
Scenario: dcproto delete fails when container is stopped
Given dcproto a devcontainer handler
And dcproto a devcontainer-instance resource with location "/ws/project"
And dcproto subprocess returns failed rm output with stderr "container not running"
When dcproto I delete path "file.txt" from the resource
Then dcproto the delete should fail
And dcproto the delete failure message should contain "container not running"
Scenario: dcproto delete from resource with no location raises ValueError
Given dcproto a devcontainer handler
And dcproto a devcontainer-instance resource with no location
When dcproto I delete path "any/file.txt" from the resource
Then dcproto the delete should raise ValueError containing "no location"
# ── list_children() method ───────────────────────────────────
Scenario: dcproto list_children returns sorted names from devcontainer
Given dcproto a devcontainer handler
And dcproto a devcontainer-instance resource with location "/ws/project"
And dcproto subprocess returns successful ls listing "src\ntests\nREADME.md"
When dcproto I list children of the resource
Then dcproto the children list should have 3 entries
And dcproto the children list should be "README.md,src,tests"
Scenario: dcproto list_children returns empty list when container is stopped
Given dcproto a devcontainer handler
And dcproto a devcontainer-instance resource with location "/ws/project"
And dcproto subprocess returns failed ls output
When dcproto I list children of the resource
Then dcproto the children list should have 0 entries
Scenario: dcproto list_children filters blank lines from ls output
Given dcproto a devcontainer handler
And dcproto a devcontainer-instance resource with location "/ws/project"
And dcproto subprocess returns successful ls listing "alpha\n\n\nbeta\n"
When dcproto I list children of the resource
Then dcproto the children list should have 2 entries
And dcproto the children list should be "alpha,beta"
Scenario: dcproto list_children from resource with no location raises ValueError
Given dcproto a devcontainer handler
And dcproto a devcontainer-instance resource with no location
When dcproto I list children of the resource
Then dcproto the list children should raise ValueError containing "no location"
Scenario: dcproto list_children uses devcontainer exec ls command
Given dcproto a devcontainer handler
And dcproto a devcontainer-instance resource with location "/ws/project"
And dcproto subprocess returns successful ls listing "main.py"
When dcproto I list children of the resource
Then dcproto the subprocess should have been called with "ls" and "--workspace-folder"
# ── diff() method ────────────────────────────────────────────
Scenario: dcproto diff detects changes when hashes differ
Given dcproto a devcontainer handler
And dcproto a devcontainer-instance resource with location "/ws/project"
And dcproto the container content hash is "aabbccdd"
And dcproto a temporary directory with a different file
When dcproto I diff the resource against the temporary directory
Then dcproto the diff should report has_changes true
And dcproto the diff files_changed should be 1
Scenario: dcproto diff reports no changes when hashes match
Given dcproto a devcontainer handler
And dcproto a devcontainer-instance resource with location "/ws/project"
And dcproto the container content hash matches the other location
When dcproto I diff the resource against the same location
Then dcproto the diff should report has_changes false
And dcproto the diff files_changed should be 0
Scenario: dcproto diff reports no changes when both sides are absent
Given dcproto a devcontainer handler
And dcproto a devcontainer-instance resource with location "/nonexistent/path"
When dcproto I diff the resource against "/another/nonexistent/path"
Then dcproto the diff should report has_changes false
Scenario: dcproto diff from resource with no location raises ValueError
Given dcproto a devcontainer handler
And dcproto a devcontainer-instance resource with no location
When dcproto I diff the resource against "/some/path"
Then dcproto the diff should raise ValueError containing "no location"
Scenario: dcproto diff unified_diff is always empty string
Given dcproto a devcontainer handler
And dcproto a devcontainer-instance resource with location "/ws/project"
And dcproto the container content hash is "aabbccdd"
And dcproto a temporary directory with a different file
When dcproto I diff the resource against the temporary directory
Then dcproto the diff unified_diff should be empty
# ── create_sandbox() method ──────────────────────────────────
Scenario: dcproto create_sandbox delegates to base class for running container
Given dcproto a devcontainer handler
And dcproto a devcontainer-instance resource with location "/ws/project"
And dcproto the container is in RUNNING state
And dcproto a mock sandbox manager that returns a valid sandbox
When dcproto I create a sandbox for the resource with plan "plan-001"
Then dcproto the sandbox result should have strategy "none"
And dcproto the sandbox result should have a sandbox path
Scenario: dcproto create_sandbox activates container when in DETECTED state
Given dcproto a devcontainer handler
And dcproto a devcontainer-instance resource with location "/ws/project"
And dcproto the container is in DETECTED state
And dcproto a mock sandbox manager that returns a valid sandbox
And dcproto activate_container is mocked to succeed
When dcproto I create a sandbox for the resource with plan "plan-002"
Then dcproto activate_container should have been called
Scenario: dcproto create_sandbox activates container when in STOPPED state
Given dcproto a devcontainer handler
And dcproto a devcontainer-instance resource with location "/ws/project"
And dcproto the container is in STOPPED state
And dcproto a mock sandbox manager that returns a valid sandbox
And dcproto activate_container is mocked to succeed
When dcproto I create a sandbox for the resource with plan "plan-003"
Then dcproto activate_container should have been called
Scenario: dcproto create_sandbox raises ValueError for resource with no location
Given dcproto a devcontainer handler
And dcproto a devcontainer-instance resource with no location
And dcproto a mock sandbox manager that returns a valid sandbox
When dcproto I create a sandbox for the resource with plan "plan-004"
Then dcproto the create_sandbox should raise ValueError containing "no location"
@@ -0,0 +1,576 @@
"""Step definitions for devcontainer_handler_protocol_methods.feature.
Exercises the four new protocol methods added to DevcontainerHandler
by issue #1242:
- delete() — devcontainer exec rm
- list_children() — devcontainer exec ls (returns list[str])
- diff() — content hash comparison
- create_sandbox() — container clone/snapshot via base-class delegation
All subprocess.run calls are mocked via unittest.mock.patch.
Step prefix: dcproto (devcontainer protocol methods).
"""
from __future__ import annotations
import os
import tempfile
from unittest.mock import MagicMock, patch
from behave import given, then, when
from behave.runner import Context
from cleveragents.domain.models.core.container_lifecycle import (
ContainerLifecycleState,
ContainerLifecycleTracker,
)
from cleveragents.domain.models.core.resource import PhysVirt, Resource
from cleveragents.resource.handlers.devcontainer import DevcontainerHandler
# Valid 26-char Crockford Base32 ULID for test resources.
# Crockford Base32 excludes I, L, O, U — use only [0-9A-HJKMNP-TV-Z].
_TEST_ULID = "01JTESTDCPRTM000000000000A"
def _make_resource(
location: str | None = "/ws/project",
resource_id: str = _TEST_ULID,
) -> Resource:
"""Create a minimal devcontainer-instance Resource for testing."""
return Resource(
resource_id=resource_id,
name="test-devcontainer",
resource_type_name="devcontainer-instance",
classification=PhysVirt.PHYSICAL,
description="Test devcontainer resource",
location=location,
)
# ── Given steps ──────────────────────────────────────────────
@given("dcproto a devcontainer handler")
def step_dcproto_create_handler(context: Context) -> None:
context.dcproto_handler = DevcontainerHandler()
@given('dcproto a devcontainer-instance resource with location "{loc}"')
def step_dcproto_resource_with_location(context: Context, loc: str) -> None:
context.dcproto_resource = _make_resource(location=loc)
@given("dcproto a devcontainer-instance resource with no location")
def step_dcproto_resource_no_location(context: Context) -> None:
context.dcproto_resource = _make_resource(location=None)
@given("dcproto subprocess returns successful rm output")
def step_dcproto_subprocess_rm_success(context: Context) -> None:
mock_result = MagicMock()
mock_result.returncode = 0
mock_result.stdout = b""
mock_result.stderr = b""
context.dcproto_mock_result = mock_result
context.dcproto_mock_mode = "rm_success"
@given('dcproto subprocess returns failed rm output with stderr "{stderr}"')
def step_dcproto_subprocess_rm_failure(context: Context, stderr: str) -> None:
mock_result = MagicMock()
mock_result.returncode = 1
mock_result.stdout = b""
mock_result.stderr = stderr.encode("utf-8")
context.dcproto_mock_result = mock_result
context.dcproto_mock_mode = "rm_failure"
@given("dcproto subprocess returns failed ls output")
def step_dcproto_subprocess_ls_failure(context: Context) -> None:
mock_result = MagicMock()
mock_result.returncode = 1
mock_result.stdout = ""
mock_result.stderr = "ls: cannot access"
context.dcproto_mock_result = mock_result
context.dcproto_mock_mode = "ls_failure"
@given('dcproto subprocess returns successful ls listing "{text}"')
def step_dcproto_subprocess_ls_listing(context: Context, text: str) -> None:
"""Configure mock for list_children which uses text=True."""
mock_result = MagicMock()
mock_result.returncode = 0
raw = text.replace("\\n", "\n")
mock_result.stdout = raw
mock_result.stderr = ""
context.dcproto_mock_result = mock_result
context.dcproto_mock_mode = "ls_listing"
@given('dcproto the container content hash is "{fake_hash}"')
def step_dcproto_container_hash(context: Context, fake_hash: str) -> None:
"""Patch content_hash to return a fixed value for the container side."""
context.dcproto_container_hash = fake_hash
@given("dcproto a temporary directory with a different file")
def step_dcproto_temp_dir_different(context: Context) -> None:
"""Create a temp dir with a file so its hash differs from the container."""
tmpdir = tempfile.mkdtemp()
context.add_cleanup(lambda: __import__("shutil").rmtree(tmpdir, ignore_errors=True))
with open(os.path.join(tmpdir, "different.txt"), "w") as f:
f.write("different content that will produce a unique hash")
context.dcproto_other_location = tmpdir
@given("dcproto the container content hash matches the other location")
def step_dcproto_hash_matches(context: Context) -> None:
"""Signal that we want the hashes to match (same location used for both)."""
context.dcproto_hash_should_match = True
@given("dcproto I diff the resource against the same location")
def step_dcproto_same_location_setup(context: Context) -> None:
"""Use the resource's own location as other_location for a no-diff scenario."""
context.dcproto_other_location = context.dcproto_resource.location
@given("dcproto the container is in RUNNING state")
def step_dcproto_container_running(context: Context) -> None:
from cleveragents.resource.handlers._devcontainer_internals import (
set_lifecycle_tracker,
)
# Use model_copy to bypass transition validation for test setup.
tracker = ContainerLifecycleTracker(
resource_id=context.dcproto_resource.resource_id,
)
tracker = tracker.model_copy(
update={"current_state": ContainerLifecycleState.RUNNING}
)
set_lifecycle_tracker(tracker)
context.dcproto_container_state = ContainerLifecycleState.RUNNING
@given("dcproto the container is in DETECTED state")
def step_dcproto_container_detected(context: Context) -> None:
from cleveragents.resource.handlers._devcontainer_internals import (
set_lifecycle_tracker,
)
# DETECTED is the initial state — no transition needed.
tracker = ContainerLifecycleTracker(
resource_id=context.dcproto_resource.resource_id,
)
set_lifecycle_tracker(tracker)
context.dcproto_container_state = ContainerLifecycleState.DETECTED
@given("dcproto the container is in STOPPED state")
def step_dcproto_container_stopped(context: Context) -> None:
from cleveragents.resource.handlers._devcontainer_internals import (
set_lifecycle_tracker,
)
# Use model_copy to bypass transition validation for test setup.
tracker = ContainerLifecycleTracker(
resource_id=context.dcproto_resource.resource_id,
)
tracker = tracker.model_copy(
update={"current_state": ContainerLifecycleState.STOPPED}
)
set_lifecycle_tracker(tracker)
context.dcproto_container_state = ContainerLifecycleState.STOPPED
@given("dcproto a mock sandbox manager that returns a valid sandbox")
def step_dcproto_mock_sandbox_manager(context: Context) -> None:
"""Create a mock SandboxManager that returns a valid sandbox."""
mock_context = MagicMock()
mock_context.sandbox_path = "/tmp/sandbox/test"
mock_sandbox = MagicMock()
mock_sandbox.sandbox_id = "sandbox-001"
mock_sandbox.context = mock_context
mock_manager = MagicMock()
mock_manager.get_sandbox.return_value = None
mock_manager.get_or_create_sandbox.return_value = mock_sandbox
context.dcproto_sandbox_manager = mock_manager
@given("dcproto activate_container is mocked to succeed")
def step_dcproto_mock_activate(context: Context) -> None:
"""Patch activate_container to avoid real subprocess calls."""
context.dcproto_activate_calls: list[tuple[object, ...]] = []
def _fake_activate(resource_id: str, location: str, *, session_id: str) -> None:
context.dcproto_activate_calls.append((resource_id, location, session_id))
patcher = patch(
"cleveragents.resource.handlers.devcontainer.activate_container",
side_effect=_fake_activate,
)
patcher.start()
context.add_cleanup(patcher.stop)
# ── When steps ───────────────────────────────────────────────
def _do_delete(context: Context, path: str) -> None:
"""Shared implementation for delete steps."""
context.dcproto_error = None
context.dcproto_delete_result = None
context.dcproto_subprocess_calls = []
if not hasattr(context, "dcproto_mock_result"):
try:
context.dcproto_delete_result = context.dcproto_handler.delete(
resource=context.dcproto_resource,
path=path,
)
except (ValueError, NotImplementedError) as exc:
context.dcproto_error = exc
return
def _capture_run(*args: object, **kwargs: object) -> object:
context.dcproto_subprocess_calls.append((args, kwargs))
return context.dcproto_mock_result
patcher = patch(
"cleveragents.resource.handlers.devcontainer.subprocess.run",
side_effect=_capture_run,
)
patcher.start()
context.add_cleanup(patcher.stop)
try:
context.dcproto_delete_result = context.dcproto_handler.delete(
resource=context.dcproto_resource,
path=path,
)
except (ValueError, NotImplementedError) as exc:
context.dcproto_error = exc
@when('dcproto I delete path "{path}" from the resource')
def step_dcproto_delete_path(context: Context, path: str) -> None:
_do_delete(context, path)
@when("dcproto I delete with empty path from the resource")
def step_dcproto_delete_empty_path(context: Context) -> None:
_do_delete(context, "")
@when("dcproto I list children of the resource")
def step_dcproto_list_children(context: Context) -> None:
context.dcproto_error = None
context.dcproto_list_result = None
context.dcproto_subprocess_calls = []
if not hasattr(context, "dcproto_mock_result"):
try:
context.dcproto_list_result = context.dcproto_handler.list_children(
resource=context.dcproto_resource,
)
except (ValueError, NotImplementedError) as exc:
context.dcproto_error = exc
return
def _capture_run(*args: object, **kwargs: object) -> object:
context.dcproto_subprocess_calls.append((args, kwargs))
return context.dcproto_mock_result
patcher = patch(
"cleveragents.resource.handlers.devcontainer.subprocess.run",
side_effect=_capture_run,
)
patcher.start()
context.add_cleanup(patcher.stop)
try:
context.dcproto_list_result = context.dcproto_handler.list_children(
resource=context.dcproto_resource,
)
except (ValueError, NotImplementedError) as exc:
context.dcproto_error = exc
@when("dcproto I diff the resource against the temporary directory")
def step_dcproto_diff_against_temp(context: Context) -> None:
context.dcproto_error = None
context.dcproto_diff_result = None
# Patch content_hash to return the configured fake hash for the container
if hasattr(context, "dcproto_container_hash"):
fake_hash = context.dcproto_container_hash
def _fake_content_hash(resource: Resource, *, algorithm: str = "sha256") -> str:
# Only intercept for the resource under test (not the other_location)
if resource.location == context.dcproto_resource.location:
return fake_hash
# Fall through to real implementation for other_location
from cleveragents.resource.handlers._base import BaseResourceHandler
return BaseResourceHandler.content_hash(
context.dcproto_handler, resource, algorithm=algorithm
)
patcher = patch.object(
context.dcproto_handler,
"content_hash",
side_effect=_fake_content_hash,
)
patcher.start()
context.add_cleanup(patcher.stop)
try:
context.dcproto_diff_result = context.dcproto_handler.diff(
resource=context.dcproto_resource,
other_location=context.dcproto_other_location,
)
except (ValueError, NotImplementedError) as exc:
context.dcproto_error = exc
@when('dcproto I diff the resource against "/another/nonexistent/path"')
def step_dcproto_diff_both_absent(context: Context) -> None:
context.dcproto_error = None
context.dcproto_diff_result = None
try:
context.dcproto_diff_result = context.dcproto_handler.diff(
resource=context.dcproto_resource,
other_location="/another/nonexistent/path",
)
except (ValueError, NotImplementedError) as exc:
context.dcproto_error = exc
@when('dcproto I diff the resource against "/some/path"')
def step_dcproto_diff_no_location(context: Context) -> None:
context.dcproto_error = None
context.dcproto_diff_result = None
try:
context.dcproto_diff_result = context.dcproto_handler.diff(
resource=context.dcproto_resource,
other_location="/some/path",
)
except (ValueError, NotImplementedError) as exc:
context.dcproto_error = exc
@when("dcproto I diff the resource against the same location")
def step_dcproto_diff_same_location(context: Context) -> None:
context.dcproto_error = None
context.dcproto_diff_result = None
# Use the resource's own location as other_location
other = context.dcproto_resource.location or "/ws/project"
try:
context.dcproto_diff_result = context.dcproto_handler.diff(
resource=context.dcproto_resource,
other_location=other,
)
except (ValueError, NotImplementedError) as exc:
context.dcproto_error = exc
@when('dcproto I create a sandbox for the resource with plan "{plan_id}"')
def step_dcproto_create_sandbox(context: Context, plan_id: str) -> None:
context.dcproto_error = None
context.dcproto_sandbox_result = None
try:
context.dcproto_sandbox_result = context.dcproto_handler.create_sandbox(
resource=context.dcproto_resource,
plan_id=plan_id,
sandbox_manager=context.dcproto_sandbox_manager,
)
except (ValueError, RuntimeError, NotImplementedError) as exc:
context.dcproto_error = exc
# ── Then steps ───────────────────────────────────────────────
@then("dcproto the delete should succeed")
def step_dcproto_delete_success(context: Context) -> None:
assert context.dcproto_error is None, (
f"Expected no error, got {context.dcproto_error!r}"
)
assert context.dcproto_delete_result is not None, "No delete result returned"
assert context.dcproto_delete_result.success is True, (
f"Expected success, got failure: {context.dcproto_delete_result.message}"
)
@then('dcproto the delete message should contain "{text}"')
def step_dcproto_delete_msg_contains(context: Context, text: str) -> None:
assert context.dcproto_delete_result is not None
assert text in context.dcproto_delete_result.message, (
f"Expected '{text}' in message: {context.dcproto_delete_result.message}"
)
@then("dcproto the delete should fail")
def step_dcproto_delete_failure(context: Context) -> None:
assert context.dcproto_error is None, (
f"Unexpected exception: {context.dcproto_error!r}"
)
assert context.dcproto_delete_result is not None, "No delete result returned"
assert context.dcproto_delete_result.success is False, (
"Expected delete to fail but it succeeded"
)
@then('dcproto the delete failure message should contain "{text}"')
def step_dcproto_delete_fail_msg(context: Context, text: str) -> None:
assert context.dcproto_delete_result is not None
assert context.dcproto_delete_result.success is False
assert text in context.dcproto_delete_result.message, (
f"Expected '{text}' in failure message: {context.dcproto_delete_result.message}"
)
@then('dcproto the delete should raise ValueError containing "{text}"')
def step_dcproto_delete_value_error(context: Context, text: str) -> None:
assert context.dcproto_error is not None, "Expected ValueError"
assert isinstance(context.dcproto_error, ValueError), (
f"Expected ValueError, got {type(context.dcproto_error).__name__}"
)
assert text in str(context.dcproto_error), (
f"Expected '{text}' in error message: {context.dcproto_error}"
)
@then('dcproto the subprocess should have been called with "{cmd}" and "{arg}"')
def step_dcproto_check_subprocess_args(context: Context, cmd: str, arg: str) -> None:
assert len(context.dcproto_subprocess_calls) > 0, "No subprocess calls recorded"
call_args, _call_kwargs = context.dcproto_subprocess_calls[0]
cmd_list = call_args[0]
assert cmd in cmd_list, f"Expected '{cmd}' in command {cmd_list}"
assert arg in cmd_list, f"Expected '{arg}' in command {cmd_list}"
@then("dcproto the children list should have {count:d} entries")
def step_dcproto_children_count(context: Context, count: int) -> None:
assert context.dcproto_error is None, (
f"Expected no error, got {context.dcproto_error!r}"
)
assert context.dcproto_list_result is not None, "No list_children result"
assert len(context.dcproto_list_result) == count, (
f"Expected {count} children, got {len(context.dcproto_list_result)}: "
f"{context.dcproto_list_result}"
)
@then('dcproto the children list should be "{expected_csv}"')
def step_dcproto_children_list_values(context: Context, expected_csv: str) -> None:
expected = expected_csv.split(",")
assert context.dcproto_list_result == expected, (
f"Expected {expected}, got {context.dcproto_list_result}"
)
@then('dcproto the list children should raise ValueError containing "{text}"')
def step_dcproto_list_value_error(context: Context, text: str) -> None:
assert context.dcproto_error is not None, "Expected ValueError"
assert isinstance(context.dcproto_error, ValueError), (
f"Expected ValueError, got {type(context.dcproto_error).__name__}"
)
assert text in str(context.dcproto_error), (
f"Expected '{text}' in error message: {context.dcproto_error}"
)
@then("dcproto the diff should report has_changes true")
def step_dcproto_diff_has_changes_true(context: Context) -> None:
assert context.dcproto_error is None, (
f"Expected no error, got {context.dcproto_error!r}"
)
assert context.dcproto_diff_result is not None, "No diff result returned"
assert context.dcproto_diff_result.has_changes is True, (
"Expected has_changes=True but got False"
)
@then("dcproto the diff should report has_changes false")
def step_dcproto_diff_has_changes_false(context: Context) -> None:
assert context.dcproto_error is None, (
f"Expected no error, got {context.dcproto_error!r}"
)
assert context.dcproto_diff_result is not None, "No diff result returned"
assert context.dcproto_diff_result.has_changes is False, (
"Expected has_changes=False but got True"
)
@then("dcproto the diff files_changed should be {count:d}")
def step_dcproto_diff_files_changed(context: Context, count: int) -> None:
assert context.dcproto_diff_result is not None
assert context.dcproto_diff_result.files_changed == count, (
f"Expected files_changed={count}, got {context.dcproto_diff_result.files_changed}"
)
@then("dcproto the diff unified_diff should be empty")
def step_dcproto_diff_unified_empty(context: Context) -> None:
assert context.dcproto_diff_result is not None
assert context.dcproto_diff_result.unified_diff == "", (
f"Expected empty unified_diff, got: {context.dcproto_diff_result.unified_diff!r}"
)
@then('dcproto the diff should raise ValueError containing "{text}"')
def step_dcproto_diff_value_error(context: Context, text: str) -> None:
assert context.dcproto_error is not None, "Expected ValueError"
assert isinstance(context.dcproto_error, ValueError), (
f"Expected ValueError, got {type(context.dcproto_error).__name__}"
)
assert text in str(context.dcproto_error), (
f"Expected '{text}' in error message: {context.dcproto_error}"
)
@then('dcproto the sandbox result should have strategy "{strategy}"')
def step_dcproto_sandbox_strategy(context: Context, strategy: str) -> None:
assert context.dcproto_error is None, (
f"Expected no error, got {context.dcproto_error!r}"
)
assert context.dcproto_sandbox_result is not None, "No sandbox result returned"
assert context.dcproto_sandbox_result.strategy == strategy, (
f"Expected strategy '{strategy}', got '{context.dcproto_sandbox_result.strategy}'"
)
@then("dcproto the sandbox result should have a sandbox path")
def step_dcproto_sandbox_has_path(context: Context) -> None:
assert context.dcproto_sandbox_result is not None
assert context.dcproto_sandbox_result.sandbox_path, (
"Expected a non-empty sandbox_path"
)
@then("dcproto activate_container should have been called")
def step_dcproto_activate_was_called(context: Context) -> None:
assert context.dcproto_error is None, (
f"Expected no error, got {context.dcproto_error!r}"
)
assert hasattr(context, "dcproto_activate_calls"), (
"activate_container mock was not set up"
)
assert len(context.dcproto_activate_calls) > 0, (
"Expected activate_container to be called but it was not"
)
@then('dcproto the create_sandbox should raise ValueError containing "{text}"')
def step_dcproto_sandbox_value_error(context: Context, text: str) -> None:
assert context.dcproto_error is not None, "Expected ValueError"
assert isinstance(context.dcproto_error, ValueError), (
f"Expected ValueError, got {type(context.dcproto_error).__name__}"
)
assert text in str(context.dcproto_error), (
f"Expected '{text}' in error message: {context.dcproto_error}"
)
@@ -89,7 +89,13 @@ from cleveragents.resource.handlers.devcontainer_lifecycle import (
rebuild_container,
stop_container,
)
from cleveragents.resource.handlers.protocol import Content, WriteResult
from cleveragents.resource.handlers.protocol import (
Content,
DeleteResult,
DiffResult,
SandboxResult,
WriteResult,
)
from cleveragents.tool.context import BoundResource
logger = logging.getLogger(__name__)
@@ -483,3 +489,200 @@ class DevcontainerHandler(BaseResourceHandler):
)
data = result.stdout if result.returncode == 0 else b""
return Content(data=data, encoding="utf-8")
# -- Missing protocol methods (issue #1242) ----------------------------
def delete(self, *, resource: Resource, path: str = "") -> DeleteResult:
"""Delete a file inside a running devcontainer via ``devcontainer exec rm``.
If *path* is empty, the workspace root is targeted (``rm -rf .``).
The container must be reachable; a missing or stopped container
results in a failure :class:`DeleteResult` rather than an exception.
Args:
resource: The devcontainer resource.
path: Path inside the container to delete. Empty string
targets the workspace root.
Returns:
A :class:`DeleteResult` indicating success or failure.
Raises:
ValueError: If the resource has no location.
"""
location = self._require_location(resource)
target = path if path else "."
result = subprocess.run(
[
"devcontainer",
"exec",
"--workspace-folder",
location,
"rm",
"-rf",
target,
],
capture_output=True,
timeout=30,
check=False,
)
if result.returncode != 0:
stderr = result.stderr.decode(errors="replace").strip()
return DeleteResult(
success=False,
message=f"Delete failed in devcontainer at '{location}': {stderr}",
)
return DeleteResult(
success=True,
message=f"Deleted '{target}' in devcontainer at '{location}'",
)
def list_children(self, *, resource: Resource) -> list[str]:
"""List direct children of the workspace inside a running devcontainer.
Uses ``devcontainer exec ls -1`` to enumerate entries at the
workspace root. Returns an empty list if the container is
missing or stopped rather than raising an exception.
Args:
resource: The devcontainer resource.
Returns:
Sorted list of entry names (files and directories) at the
workspace root. Empty list on container failure.
Raises:
ValueError: If the resource has no location.
"""
location = self._require_location(resource)
result = subprocess.run(
[
"devcontainer",
"exec",
"--workspace-folder",
location,
"ls",
"-1",
],
capture_output=True,
text=True,
timeout=30,
check=False,
)
if result.returncode != 0:
return []
names: list[str] = []
for name in result.stdout.strip().split("\n"):
name = name.strip()
if name:
names.append(name)
return sorted(names)
def diff(self, *, resource: Resource, other_location: str) -> DiffResult:
"""Compare the devcontainer workspace against another location.
Computes the content hash of this resource and the content hash
of the *other_location* (using the base-class filesystem hash)
and reports whether they differ. A full unified diff is not
produced because the container filesystem is not directly
accessible from the host; instead ``has_changes`` is set based
on hash comparison and ``unified_diff`` is left empty.
If the container is missing or stopped, the exec hash falls back
to the config-file hash (see :meth:`content_hash`).
Args:
resource: The devcontainer resource.
other_location: Filesystem path to compare against.
Returns:
A :class:`DiffResult` with ``has_changes`` set.
Raises:
ValueError: If the resource has no location.
"""
self._require_location(resource)
this_hash = self.content_hash(resource)
# Hash the other location using the base-class filesystem logic.
# We construct a temporary resource pointing at other_location and
# call the base-class content_hash directly (bypassing the
# devcontainer override which would try to exec into the container).
from cleveragents.resource.handlers._base import (
EMPTY_CONTENT_HASH,
BaseResourceHandler,
)
other_resource = resource.__class__(
resource_id=resource.resource_id,
name=resource.name,
resource_type_name=resource.resource_type_name,
classification=resource.classification,
description=resource.description,
location=other_location,
)
other_hash = BaseResourceHandler.content_hash(self, other_resource)
has_changes = this_hash != other_hash
# Both hashes being EMPTY_CONTENT_HASH means both sides are absent —
# treat as no changes.
if this_hash == EMPTY_CONTENT_HASH and other_hash == EMPTY_CONTENT_HASH:
has_changes = False
return DiffResult(
has_changes=has_changes,
unified_diff="",
files_changed=1 if has_changes else 0,
)
def create_sandbox(
self,
*,
resource: Resource,
plan_id: str,
sandbox_manager: SandboxManager,
) -> SandboxResult:
"""Create an isolated sandbox for a devcontainer resource.
For devcontainer resources the container itself provides
isolation (``none`` sandbox strategy — F22/F25 fix). This
method delegates to the base-class implementation which calls
:meth:`SandboxManager.get_or_create_sandbox` with the ``none``
strategy.
If the container is in an activatable state (DETECTED, STOPPED,
FAILED) it is lazily activated before the sandbox is created so
that the workspace path is available.
Args:
resource: The devcontainer resource.
plan_id: The plan requesting the sandbox.
sandbox_manager: The sandbox lifecycle manager.
Returns:
A :class:`SandboxResult` with the sandbox path and strategy.
Raises:
ValueError: If the resource has no location.
RuntimeError: If the sandbox was created but has no context.
"""
if resource.resource_type_name == "devcontainer-instance" and resource.location:
tracker = get_lifecycle_tracker(resource.resource_id)
if tracker.current_state in self._ACTIVATABLE_STATES:
logger.info(
"Lazy-activating devcontainer %s before sandbox creation"
" (state=%s)",
resource.resource_id,
tracker.current_state.value,
)
activate_container(
resource.resource_id,
resource.location,
session_id=plan_id,
)
return super().create_sandbox(
resource=resource,
plan_id=plan_id,
sandbox_manager=sandbox_manager,
)