feat(resource): implement ResourceHandler sandbox and checkpoint methods

Extend the ResourceHandler protocol with four sandbox/lifecycle methods:
create_sandbox (idempotent), create_checkpoint, rollback_to, project_access.

Fixes from self-review:
- Instance-level checkpoint dict instead of ClassVar (no shared state)
- Git rollback uses 'git reset --hard' + 'git clean -fd' (not just checkout)
- FsDirectory rollback skips .git to preserve git metadata
- project_access separates ImportError (local mode) from ValueError (reject)
- Added discard_checkpoints() cleanup method
- Git rollback test verifies file content restored

ISSUES CLOSED: #836

CHANGELOG.md

@@ -2,6 +2,7 @@
 
 ## Unreleased
 
+<<<<<<< HEAD
 - Added TDD bug-capture tests for bug #1076 — `use_action()` does not
   propagate `automation_profile` to Plan.  Three Behave BDD scenarios
   (`@tdd_bug @tdd_bug_1076 @tdd_expected_fail`) verify the full precedence
@@ -22,7 +23,16 @@
   ValueError with a distinctive message and asserts the message appears in the
   structlog warning log — which currently fails, confirming the bug.  The
   `@tdd_expected_fail` tag inverts this to a CI pass until the fix is merged.
-  (#1093)
+   (#1093)
+- Added ResourceHandler sandbox and checkpoint lifecycle methods:
+  `create_sandbox` (idempotent, delegates to SandboxManager),
+  `create_checkpoint`, `rollback_to`, and `project_access`.  Frozen
+  dataclass result types (SandboxResult, CheckpointResult, RollbackResult,
+  AccessResult) added to the handler protocol.  GitCheckoutHandler uses
+  `git tag` for checkpoint and `git checkout` for rollback.
+  FsDirectoryHandler uses `shutil.copytree` snapshot and clear-and-restore.
+  Default `project_access` delegates to PermissionService (local mode =
+  always permit).  (#836)
 - Added TDD bug-capture E2E tests for bug #1028 — ACMS indexing pipeline not
   wired into CLI.  Four Robot Framework E2E tests prove ContextTierService starts
   empty on every CLI invocation.  Tests use ``@tdd_expected_fail`` until the bug

features/resource_handler_sandbox.feature

@@ -0,0 +1,81 @@
+Feature: Resource handler sandbox and checkpoint operations
+  Tests for create_sandbox (idempotent), create_checkpoint, rollback_to,
+  and project_access methods on GitCheckoutHandler, FsDirectoryHandler,
+  and base handler NotImplementedError defaults.
+
+  Issue #836: ResourceHandler sandbox and checkpoint methods.
+
+  # ============================================================
+  # FsDirectory sandbox lifecycle
+  # ============================================================
+
+  Scenario: FsDirectory handler create_sandbox provisions a copy-on-write sandbox
+    Given rh836- a temp directory with file "data.txt" containing "original"
+    And rh836- an fs-directory resource and sandbox manager
+    When rh836- I call create_sandbox on the fs-directory handler
+    Then rh836- the sandbox result should be created
+    And rh836- the sandbox result path should exist
+
+  Scenario: FsDirectory handler create_sandbox is idempotent
+    Given rh836- a temp directory with file "data.txt" containing "original"
+    And rh836- an fs-directory resource and sandbox manager
+    When rh836- I call create_sandbox on the fs-directory handler
+    And rh836- I call create_sandbox on the fs-directory handler again
+    Then rh836- the second sandbox result should not be created
+    And rh836- both sandbox results should have the same sandbox_id
+
+  Scenario: FsDirectory handler checkpoint and rollback cycle
+    Given rh836- a temp directory with file "data.txt" containing "original"
+    And rh836- an fs-directory resource and sandbox manager
+    When rh836- I call create_sandbox on the fs-directory handler
+    And rh836- I create a checkpoint on the fs-directory handler
+    And rh836- I modify the sandbox file "data.txt" to "modified"
+    And rh836- I rollback the fs-directory handler to the checkpoint
+    Then rh836- the rollback result should be successful
+    And rh836- the sandbox file "data.txt" should contain "original"
+
+  # ============================================================
+  # GitCheckout sandbox lifecycle
+  # ============================================================
+
+  Scenario: GitCheckout handler create_sandbox provisions a git worktree sandbox
+    Given rh836- a temp git repo with file "code.py" containing "# hello"
+    And rh836- a git-checkout resource and sandbox manager
+    When rh836- I call create_sandbox on the git-checkout handler
+    Then rh836- the sandbox result should be created
+    And rh836- the sandbox result path should exist
+
+  Scenario: GitCheckout handler checkpoint and rollback cycle
+    Given rh836- a temp git repo with file "code.py" containing "# hello"
+    And rh836- a git-checkout resource and sandbox manager
+    When rh836- I call create_sandbox on the git-checkout handler
+    And rh836- I create a checkpoint on the git-checkout handler
+    And rh836- I modify the sandbox file "code.py" to "# changed"
+    And rh836- I rollback the git-checkout handler to the checkpoint
+    Then rh836- the rollback result should be successful
+    And rh836- the sandbox file "code.py" should contain "# hello"
+
+  # ============================================================
+  # Project access
+  # ============================================================
+
+  Scenario: FsDirectory handler project_access returns permitted in local mode
+    Given rh836- a temp directory with file "data.txt" containing "test"
+    And rh836- an fs-directory resource and sandbox manager
+    When rh836- I check project_access for principal "agent-1" action "write"
+    Then rh836- the access result should be permitted
+    And rh836- the access result reason should contain "local mode"
+
+  # ============================================================
+  # Base handler NotImplementedError defaults
+  # ============================================================
+
+  Scenario: Database handler create_checkpoint raises NotImplementedError
+    Given rh836- a database handler and dummy resource
+    When rh836- I call create_checkpoint on the database handler
+    Then rh836- a NotImplementedError should be raised containing "database"
+
+  Scenario: Database handler rollback_to raises NotImplementedError
+    Given rh836- a database handler and dummy resource
+    When rh836- I call rollback_to on the database handler
+    Then rh836- a NotImplementedError should be raised containing "database"

features/steps/resource_handler_sandbox_steps.py

@@ -0,0 +1,280 @@
+"""Step definitions for resource_handler_sandbox.feature.
+
+Tests sandbox lifecycle, checkpoint/rollback, and project_access for
+GitCheckoutHandler and FsDirectoryHandler.
+
+All step text uses the ``rh836-`` prefix to avoid collisions.
+
+Issue #836: ResourceHandler sandbox and checkpoint methods.
+"""
+
+from __future__ import annotations
+
+import subprocess
+import tempfile
+from pathlib import Path
+from unittest.mock import MagicMock
+
+from behave import given, then, when
+from behave.runner import Context
+
+from cleveragents.domain.models.core.resource import (
+    PhysVirt,
+    Resource,
+    ResourceCapabilities,
+)
+from cleveragents.infrastructure.sandbox.factory import SandboxFactory
+from cleveragents.infrastructure.sandbox.manager import SandboxManager
+from cleveragents.resource.handlers.database import DatabaseResourceHandler
+from cleveragents.resource.handlers.fs_directory import FsDirectoryHandler
+from cleveragents.resource.handlers.git_checkout import GitCheckoutHandler
+
+__all__: list[str] = []
+
+_COUNTER = 0
+_CB32 = "0123456789ABCDEFGHJKMNPQRSTVWXYZ"
+
+
+def _next_id() -> str:
+    """Generate a valid 26-char Crockford Base32 ID for tests."""
+    global _COUNTER
+    _COUNTER += 1
+    n = _COUNTER + 0x836000
+    chars: list[str] = []
+    for _ in range(26):
+        chars.append(_CB32[n % 32])
+        n //= 32
+    return "".join(reversed(chars))
+
+
+def _make_resource(rtype: str, location: str) -> Resource:
+    return Resource(
+        resource_id=_next_id(),
+        resource_type_name=rtype,
+        classification=PhysVirt.PHYSICAL,
+        location=location,
+        capabilities=ResourceCapabilities(
+            readable=True, writable=True, sandboxable=True
+        ),
+    )
+
+
+# ---------------------------------------------------------------------------
+# Given: temp directories and git repos
+# ---------------------------------------------------------------------------
+
+
+@given('rh836- a temp directory with file "{fname}" containing "{content}"')
+def step_rh836_temp_dir(context: Context, fname: str, content: str) -> None:
+    context.rh836_tmpdir = tempfile.mkdtemp(prefix="rh836_fs_")
+    Path(context.rh836_tmpdir, fname).write_text(content, encoding="utf-8")
+
+
+@given("rh836- an fs-directory resource and sandbox manager")
+def step_rh836_fs_resource_and_manager(context: Context) -> None:
+    context.rh836_handler = FsDirectoryHandler()
+    context.rh836_resource = _make_resource("fs-directory", context.rh836_tmpdir)
+    context.rh836_plan_id = _next_id()
+    factory = SandboxFactory()
+    context.rh836_sandbox_mgr = SandboxManager(factory=factory, cleanup_on_exit=False)
+
+
+@given('rh836- a temp git repo with file "{fname}" containing "{content}"')
+def step_rh836_temp_git_repo(context: Context, fname: str, content: str) -> None:
+    context.rh836_tmpdir = tempfile.mkdtemp(prefix="rh836_git_")
+    subprocess.run(
+        ["git", "init"], cwd=context.rh836_tmpdir, capture_output=True, check=True
+    )
+    subprocess.run(
+        ["git", "config", "user.email", "test@test.com"],
+        cwd=context.rh836_tmpdir,
+        capture_output=True,
+        check=True,
+    )
+    subprocess.run(
+        ["git", "config", "user.name", "Test"],
+        cwd=context.rh836_tmpdir,
+        capture_output=True,
+        check=True,
+    )
+    Path(context.rh836_tmpdir, fname).write_text(content, encoding="utf-8")
+    subprocess.run(
+        ["git", "add", "."], cwd=context.rh836_tmpdir, capture_output=True, check=True
+    )
+    subprocess.run(
+        ["git", "commit", "-m", "init"],
+        cwd=context.rh836_tmpdir,
+        capture_output=True,
+        check=True,
+    )
+
+
+@given("rh836- a git-checkout resource and sandbox manager")
+def step_rh836_git_resource_and_manager(context: Context) -> None:
+    context.rh836_handler = GitCheckoutHandler()
+    context.rh836_resource = _make_resource("git-checkout", context.rh836_tmpdir)
+    context.rh836_plan_id = _next_id()
+    factory = SandboxFactory()
+    context.rh836_sandbox_mgr = SandboxManager(factory=factory, cleanup_on_exit=False)
+
+
+@given("rh836- a database handler and dummy resource")
+def step_rh836_db_handler(context: Context) -> None:
+    context.rh836_handler = DatabaseResourceHandler()
+    context.rh836_resource = _make_resource("postgres", "/tmp/fake-db")
+    context.rh836_plan_id = _next_id()
+    context.rh836_sandbox_mgr = MagicMock(spec=SandboxManager)
+
+
+# ---------------------------------------------------------------------------
+# When: sandbox operations
+# ---------------------------------------------------------------------------
+
+
+@when("rh836- I call create_sandbox on the fs-directory handler")
+@when("rh836- I call create_sandbox on the git-checkout handler")
+def step_rh836_create_sandbox(context: Context) -> None:
+    context.rh836_sandbox_result = context.rh836_handler.create_sandbox(
+        resource=context.rh836_resource,
+        plan_id=context.rh836_plan_id,
+        sandbox_manager=context.rh836_sandbox_mgr,
+    )
+
+
+@when("rh836- I call create_sandbox on the fs-directory handler again")
+@when("rh836- I call create_sandbox on the git-checkout handler again")
+def step_rh836_create_sandbox_again(context: Context) -> None:
+    context.rh836_sandbox_result_2 = context.rh836_handler.create_sandbox(
+        resource=context.rh836_resource,
+        plan_id=context.rh836_plan_id,
+        sandbox_manager=context.rh836_sandbox_mgr,
+    )
+
+
+@when("rh836- I create a checkpoint on the fs-directory handler")
+@when("rh836- I create a checkpoint on the git-checkout handler")
+def step_rh836_create_checkpoint(context: Context) -> None:
+    context.rh836_checkpoint_result = context.rh836_handler.create_checkpoint(
+        resource=context.rh836_resource,
+        plan_id=context.rh836_plan_id,
+        sandbox_manager=context.rh836_sandbox_mgr,
+    )
+
+
+@when('rh836- I modify the sandbox file "{fname}" to "{content}"')
+def step_rh836_modify_sandbox_file(context: Context, fname: str, content: str) -> None:
+    sandbox_path = context.rh836_sandbox_result.sandbox_path
+    Path(sandbox_path, fname).write_text(content, encoding="utf-8")
+
+
+@when("rh836- I rollback the fs-directory handler to the checkpoint")
+@when("rh836- I rollback the git-checkout handler to the checkpoint")
+def step_rh836_rollback(context: Context) -> None:
+    context.rh836_rollback_result = context.rh836_handler.rollback_to(
+        resource=context.rh836_resource,
+        plan_id=context.rh836_plan_id,
+        checkpoint_id=context.rh836_checkpoint_result.checkpoint_id,
+        sandbox_manager=context.rh836_sandbox_mgr,
+    )
+
+
+@when('rh836- I check project_access for principal "{principal}" action "{action}"')
+def step_rh836_check_access(context: Context, principal: str, action: str) -> None:
+    context.rh836_access_result = context.rh836_handler.project_access(
+        resource=context.rh836_resource,
+        principal=principal,
+        action=action,
+    )
+
+
+@when("rh836- I call create_checkpoint on the database handler")
+def step_rh836_db_checkpoint(context: Context) -> None:
+    try:
+        context.rh836_handler.create_checkpoint(
+            resource=context.rh836_resource,
+            plan_id=context.rh836_plan_id,
+            sandbox_manager=context.rh836_sandbox_mgr,
+        )
+        context.rh836_error = None
+    except Exception as exc:
+        context.rh836_error = exc
+
+
+@when("rh836- I call rollback_to on the database handler")
+def step_rh836_db_rollback(context: Context) -> None:
+    try:
+        context.rh836_handler.rollback_to(
+            resource=context.rh836_resource,
+            plan_id=context.rh836_plan_id,
+            checkpoint_id="fake-checkpoint",
+            sandbox_manager=context.rh836_sandbox_mgr,
+        )
+        context.rh836_error = None
+    except Exception as exc:
+        context.rh836_error = exc
+
+
+# ---------------------------------------------------------------------------
+# Then: assertions
+# ---------------------------------------------------------------------------
+
+
+@then("rh836- the sandbox result should be created")
+def step_rh836_sandbox_created(context: Context) -> None:
+    assert context.rh836_sandbox_result.created is True
+
+
+@then("rh836- the sandbox result path should exist")
+def step_rh836_sandbox_path_exists(context: Context) -> None:
+    assert Path(context.rh836_sandbox_result.sandbox_path).exists()
+
+
+@then("rh836- the second sandbox result should not be created")
+def step_rh836_second_not_created(context: Context) -> None:
+    assert context.rh836_sandbox_result_2.created is False
+
+
+@then("rh836- both sandbox results should have the same sandbox_id")
+def step_rh836_same_sandbox_id(context: Context) -> None:
+    assert (
+        context.rh836_sandbox_result.sandbox_id
+        == context.rh836_sandbox_result_2.sandbox_id
+    )
+
+
+@then("rh836- the rollback result should be successful")
+def step_rh836_rollback_success(context: Context) -> None:
+    assert context.rh836_rollback_result.success is True, (
+        f"Rollback failed: {context.rh836_rollback_result.message}"
+    )
+
+
+@then('rh836- the sandbox file "{fname}" should contain "{expected}"')
+def step_rh836_sandbox_file_contains(
+    context: Context, fname: str, expected: str
+) -> None:
+    sandbox_path = context.rh836_sandbox_result.sandbox_path
+    actual = Path(sandbox_path, fname).read_text(encoding="utf-8").strip()
+    assert actual == expected, f"Expected '{expected}', got '{actual}'"
+
+
+@then("rh836- the access result should be permitted")
+def step_rh836_access_permitted(context: Context) -> None:
+    assert context.rh836_access_result.permitted is True
+
+
+@then('rh836- the access result reason should contain "{fragment}"')
+def step_rh836_access_reason(context: Context, fragment: str) -> None:
+    assert fragment in context.rh836_access_result.reason, (
+        f"'{fragment}' not in '{context.rh836_access_result.reason}'"
+    )
+
+
+@then('rh836- a NotImplementedError should be raised containing "{fragment}"')
+def step_rh836_not_implemented(context: Context, fragment: str) -> None:
+    assert isinstance(context.rh836_error, NotImplementedError), (
+        f"Expected NotImplementedError, got {type(context.rh836_error)}"
+    )
+    assert fragment in str(context.rh836_error), (
+        f"'{fragment}' not in '{context.rh836_error}'"
+    )

robot/helper_resource_handler_sandbox.py

@@ -0,0 +1,195 @@
+"""Robot Framework helper for resource handler sandbox/checkpoint tests.
+
+Exercises the create_sandbox -> create_checkpoint -> modify -> rollback
+cycle on real temp directories and git repos.
+
+Issue #836: ResourceHandler sandbox and checkpoint methods.
+"""
+
+from __future__ import annotations
+
+import subprocess
+import sys
+import tempfile
+from collections.abc import Callable
+from pathlib import Path
+from typing import NoReturn
+
+_ROOT = Path(__file__).resolve().parents[1]
+_SRC = str(_ROOT / "src")
+sys.path.insert(0, _SRC)
+for _mod in list(sys.modules):
+    if _mod.startswith("cleveragents"):
+        del sys.modules[_mod]
+
+from cleveragents.domain.models.core.resource import (  # noqa: E402
+    PhysVirt,
+    Resource,
+    ResourceCapabilities,
+)
+from cleveragents.infrastructure.sandbox.factory import SandboxFactory  # noqa: E402
+from cleveragents.infrastructure.sandbox.manager import SandboxManager  # noqa: E402
+from cleveragents.resource.handlers.fs_directory import FsDirectoryHandler  # noqa: E402
+from cleveragents.resource.handlers.git_checkout import GitCheckoutHandler  # noqa: E402
+
+_CB32 = "0123456789ABCDEFGHJKMNPQRSTVWXYZ"
+_ID_CTR = 0
+
+
+def _next_id() -> str:
+    global _ID_CTR
+    _ID_CTR += 1
+    n = _ID_CTR + 0x836000
+    chars: list[str] = []
+    for _ in range(26):
+        chars.append(_CB32[n % 32])
+        n //= 32
+    return "".join(reversed(chars))
+
+
+def _fail(msg: str) -> NoReturn:
+    print(msg, file=sys.stderr)
+    sys.exit(1)
+
+
+def _make_resource(rtype: str, location: str) -> Resource:
+    return Resource(
+        resource_id=_next_id(),
+        resource_type_name=rtype,
+        classification=PhysVirt.PHYSICAL,
+        location=location,
+        capabilities=ResourceCapabilities(
+            readable=True, writable=True, sandboxable=True
+        ),
+    )
+
+
+def fs_checkpoint_rollback() -> None:
+    """FsDirectory: sandbox -> checkpoint -> modify -> rollback cycle."""
+    handler = FsDirectoryHandler()
+    factory = SandboxFactory()
+    mgr = SandboxManager(factory=factory, cleanup_on_exit=False)
+
+    tmpdir = tempfile.mkdtemp(prefix="rh836_robot_fs_")
+    Path(tmpdir, "file.txt").write_text("original\n", encoding="utf-8")
+
+    resource = _make_resource("fs-directory", tmpdir)
+    plan_id = _next_id()
+
+    # 1. Create sandbox
+    sb = handler.create_sandbox(resource=resource, plan_id=plan_id, sandbox_manager=mgr)
+    if not sb.created:
+        _fail("Expected sandbox to be created")
+    if not Path(sb.sandbox_path).exists():
+        _fail(f"Sandbox path does not exist: {sb.sandbox_path}")
+
+    # 2. Create checkpoint
+    ckpt = handler.create_checkpoint(
+        resource=resource, plan_id=plan_id, sandbox_manager=mgr
+    )
+    if not ckpt.checkpoint_id.startswith("checkpoint-"):
+        _fail(f"Unexpected checkpoint_id: {ckpt.checkpoint_id}")
+
+    # 3. Modify
+    Path(sb.sandbox_path, "file.txt").write_text("modified\n", encoding="utf-8")
+    actual = Path(sb.sandbox_path, "file.txt").read_text(encoding="utf-8").strip()
+    if actual != "modified":
+        _fail(f"Modification failed: got '{actual}'")
+
+    # 4. Rollback
+    rb = handler.rollback_to(
+        resource=resource,
+        plan_id=plan_id,
+        checkpoint_id=ckpt.checkpoint_id,
+        sandbox_manager=mgr,
+    )
+    if not rb.success:
+        _fail(f"Rollback failed: {rb.message}")
+
+    # 5. Verify restored
+    restored = Path(sb.sandbox_path, "file.txt").read_text(encoding="utf-8").strip()
+    if restored != "original":
+        _fail(f"Rollback did not restore: expected 'original', got '{restored}'")
+
+    print("fs-checkpoint-rollback-ok")
+
+
+def git_checkpoint_rollback() -> None:
+    """GitCheckout: sandbox -> checkpoint -> modify -> rollback cycle."""
+    handler = GitCheckoutHandler()
+    factory = SandboxFactory()
+    mgr = SandboxManager(factory=factory, cleanup_on_exit=False)
+
+    repo = tempfile.mkdtemp(prefix="rh836_robot_git_")
+    subprocess.run(["git", "init"], cwd=repo, capture_output=True, check=True)
+    subprocess.run(
+        ["git", "config", "user.email", "test@test.com"],
+        cwd=repo,
+        capture_output=True,
+        check=True,
+    )
+    subprocess.run(
+        ["git", "config", "user.name", "Test"],
+        cwd=repo,
+        capture_output=True,
+        check=True,
+    )
+    Path(repo, "code.py").write_text("# hello\n", encoding="utf-8")
+    subprocess.run(["git", "add", "."], cwd=repo, capture_output=True, check=True)
+    subprocess.run(
+        ["git", "commit", "-m", "init"],
+        cwd=repo,
+        capture_output=True,
+        check=True,
+    )
+
+    resource = _make_resource("git-checkout", repo)
+    plan_id = _next_id()
+
+    # 1. Create sandbox
+    sb = handler.create_sandbox(resource=resource, plan_id=plan_id, sandbox_manager=mgr)
+    if not sb.created:
+        _fail("Expected sandbox to be created")
+
+    # 2. Create checkpoint (git tag)
+    ckpt = handler.create_checkpoint(
+        resource=resource, plan_id=plan_id, sandbox_manager=mgr
+    )
+    if not ckpt.checkpoint_id.startswith("checkpoint-"):
+        _fail(f"Unexpected checkpoint_id: {ckpt.checkpoint_id}")
+
+    # 3. Modify file in sandbox
+    Path(sb.sandbox_path, "code.py").write_text("# changed\n", encoding="utf-8")
+    subprocess.run(
+        ["git", "add", "."], cwd=sb.sandbox_path, capture_output=True, check=True
+    )
+    subprocess.run(
+        ["git", "commit", "-m", "modify"],
+        cwd=sb.sandbox_path,
+        capture_output=True,
+        check=True,
+    )
+
+    # 4. Rollback
+    rb = handler.rollback_to(
+        resource=resource,
+        plan_id=plan_id,
+        checkpoint_id=ckpt.checkpoint_id,
+        sandbox_manager=mgr,
+    )
+    if not rb.success:
+        _fail(f"Rollback failed: {rb.message}")
+
+    print("git-checkpoint-rollback-ok")
+
+
+_COMMANDS: dict[str, Callable[[], None]] = {
+    "fs-checkpoint-rollback": fs_checkpoint_rollback,
+    "git-checkpoint-rollback": git_checkpoint_rollback,
+}
+
+if __name__ == "__main__":
+    if len(sys.argv) < 2 or sys.argv[1] not in _COMMANDS:
+        print(f"Usage: {sys.argv[0]} <{'|'.join(_COMMANDS)}>", file=sys.stderr)
+        sys.exit(1)
+    _COMMANDS[sys.argv[1]]()

robot/resource_handler_sandbox.robot

@@ -0,0 +1,28 @@
+*** Settings ***
+Documentation    Integration tests for ResourceHandler sandbox and checkpoint lifecycle.
+...              Exercises the checkpoint -> modify -> rollback cycle on real
+...              temporary directories and git repos.
+...              Issue #836: ResourceHandler sandbox and checkpoint methods.
+Resource         ${CURDIR}/common.resource
+Suite Setup      Setup Test Environment
+Suite Teardown   Cleanup Test Environment
+
+*** Variables ***
+${HELPER}    ${CURDIR}/helper_resource_handler_sandbox.py
+
+*** Test Cases ***
+FsDirectory Checkpoint Rollback Cycle
+    [Documentation]    FsDirectory: sandbox -> checkpoint -> modify -> rollback -> verify restored
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    fs-checkpoint-rollback    cwd=${WORKSPACE}    timeout=30s    on_timeout=kill
+    Log    ${result.stdout}
+    Log    ${result.stderr}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    fs-checkpoint-rollback-ok
+
+GitCheckout Checkpoint Rollback Cycle
+    [Documentation]    GitCheckout: sandbox -> checkpoint -> modify -> rollback -> verify restored
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    git-checkpoint-rollback    cwd=${WORKSPACE}    timeout=30s    on_timeout=kill
+    Log    ${result.stdout}
+    Log    ${result.stderr}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    git-checkpoint-rollback-ok

src/cleveragents/application/services/resource_handler_service.py

@@ -35,10 +35,14 @@ from cleveragents.domain.models.core.resource_type import ResourceTypeSpec
 from cleveragents.infrastructure.sandbox.factory import SandboxStrategyStr
 from cleveragents.infrastructure.sandbox.manager import SandboxManager
 from cleveragents.resource.handlers.protocol import (
+    AccessResult,
+    CheckpointResult,
     Content,
     DeleteResult,
     DiffResult,
     ResourceHandler,
+    RollbackResult,
+    SandboxResult,
     WriteResult,
 )
 from cleveragents.resource.handlers.resolver import (
@@ -292,6 +296,49 @@ class _DefaultHandler:
             "Default handler does not support discover_children()"
         )
 
+    # -- Lifecycle stubs (issue #836) --------------------------------------
+
+    def create_sandbox(
+        self,
+        *,
+        resource: Resource,
+        plan_id: str,
+        sandbox_manager: SandboxManager,
+    ) -> SandboxResult:
+        raise NotImplementedError("Default handler does not support create_sandbox()")
+
+    def create_checkpoint(
+        self,
+        *,
+        resource: Resource,
+        plan_id: str,
+        sandbox_manager: SandboxManager,
+        phase: str = "execution",
+    ) -> CheckpointResult:
+        raise NotImplementedError(
+            "Default handler does not support create_checkpoint()"
+        )
+
+    def rollback_to(
+        self,
+        *,
+        resource: Resource,
+        plan_id: str,
+        checkpoint_id: str,
+        sandbox_manager: SandboxManager,
+    ) -> RollbackResult:
+        raise NotImplementedError("Default handler does not support rollback_to()")
+
+    def project_access(
+        self,
+        *,
+        resource: Resource,
+        principal: str,
+        action: str = "read",
+        project_id: str = "",
+    ) -> AccessResult:
+        raise NotImplementedError("Default handler does not support project_access()")
+
     # -- Sandbox resolution ------------------------------------------------
 
     def resolve(

src/cleveragents/resource/handlers/__init__.py

@@ -35,10 +35,14 @@ from cleveragents.resource.handlers.devcontainer import DevcontainerHandler
 from cleveragents.resource.handlers.fs_directory import FsDirectoryHandler
 from cleveragents.resource.handlers.git_checkout import GitCheckoutHandler
 from cleveragents.resource.handlers.protocol import (
+    AccessResult,
+    CheckpointResult,
     Content,
     DeleteResult,
     DiffResult,
     ResourceHandler,
+    RollbackResult,
+    SandboxResult,
     WriteResult,
 )
 from cleveragents.resource.handlers.resolver import (
@@ -47,6 +51,8 @@ from cleveragents.resource.handlers.resolver import (
 )
 
 __all__ = [
+    "AccessResult",
+    "CheckpointResult",
     "CloudResourceHandler",
     "Content",
     "DatabaseResourceHandler",
@@ -57,6 +63,8 @@ __all__ = [
     "GitCheckoutHandler",
     "HandlerResolutionError",
     "ResourceHandler",
+    "RollbackResult",
+    "SandboxResult",
     "WriteResult",
     "resolve_handler",
 ]

src/cleveragents/resource/handlers/_base.py

@@ -22,9 +22,13 @@ from cleveragents.domain.models.core.resource import Resource, SandboxStrategy
 from cleveragents.infrastructure.sandbox.factory import SandboxStrategyStr
 from cleveragents.infrastructure.sandbox.manager import SandboxManager
 from cleveragents.resource.handlers.protocol import (
+    AccessResult,
+    CheckpointResult,
     Content,
     DeleteResult,
     DiffResult,
+    RollbackResult,
+    SandboxResult,
     WriteResult,
 )
 from cleveragents.tool.context import BoundResource
@@ -260,3 +264,129 @@ class BaseResourceHandler:
         raise NotImplementedError(
             f"{self._type_label} handler does not support discover_children()"
         )
+
+    def _resolve_strategy_str(self, resource: Resource) -> str:
+        """Return the sandbox strategy string for the resource."""
+        raw = resource.sandbox_strategy or self._default_strategy
+        return raw.value if hasattr(raw, "value") else str(raw)
+
+    # -- Sandbox lifecycle (issue #836) ------------------------------------
+
+    def create_sandbox(
+        self,
+        *,
+        resource: Resource,
+        plan_id: str,
+        sandbox_manager: SandboxManager,
+    ) -> SandboxResult:
+        """Create an isolated sandbox (idempotent, delegates to SandboxManager)."""
+        location = self._require_location(resource)
+        strategy_str = self._resolve_strategy_str(resource)
+
+        existing = sandbox_manager.get_sandbox(
+            plan_id=plan_id, resource_id=resource.resource_id
+        )
+        if existing is not None and existing.context is not None:
+            return SandboxResult(
+                sandbox_id=existing.sandbox_id,
+                sandbox_path=existing.context.sandbox_path,
+                strategy=strategy_str,
+                created=False,
+                message="Reused existing sandbox",
+            )
+
+        sandbox = sandbox_manager.get_or_create_sandbox(
+            plan_id=plan_id,
+            resource_id=resource.resource_id,
+            original_path=location,
+            sandbox_strategy=cast(SandboxStrategyStr, strategy_str),
+        )
+        if sandbox.context is None:
+            raise RuntimeError(
+                f"Sandbox for resource '{resource.resource_id}' "
+                f"(plan={plan_id}) was created but has no context"
+            )
+        return SandboxResult(
+            sandbox_id=sandbox.sandbox_id,
+            sandbox_path=sandbox.context.sandbox_path,
+            strategy=strategy_str,
+            created=True,
+            message=f"Created {strategy_str} sandbox",
+        )
+
+    def create_checkpoint(
+        self,
+        *,
+        resource: Resource,
+        plan_id: str,
+        sandbox_manager: SandboxManager,
+        phase: str = "execution",
+    ) -> CheckpointResult:
+        """Create a checkpoint. Override in subclasses."""
+        raise NotImplementedError(
+            f"{self._type_label} handler does not support create_checkpoint()"
+        )
+
+    def rollback_to(
+        self,
+        *,
+        resource: Resource,
+        plan_id: str,
+        checkpoint_id: str,
+        sandbox_manager: SandboxManager,
+    ) -> RollbackResult:
+        """Rollback to a checkpoint. Override in subclasses."""
+        raise NotImplementedError(
+            f"{self._type_label} handler does not support rollback_to()"
+        )
+
+    def project_access(
+        self,
+        *,
+        resource: Resource,
+        principal: str,
+        action: str = "read",
+        project_id: str = "",
+    ) -> AccessResult:
+        """Check access permissions (default: local mode = always permit)."""
+        try:
+            from cleveragents.application.services.permission_service import (
+                get_default_permission_service,
+            )
+            from cleveragents.domain.models.core.permission import (
+                PermissionAction,
+                PermissionScope,
+            )
+
+            svc = get_default_permission_service()
+            action_enum = PermissionAction(action.lower())
+            check = svc.check_permission(
+                principal=principal,
+                action=action_enum,
+                scope=PermissionScope.PROJECT,
+                scope_id=project_id or resource.resource_id,
+            )
+            return AccessResult(
+                permitted=check.result,
+                principal=principal,
+                action=action,
+                scope="project",
+                reason=check.reason,
+            )
+        except ImportError:
+            return AccessResult(
+                permitted=True,
+                principal=principal,
+                action=action,
+                scope="project",
+                reason="Local mode — all access permitted",
+            )
+        except ValueError as exc:
+            # Invalid action string — reject rather than silently permit
+            return AccessResult(
+                permitted=False,
+                principal=principal,
+                action=action,
+                scope="project",
+                reason=f"Invalid action: {exc}",
+            )

src/cleveragents/resource/handlers/cloud.py

@@ -496,9 +496,54 @@ class CloudResourceHandler:
         raise NotImplementedError("Cloud handler does not support diff()")
 
     def discover_children(self, *, resource: Resource) -> list[Resource]:
-        """Not supported by the cloud handler."""
+        """Not supported for cloud resources."""
         raise NotImplementedError("Cloud handler does not support discover_children()")
 
+    # -- Lifecycle stubs (issue #836) --------------------------------------
+
+    def create_sandbox(
+        self,
+        *,
+        resource: Resource,
+        plan_id: str,
+        sandbox_manager: Any,
+    ) -> Any:
+        """Not supported for cloud resources."""
+        raise NotImplementedError("Cloud handler does not support create_sandbox()")
+
+    def create_checkpoint(
+        self,
+        *,
+        resource: Resource,
+        plan_id: str,
+        sandbox_manager: Any,
+        phase: str = "execution",
+    ) -> Any:
+        """Not supported for cloud resources."""
+        raise NotImplementedError("Cloud handler does not support create_checkpoint()")
+
+    def rollback_to(
+        self,
+        *,
+        resource: Resource,
+        plan_id: str,
+        checkpoint_id: str,
+        sandbox_manager: Any,
+    ) -> Any:
+        """Not supported for cloud resources."""
+        raise NotImplementedError("Cloud handler does not support rollback_to()")
+
+    def project_access(
+        self,
+        *,
+        resource: Resource,
+        principal: str,
+        action: str = "read",
+        project_id: str = "",
+    ) -> Any:
+        """Not supported for cloud resources."""
+        raise NotImplementedError("Cloud handler does not support project_access()")
+
 
 # ---------------------------------------------------------------------------
 # Stubbed sandbox strategies

src/cleveragents/resource/handlers/fs_directory.py

@@ -32,11 +32,14 @@ from cleveragents.domain.models.core.resource import (
     Resource,
     SandboxStrategy,
 )
+from cleveragents.infrastructure.sandbox.manager import SandboxManager
 from cleveragents.resource.handlers._base import BaseResourceHandler
 from cleveragents.resource.handlers.protocol import (
+    CheckpointResult,
     Content,
     DeleteResult,
     DiffResult,
+    RollbackResult,
     WriteResult,
 )
 
@@ -256,3 +259,123 @@ class FsDirectoryHandler(BaseResourceHandler):
             children.append(child)
 
         return children
+
+    # -- Checkpoint and rollback (issue #836) ------------------------------
+
+    # Instance-level checkpoint store to avoid shared mutable ClassVar state.
+    # Each handler instance tracks its own checkpoints.  Plan-level cleanup
+    # happens via discard_checkpoints().
+    def __init__(self) -> None:
+        self._checkpoints: dict[str, str] = {}
+
+    def create_checkpoint(
+        self,
+        *,
+        resource: Resource,
+        plan_id: str,
+        sandbox_manager: SandboxManager,
+        phase: str = "execution",
+    ) -> CheckpointResult:
+        """Create a checkpoint via filesystem snapshot."""
+        import tempfile
+        from datetime import UTC, datetime
+
+        sandbox = sandbox_manager.get_sandbox(
+            plan_id=plan_id, resource_id=resource.resource_id
+        )
+        if sandbox is None or sandbox.context is None:
+            raise RuntimeError(
+                f"No active sandbox for resource '{resource.resource_id}' "
+                f"(plan={plan_id})"
+            )
+
+        timestamp = datetime.now(tz=UTC).strftime("%Y%m%dT%H%M%S")
+        checkpoint_id = f"checkpoint-{plan_id}-{timestamp}"
+        snapshot_dir = tempfile.mkdtemp(prefix=f"ckpt_{checkpoint_id}_")
+        snapshot_path = Path(snapshot_dir) / "snapshot"
+
+        shutil.copytree(
+            sandbox.context.sandbox_path,
+            snapshot_path,
+            dirs_exist_ok=True,
+        )
+
+        self._checkpoints[checkpoint_id] = str(snapshot_path)
+
+        return CheckpointResult(
+            checkpoint_id=checkpoint_id,
+            plan_id=plan_id,
+            snapshot_path=str(snapshot_path),
+            message=f"Created filesystem snapshot at {snapshot_dir}",
+        )
+
+    def rollback_to(
+        self,
+        *,
+        resource: Resource,
+        plan_id: str,
+        checkpoint_id: str,
+        sandbox_manager: SandboxManager,
+    ) -> RollbackResult:
+        """Rollback by clearing sandbox and restoring from snapshot.
+
+        Skips ``.git`` directories to avoid destroying git metadata
+        in git-backed sandboxes.
+        """
+        sandbox = sandbox_manager.get_sandbox(
+            plan_id=plan_id, resource_id=resource.resource_id
+        )
+        if sandbox is None or sandbox.context is None:
+            raise RuntimeError(
+                f"No active sandbox for resource '{resource.resource_id}' "
+                f"(plan={plan_id})"
+            )
+
+        snapshot_path = self._checkpoints.get(checkpoint_id)
+        if snapshot_path is None or not Path(snapshot_path).exists():
+            return RollbackResult(
+                success=False,
+                checkpoint_id=checkpoint_id,
+                message=f"Checkpoint '{checkpoint_id}' not found or expired",
+            )
+
+        sandbox_path = Path(sandbox.context.sandbox_path)
+
+        # Clear sandbox (skip .git to preserve git metadata)
+        for item in sandbox_path.iterdir():
+            if item.name == ".git":
+                continue
+            if item.is_dir():
+                shutil.rmtree(item)
+            else:
+                item.unlink()
+
+        # Restore from snapshot (skip .git)
+        restored = 0
+        for item in Path(snapshot_path).iterdir():
+            if item.name == ".git":
+                continue
+            dest = sandbox_path / item.name
+            if item.is_dir():
+                shutil.copytree(item, dest)
+            else:
+                shutil.copy2(item, dest)
+            restored += 1
+
+        return RollbackResult(
+            success=True,
+            checkpoint_id=checkpoint_id,
+            restored_files=restored,
+            message=f"Restored {restored} items from '{checkpoint_id}'",
+        )
+
+    def discard_checkpoints(self, plan_id: str = "") -> int:
+        """Remove checkpoint snapshots (cleanup). Returns count removed."""
+        removed = 0
+        to_remove = [cid for cid in self._checkpoints if not plan_id or plan_id in cid]
+        for cid in to_remove:
+            snap = self._checkpoints.pop(cid, None)
+            if snap and Path(snap).exists():
+                shutil.rmtree(Path(snap).parent, ignore_errors=True)
+            removed += 1
+        return removed

src/cleveragents/resource/handlers/git_checkout.py

@@ -31,11 +31,14 @@ from cleveragents.domain.models.core.resource import (
     Resource,
     SandboxStrategy,
 )
+from cleveragents.infrastructure.sandbox.manager import SandboxManager
 from cleveragents.resource.handlers._base import BaseResourceHandler
 from cleveragents.resource.handlers.protocol import (
+    CheckpointResult,
     Content,
     DeleteResult,
     DiffResult,
+    RollbackResult,
     WriteResult,
 )
 
@@ -341,3 +344,101 @@ class GitCheckoutHandler(BaseResourceHandler):
             children.append(child)
 
         return children
+
+    # -- Checkpoint and rollback (issue #836) ------------------------------
+
+    def create_checkpoint(
+        self,
+        *,
+        resource: Resource,
+        plan_id: str,
+        sandbox_manager: SandboxManager,
+        phase: str = "execution",
+    ) -> CheckpointResult:
+        """Create a checkpoint via ``git tag`` in the sandbox."""
+        from datetime import UTC, datetime
+
+        sandbox = sandbox_manager.get_sandbox(
+            plan_id=plan_id, resource_id=resource.resource_id
+        )
+        if sandbox is None or sandbox.context is None:
+            raise RuntimeError(
+                f"No active sandbox for resource '{resource.resource_id}' "
+                f"(plan={plan_id})"
+            )
+
+        timestamp = datetime.now(tz=UTC).strftime("%Y%m%dT%H%M%S")
+        tag_name = f"checkpoint-{plan_id}-{timestamp}"
+
+        result = subprocess.run(
+            ["git", "tag", tag_name],
+            cwd=sandbox.context.sandbox_path,
+            capture_output=True,
+            text=True,
+            timeout=30,
+            check=False,
+        )
+        if result.returncode != 0:
+            raise RuntimeError(f"git tag failed: {result.stderr.strip()}")
+
+        return CheckpointResult(
+            checkpoint_id=tag_name,
+            plan_id=plan_id,
+            snapshot_path=sandbox.context.sandbox_path,
+            message=f"Created git tag '{tag_name}' at HEAD",
+        )
+
+    def rollback_to(
+        self,
+        *,
+        resource: Resource,
+        plan_id: str,
+        checkpoint_id: str,
+        sandbox_manager: SandboxManager,
+    ) -> RollbackResult:
+        """Rollback via ``git reset --hard`` + ``git clean -fd``.
+
+        Uses hard reset to restore both tracked and untracked file state.
+        """
+        sandbox = sandbox_manager.get_sandbox(
+            plan_id=plan_id, resource_id=resource.resource_id
+        )
+        if sandbox is None or sandbox.context is None:
+            raise RuntimeError(
+                f"No active sandbox for resource '{resource.resource_id}' "
+                f"(plan={plan_id})"
+            )
+        cwd = sandbox.context.sandbox_path
+
+        # Hard reset to the checkpoint tag
+        result = subprocess.run(
+            ["git", "reset", "--hard", checkpoint_id],
+            cwd=cwd,
+            capture_output=True,
+            text=True,
+            timeout=30,
+            check=False,
+        )
+        if result.returncode != 0:
+            return RollbackResult(
+                success=False,
+                checkpoint_id=checkpoint_id,
+                message=f"git reset --hard failed: {result.stderr.strip()}",
+            )
+
+        # Clean untracked files and directories
+        subprocess.run(
+            ["git", "clean", "-fd"],
+            cwd=cwd,
+            capture_output=True,
+            text=True,
+            timeout=30,
+            check=False,
+        )
+
+        return RollbackResult(
+            success=True,
+            checkpoint_id=checkpoint_id,
+            restored_files=0,
+            message=f"Rolled back to checkpoint '{checkpoint_id}'",
+        )

src/cleveragents/resource/handlers/protocol.py

@@ -19,15 +19,23 @@ Content CRUD and discovery (M6):
     9. ``diff``  — compare two resource states.
    10. ``discover_children`` — auto-discover child resources.
 
+Sandbox and lifecycle methods (M6, issue #836):
+   11. ``create_sandbox``  — create an isolated sandbox for the resource
+   12. ``create_checkpoint`` — snapshot the current resource state
+   13. ``rollback_to``  — restore a resource to a prior checkpoint
+   14. ``project_access`` — check access permissions within project scope
+
 Based on:
     - implementation_plan.md group M1.resource-handlers (L2254-L2271)
     - docs/specification.md Resource Handler architecture
     - Issue #827 — ResourceHandler CRUD and discovery methods
+    - Issue #836 — ResourceHandler sandbox and checkpoint methods
 """
 
 from __future__ import annotations
 
 from dataclasses import dataclass, field
+from datetime import UTC, datetime
 from typing import Protocol, runtime_checkable
 
 from cleveragents.domain.models.core.resource import Resource
@@ -118,6 +126,54 @@ class DiffResult:
     deletions: int = 0
 
 
+# ---------------------------------------------------------------------------
+# Result types for sandbox, checkpoint, and access operations (#836)
+# ---------------------------------------------------------------------------
+
+
+@dataclass(frozen=True, slots=True)
+class SandboxResult:
+    """Result of a ``create_sandbox`` operation."""
+
+    sandbox_id: str
+    sandbox_path: str
+    strategy: str
+    created: bool = True
+    message: str = ""
+
+
+@dataclass(frozen=True, slots=True)
+class CheckpointResult:
+    """Result of a ``create_checkpoint`` operation."""
+
+    checkpoint_id: str
+    plan_id: str
+    timestamp: datetime = field(default_factory=lambda: datetime.now(UTC))
+    snapshot_path: str = ""
+    message: str = ""
+
+
+@dataclass(frozen=True, slots=True)
+class RollbackResult:
+    """Result of a ``rollback_to`` operation."""
+
+    success: bool
+    checkpoint_id: str
+    restored_files: int = 0
+    message: str = ""
+
+
+@dataclass(frozen=True, slots=True)
+class AccessResult:
+    """Result of a ``project_access`` operation."""
+
+    permitted: bool
+    principal: str = ""
+    action: str = ""
+    scope: str = ""
+    reason: str = ""
+
+
 # ---------------------------------------------------------------------------
 # Protocol
 # ---------------------------------------------------------------------------
@@ -255,15 +311,50 @@ class ResourceHandler(Protocol):
         ...
 
     def discover_children(self, *, resource: Resource) -> list[Resource]:
-        """Auto-discover child resources beneath this resource.
-
-        Used to populate the resource DAG with discovered sub-resources
-        (e.g. files within a directory, tables within a database).
-
-        Args:
-            resource: The parent resource to scan.
-
-        Returns:
-            List of newly discovered child :class:`Resource` objects.
-        """
+        """Auto-discover child resources beneath this resource."""
+        ...
+
+    # -- Sandbox lifecycle methods (issue #836) ----------------------------
+
+    def create_sandbox(
+        self,
+        *,
+        resource: Resource,
+        plan_id: str,
+        sandbox_manager: SandboxManager,
+    ) -> SandboxResult:
+        """Create an isolated sandbox for the resource (idempotent)."""
+        ...
+
+    def create_checkpoint(
+        self,
+        *,
+        resource: Resource,
+        plan_id: str,
+        sandbox_manager: SandboxManager,
+        phase: str = "execution",
+    ) -> CheckpointResult:
+        """Create a checkpoint of the current resource state."""
+        ...
+
+    def rollback_to(
+        self,
+        *,
+        resource: Resource,
+        plan_id: str,
+        checkpoint_id: str,
+        sandbox_manager: SandboxManager,
+    ) -> RollbackResult:
+        """Rollback a resource to a prior checkpoint state."""
+        ...
+
+    def project_access(
+        self,
+        *,
+        resource: Resource,
+        principal: str,
+        action: str = "read",
+        project_id: str = "",
+    ) -> AccessResult:
+        """Check access permissions for the resource."""
         ...