feat(extensibility): implement Custom Sandbox Strategy Registration via SandboxStrategy Protocol

Implement SandboxStrategyProtocol, a 9-method @runtime_checkable Protocol enabling
third-party sandbox strategy registration. Includes:

- SandboxStrategyProtocol with create/read/write/diff/commit/rollback/checkpoint/
  restore_checkpoint/cleanup methods
- SandboxRef (frozen dataclass) and DiffView/DiffEntry (Pydantic models)
- SandboxStrategyRegistry with config-driven registration, Protocol validation,
  thread safety, and clear/list/has operations
- BuiltInSandboxStrategyAdapter wrapping existing Sandbox implementations to conform
  to the new Protocol
- CustomStrategyConfig for YAML/dict-based strategy registration
- SandboxFactory integration with custom_registry parameter,
  has_custom_strategy() and get_custom_strategy_class()
- 25 Behave BDD scenarios (85 steps) covering protocol, registry, adapter, config,
  and factory integration
- 8 Robot Framework integration tests with real filesystem operations
- ASV benchmarks for registry and adapter operations
- Developer documentation

ISSUES CLOSED: #586

benchmarks/bench_sandbox_strategy.py

@@ -0,0 +1,242 @@
+"""ASV benchmarks for custom sandbox strategy registration.
+
+Measures the performance of:
+- SandboxRef creation
+- DiffView / DiffEntry model creation
+- SandboxStrategyRegistry registration and lookup
+- BuiltInSandboxStrategyAdapter create/write/read cycle
+- Protocol runtime checking via isinstance
+"""
+
+from __future__ import annotations
+
+import importlib
+import os
+import sys
+import tempfile
+from datetime import datetime
+from pathlib import Path
+
+# Ensure the local *source* tree is importable even when ASV has an
+# older build of the package installed.
+_SRC = str(Path(__file__).resolve().parents[1] / "src")
+if _SRC not in sys.path:
+    sys.path.insert(0, _SRC)
+
+# Force-reload so ASV picks up the source tree version.
+import cleveragents  # noqa: E402
+
+importlib.reload(cleveragents)
+
+from cleveragents.domain.models.core.resource import (  # noqa: E402
+    PhysVirt,
+    Resource,
+    ResourceCapabilities,
+    SandboxStrategy as SandboxStrategyEnum,
+)
+from cleveragents.domain.models.core.sandbox_strategy import (  # noqa: E402
+    DiffEntry,
+    DiffView,
+    SandboxRef,
+    SandboxStrategyProtocol,
+)
+from cleveragents.infrastructure.sandbox.strategy_adapter import (  # noqa: E402
+    BuiltInSandboxStrategyAdapter,
+)
+from cleveragents.infrastructure.sandbox.strategy_registry import (  # noqa: E402
+    SandboxStrategyRegistry,
+)
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+_COUNTER = 0
+
+
+def _ulid(name: str) -> str:
+    global _COUNTER
+    _COUNTER += 1
+    base = abs(hash(name)) % (10**20)
+    raw = f"{_COUNTER:06d}{base:020d}"
+    charset = "0123456789ABCDEFGHJKMNPQRSTVWXYZ"
+    result = ""
+    for ch in raw:
+        result += charset[int(ch)]
+    return (result + "0" * 26)[:26]
+
+
+class _MockStrategy:
+    """Minimal strategy satisfying the Protocol for benchmarking."""
+
+    def create(self, plan_id, resource):  # noqa: ANN001, ANN201
+        return None
+
+    def read(self, ref, path):  # noqa: ANN001, ANN201
+        return b""
+
+    def write(self, ref, path, content):  # noqa: ANN001, ANN201
+        return None
+
+    def diff(self, ref):  # noqa: ANN001, ANN201
+        return None
+
+    def commit(self, ref):  # noqa: ANN001, ANN201
+        pass
+
+    def rollback(self, ref):  # noqa: ANN001, ANN201
+        pass
+
+    def checkpoint(self, ref, checkpoint_id):  # noqa: ANN001, ANN201
+        pass
+
+    def restore_checkpoint(self, ref, checkpoint_id):  # noqa: ANN001, ANN201
+        pass
+
+    def cleanup(self, ref):  # noqa: ANN001, ANN201
+        pass
+
+
+# ---------------------------------------------------------------------------
+# SandboxRef benchmarks
+# ---------------------------------------------------------------------------
+
+
+class TimeSandboxRef:
+    """Benchmark SandboxRef creation."""
+
+    def time_create_ref(self) -> None:
+        SandboxRef(
+            sandbox_id="bench-ref",
+            plan_id="bench-plan",
+            resource_id="bench-res",
+            created_at=datetime.now(),
+        )
+
+    def time_create_ref_with_metadata(self) -> None:
+        SandboxRef(
+            sandbox_id="bench-ref-meta",
+            plan_id="bench-plan",
+            resource_id="bench-res",
+            created_at=datetime.now(),
+            metadata={"key": "value", "num": 42},
+        )
+
+
+# ---------------------------------------------------------------------------
+# DiffEntry / DiffView benchmarks
+# ---------------------------------------------------------------------------
+
+
+class TimeDiffModels:
+    """Benchmark DiffEntry and DiffView creation."""
+
+    def time_create_diff_entry(self) -> None:
+        DiffEntry(path="bench.txt", operation="modified")
+
+    def time_create_diff_entry_with_content(self) -> None:
+        DiffEntry(
+            path="bench.txt",
+            operation="modified",
+            before=b"old",
+            after=b"new",
+        )
+
+    def time_create_diff_view_10_entries(self) -> None:
+        entries = [
+            DiffEntry(path=f"file{i}.txt", operation="modified") for i in range(10)
+        ]
+        DiffView(sandbox_id="bench-dv", entries=entries)
+
+
+# ---------------------------------------------------------------------------
+# Protocol checking benchmarks
+# ---------------------------------------------------------------------------
+
+
+class TimeProtocolCheck:
+    """Benchmark SandboxStrategyProtocol isinstance checks."""
+
+    def setup(self) -> None:
+        self.good = _MockStrategy()
+
+    def time_isinstance_check_pass(self) -> None:
+        isinstance(self.good, SandboxStrategyProtocol)
+
+    def time_isinstance_check_fail(self) -> None:
+        isinstance("not a strategy", SandboxStrategyProtocol)
+
+
+# ---------------------------------------------------------------------------
+# SandboxStrategyRegistry benchmarks
+# ---------------------------------------------------------------------------
+
+
+class TimeStrategyRegistry:
+    """Benchmark registry operations."""
+
+    def setup(self) -> None:
+        self.registry = SandboxStrategyRegistry(
+            allowed_prefixes=("cleveragents.",),
+        )
+        self.registry.register(
+            "bench_adapter",
+            "cleveragents.infrastructure.sandbox.strategy_adapter",
+            "BuiltInSandboxStrategyAdapter",
+        )
+
+    def time_lookup_hit(self) -> None:
+        self.registry.get("bench_adapter")
+
+    def time_lookup_miss(self) -> None:
+        self.registry.get("nonexistent")
+
+    def time_has_check(self) -> None:
+        self.registry.has("bench_adapter")
+
+    def time_list_strategies(self) -> None:
+        self.registry.list_strategies()
+
+
+# ---------------------------------------------------------------------------
+# BuiltInSandboxStrategyAdapter benchmarks
+# ---------------------------------------------------------------------------
+
+
+class TimeAdapterLifecycle:
+    """Benchmark adapter create/write/read cycle."""
+
+    def setup(self) -> None:
+        self.tmpdir = tempfile.mkdtemp(prefix="ca-bench-adapter-")
+        with open(os.path.join(self.tmpdir, "seed.txt"), "w") as f:
+            f.write("seed")
+        self.adapter = BuiltInSandboxStrategyAdapter(strategy_name="copy_on_write")
+        self.resource = Resource(
+            resource_id=_ulid("bench-adapter"),
+            name="bench-adapter",
+            resource_type_name="fs-directory",
+            classification=PhysVirt.PHYSICAL,
+            sandbox_strategy=SandboxStrategyEnum.COPY_ON_WRITE,
+            location=self.tmpdir,
+            capabilities=ResourceCapabilities(
+                readable=True,
+                writable=True,
+                sandboxable=True,
+                checkpointable=False,
+            ),
+        )
+
+    def time_create_and_cleanup(self) -> None:
+        ref = self.adapter.create("bench-plan", self.resource)
+        self.adapter.cleanup(ref)
+
+    def time_write_read_cycle(self) -> None:
+        ref = self.adapter.create("bench-plan-wr", self.resource)
+        self.adapter.write(ref, "bench.txt", b"bench content")
+        self.adapter.read(ref, "bench.txt")
+        self.adapter.cleanup(ref)
+
+    def teardown(self) -> None:
+        import shutil
+
+        shutil.rmtree(self.tmpdir, ignore_errors=True)

docs/development/custom_sandbox_strategy.md

@@ -0,0 +1,150 @@
+# Custom Sandbox Strategy Registration
+
+## Overview
+
+CleverAgents supports custom sandbox strategies via the `SandboxStrategyProtocol`.
+Custom strategies enable domain-specific isolation for specialized resource types
+that are not covered by the built-in strategies (none, git\_worktree, copy\_on\_write,
+transaction\_rollback).
+
+## SandboxStrategy Protocol
+
+All custom sandbox strategies must implement the 9-method `SandboxStrategyProtocol`:
+
+```python
+from cleveragents.domain.models.core.sandbox_strategy import (
+    SandboxStrategyProtocol,
+    SandboxRef,
+    DiffEntry,
+    DiffView,
+)
+from cleveragents.domain.models.core.resource import Resource
+
+
+class MyCustomSandbox:
+    def create(self, plan_id: str, resource: Resource) -> SandboxRef: ...
+    def read(self, ref: SandboxRef, path: str) -> bytes: ...
+    def write(self, ref: SandboxRef, path: str, content: bytes) -> DiffEntry: ...
+    def diff(self, ref: SandboxRef) -> DiffView: ...
+    def commit(self, ref: SandboxRef) -> None: ...
+    def rollback(self, ref: SandboxRef) -> None: ...
+    def checkpoint(self, ref: SandboxRef, checkpoint_id: str) -> None: ...
+    def restore_checkpoint(self, ref: SandboxRef, checkpoint_id: str) -> None: ...
+    def cleanup(self, ref: SandboxRef) -> None: ...
+```
+
+## Supporting Models
+
+### SandboxRef
+
+Opaque reference returned by `create()` and passed to all other methods:
+
+```python
+@dataclass(frozen=True)
+class SandboxRef:
+    sandbox_id: str
+    plan_id: str
+    resource_id: str
+    created_at: datetime
+    metadata: dict[str, object] = field(default_factory=dict)
+```
+
+### DiffEntry / DiffView
+
+`DiffEntry` describes a single file change; `DiffView` aggregates entries:
+
+```python
+class DiffEntry(BaseModel):
+    path: str          # relative file path
+    operation: str     # "added", "modified", or "deleted"
+    before: bytes | None
+    after: bytes | None
+
+class DiffView(BaseModel):
+    sandbox_id: str
+    entries: list[DiffEntry]
+    summary: str
+```
+
+## Config-Driven Registration
+
+Custom strategies are registered via the `SandboxStrategyRegistry`:
+
+```python
+from cleveragents.infrastructure.sandbox.strategy_registry import (
+    SandboxStrategyRegistry,
+)
+
+registry = SandboxStrategyRegistry(
+    allowed_prefixes=("cleveragents.", "mycompany."),
+)
+
+# Single registration
+registry.register(
+    "my_strategy",
+    "mycompany.sandbox.custom",
+    "MyCustomSandbox",
+)
+
+# Batch config-driven registration
+configs = {
+    "redis_sandbox": {
+        "module": "mycompany.sandbox.redis",
+        "class": "RedisSandbox",
+    },
+    "s3_sandbox": {
+        "module": "mycompany.sandbox.s3",
+        "class": "S3Sandbox",
+    },
+}
+registry.register_all_from_config(configs)
+```
+
+## Factory Integration
+
+The `SandboxFactory` can be initialized with a custom registry:
+
+```python
+from cleveragents.infrastructure.sandbox.factory import SandboxFactory
+
+factory = SandboxFactory(custom_registry=registry)
+
+# Check if custom strategy is available
+if factory.has_custom_strategy("redis_sandbox"):
+    cls = factory.get_custom_strategy_class("redis_sandbox")
+```
+
+## Built-In Strategy Adapter
+
+The `BuiltInSandboxStrategyAdapter` wraps existing `Sandbox` implementations
+to conform to the `SandboxStrategyProtocol`:
+
+```python
+from cleveragents.infrastructure.sandbox.strategy_adapter import (
+    BuiltInSandboxStrategyAdapter,
+)
+
+adapter = BuiltInSandboxStrategyAdapter(strategy_name="copy_on_write")
+ref = adapter.create("plan-001", resource)
+adapter.write(ref, "file.txt", b"content")
+data = adapter.read(ref, "file.txt")
+adapter.cleanup(ref)
+```
+
+## Protocol Validation
+
+All strategies are validated against the Protocol at registration time.
+The registry checks that all 9 required methods are present as callable
+attributes:
+
+```python
+# This will raise ProtocolMismatchError if the class
+# doesn't implement all required methods:
+registry.register("bad", "mymodule", "IncompleteClass")
+```
+
+## Security
+
+Module imports are restricted to a configurable prefix allowlist
+(default: `("cleveragents.",)`). Only modules whose fully-qualified
+name starts with an allowed prefix may be dynamically imported.

features/extensibility/custom_sandbox_strategy.feature

@@ -0,0 +1,193 @@
+@extensibility @custom_sandbox_strategy
+Feature: Custom Sandbox Strategy Registration via SandboxStrategy Protocol
+  As a CleverAgents developer
+  I want to register custom sandbox strategies via configuration
+  So that specialized resource types can use domain-specific isolation
+
+  # ---------------------------------------------------------------------------
+  # SandboxRef model
+  # ---------------------------------------------------------------------------
+
+  @sandbox_ref
+  Scenario: SandboxRef is a frozen dataclass
+    Given a SandboxRef with sandbox_id "sb-001" and plan_id "plan-001"
+    Then the SandboxRef sandbox_id should be "sb-001"
+    And the SandboxRef plan_id should be "plan-001"
+    And the SandboxRef should be immutable
+
+  @sandbox_ref
+  Scenario: SandboxRef carries metadata
+    Given a SandboxRef with metadata key "backend" value "redis"
+    Then the SandboxRef metadata should contain key "backend"
+
+  # ---------------------------------------------------------------------------
+  # DiffView / DiffEntry models
+  # ---------------------------------------------------------------------------
+
+  @diff_view
+  Scenario: DiffEntry creation with required fields
+    Given a DiffEntry with path "src/main.py" and operation "modified"
+    Then the DiffEntry path should be "src/main.py"
+    And the DiffEntry operation should be "modified"
+
+  @diff_view
+  Scenario: DiffView aggregates entries
+    Given a DiffView with 3 entries
+    Then the DiffView should have 3 entries
+    And the DiffView should have a sandbox_id
+
+  @diff_view
+  Scenario: DiffEntry rejects empty path
+    When I attempt to create a DiffEntry with empty path
+    Then a sandbox strategy validation error should be raised
+
+  # ---------------------------------------------------------------------------
+  # SandboxStrategyProtocol
+  # ---------------------------------------------------------------------------
+
+  @sandbox_strategy_protocol
+  Scenario: SandboxStrategyProtocol is runtime_checkable
+    Given the SandboxStrategyProtocol is available
+    Then it should be a runtime_checkable Protocol
+
+  @sandbox_strategy_protocol
+  Scenario: A class implementing all 9 methods satisfies the Protocol
+    Given a mock class implementing all 9 SandboxStrategy methods
+    Then the mock class should satisfy SandboxStrategyProtocol
+
+  @sandbox_strategy_protocol
+  Scenario: A class missing methods does not satisfy the Protocol
+    Given a class missing the checkpoint method
+    Then the class should not satisfy SandboxStrategyProtocol
+
+  # ---------------------------------------------------------------------------
+  # SandboxStrategyRegistry
+  # ---------------------------------------------------------------------------
+
+  @strategy_registry
+  Scenario: Register a valid custom strategy
+    Given a SandboxStrategyRegistry with test allowed prefixes
+    And a valid custom strategy class in "cleveragents.infrastructure.sandbox.strategy_adapter"
+    When I register the strategy as "test_adapter"
+    Then the registry should contain strategy "test_adapter"
+    And listing strategies should include "test_adapter"
+
+  @strategy_registry
+  Scenario: Registering a non-protocol class raises ProtocolMismatchError
+    Given a SandboxStrategyRegistry with test allowed prefixes
+    When I attempt to register a non-protocol class as "bad_strategy"
+    Then a sandbox ProtocolMismatchError should be raised
+
+  @strategy_registry
+  Scenario: Register with empty name raises ValueError
+    Given a SandboxStrategyRegistry with test allowed prefixes
+    When I attempt to register a strategy with empty name
+    Then a sandbox ValueError should be raised
+
+  @strategy_registry
+  Scenario: Get unregistered strategy returns None
+    Given an empty SandboxStrategyRegistry
+    When I look up strategy "nonexistent"
+    Then the strategy lookup result should be None
+
+  @strategy_registry
+  Scenario: Register all from config dictionary
+    Given a SandboxStrategyRegistry with test allowed prefixes
+    And a config dictionary with 2 custom strategies
+    When I register all from config
+    Then 2 strategies should be registered
+
+  @strategy_registry
+  Scenario: Clear removes all strategies
+    Given a SandboxStrategyRegistry with one registered strategy
+    When I clear the sandbox strategy registry
+    Then the sandbox strategy registry should be empty
+
+  # ---------------------------------------------------------------------------
+  # BuiltInSandboxStrategyAdapter
+  # ---------------------------------------------------------------------------
+
+  @adapter
+  Scenario: Adapter create returns SandboxRef
+    Given a BuiltInSandboxStrategyAdapter for "none" strategy
+    And a test Resource with location
+    When I call adapter.create with plan "plan-001"
+    Then I should get a SandboxRef with plan_id "plan-001"
+
+  @adapter
+  Scenario: Adapter write creates a DiffEntry
+    Given a BuiltInSandboxStrategyAdapter for "copy_on_write" strategy
+    And a test Resource with a temporary directory
+    And the adapter sandbox is created for plan "plan-write"
+    When I call adapter.write with path "test.txt" and content "hello"
+    Then I should get a DiffEntry with operation "added"
+
+  @adapter
+  Scenario: Adapter read returns file content
+    Given a BuiltInSandboxStrategyAdapter for "copy_on_write" strategy
+    And a test Resource with a temporary directory
+    And the adapter sandbox is created for plan "plan-read"
+    And I write "data" to "read_test.txt" via the adapter
+    When I call adapter.read for "read_test.txt"
+    Then I should get content "data"
+
+  @adapter
+  Scenario: Adapter diff returns DiffView
+    Given a BuiltInSandboxStrategyAdapter for "none" strategy
+    And a test Resource with location
+    And the adapter sandbox is created for plan "plan-diff"
+    When I call adapter.diff
+    Then I should get a DiffView
+
+  @adapter
+  Scenario: Adapter checkpoint and restore
+    Given a BuiltInSandboxStrategyAdapter for "copy_on_write" strategy
+    And a test Resource with a temporary directory
+    And the adapter sandbox is created for plan "plan-cp"
+    And I write "v1" to "cp_test.txt" via the adapter
+    And I checkpoint as "cp-1"
+    And I write "v2" to "cp_test.txt" via the adapter
+    When I restore checkpoint "cp-1"
+    Then reading "cp_test.txt" should return "v1"
+
+  @adapter
+  Scenario: Adapter cleanup releases resources
+    Given a BuiltInSandboxStrategyAdapter for "none" strategy
+    And a test Resource with location
+    And the adapter sandbox is created for plan "plan-cleanup"
+    When I call adapter.cleanup
+    Then the adapter should have no tracked sandboxes
+
+  # ---------------------------------------------------------------------------
+  # CustomStrategyConfig
+  # ---------------------------------------------------------------------------
+
+  @config
+  Scenario: CustomStrategyConfig rejects empty name
+    When I attempt to create a CustomStrategyConfig with empty name
+    Then a sandbox ValueError should be raised
+
+  @config
+  Scenario: CustomStrategyConfig rejects empty module
+    When I attempt to create a CustomStrategyConfig with empty module
+    Then a sandbox ValueError should be raised
+
+  @config
+  Scenario: CustomStrategyConfig rejects empty class_name
+    When I attempt to create a CustomStrategyConfig with empty class_name
+    Then a sandbox ValueError should be raised
+
+  # ---------------------------------------------------------------------------
+  # Factory custom strategy integration
+  # ---------------------------------------------------------------------------
+
+  @factory_integration
+  Scenario: Factory reports custom strategy availability
+    Given a SandboxFactory with a custom registry containing "my_custom"
+    Then the factory should report "my_custom" as a custom strategy
+    And the factory should not report "nonexistent" as a custom strategy
+
+  @factory_integration
+  Scenario: Factory without registry reports no custom strategies
+    Given a SandboxFactory without a custom registry
+    Then the factory should not report "anything" as a custom strategy

features/steps/custom_sandbox_strategy_steps.py

@@ -0,0 +1,602 @@
+"""Behave step definitions for Custom Sandbox Strategy Registration.
+
+Covers SandboxRef, DiffView, DiffEntry, SandboxStrategyProtocol,
+SandboxStrategyRegistry, BuiltInSandboxStrategyAdapter,
+CustomStrategyConfig, and factory integration.
+
+Based on issue #586.
+"""
+
+from __future__ import annotations
+
+import os
+import tempfile
+from datetime import datetime
+from typing import Any
+
+from behave import given, then, when  # type: ignore[import-untyped]
+from behave.runner import Context  # type: ignore[import-untyped]
+from pydantic import ValidationError
+
+from cleveragents.domain.models.core.resource import (
+    PhysVirt,
+    Resource,
+    ResourceCapabilities,
+)
+from cleveragents.domain.models.core.resource import (
+    SandboxStrategy as SandboxStrategyEnum,
+)
+from cleveragents.domain.models.core.sandbox_strategy import (
+    DiffEntry,
+    DiffView,
+    SandboxRef,
+    SandboxStrategyProtocol,
+)
+from cleveragents.infrastructure.plugins.exceptions import (
+    ProtocolMismatchError,
+)
+from cleveragents.infrastructure.sandbox.factory import SandboxFactory
+from cleveragents.infrastructure.sandbox.strategy_adapter import (
+    BuiltInSandboxStrategyAdapter,
+)
+from cleveragents.infrastructure.sandbox.strategy_registry import (
+    CustomStrategyConfig,
+    SandboxStrategyRegistry,
+)
+
+# ---------------------------------------------------------------------------
+# ULID helper
+# ---------------------------------------------------------------------------
+
+_COUNTER = 0
+
+
+def _ulid(name: str) -> str:
+    global _COUNTER
+    _COUNTER += 1
+    base = abs(hash(name)) % (10**20)
+    raw = f"{_COUNTER:06d}{base:020d}"
+    charset = "0123456789ABCDEFGHJKMNPQRSTVWXYZ"
+    result = ""
+    for ch in raw:
+        result += charset[int(ch)]
+    return (result + "0" * 26)[:26]
+
+
+def _make_resource(name: str, location: str | None = None) -> Resource:
+    return Resource(
+        resource_id=_ulid(name),
+        name=name,
+        resource_type_name="fs-directory",
+        classification=PhysVirt.PHYSICAL,
+        sandbox_strategy=SandboxStrategyEnum.NONE,
+        location=location,
+        capabilities=ResourceCapabilities(
+            readable=True, writable=True, sandboxable=True, checkpointable=False
+        ),
+    )
+
+
+# ---------------------------------------------------------------------------
+# SandboxRef
+# ---------------------------------------------------------------------------
+
+
+@given('a SandboxRef with sandbox_id "{sid}" and plan_id "{pid}"')
+def step_create_sandbox_ref(context: Context, sid: str, pid: str) -> None:
+    context.sandbox_ref = SandboxRef(
+        sandbox_id=sid,
+        plan_id=pid,
+        resource_id="res-001",
+        created_at=datetime.now(),
+    )
+
+
+@then('the SandboxRef sandbox_id should be "{expected}"')
+def step_sandbox_ref_sid(context: Context, expected: str) -> None:
+    assert context.sandbox_ref.sandbox_id == expected
+
+
+@then('the SandboxRef plan_id should be "{expected}"')
+def step_sandbox_ref_pid(context: Context, expected: str) -> None:
+    assert context.sandbox_ref.plan_id == expected
+
+
+@then("the SandboxRef should be immutable")
+def step_sandbox_ref_frozen(context: Context) -> None:
+    import dataclasses
+
+    assert dataclasses.is_dataclass(context.sandbox_ref)
+    try:
+        context.sandbox_ref.sandbox_id = "mutated"  # type: ignore[misc]
+        raise AssertionError("Expected FrozenInstanceError")
+    except (AttributeError, dataclasses.FrozenInstanceError):
+        pass
+
+
+@given('a SandboxRef with metadata key "{key}" value "{value}"')
+def step_sandbox_ref_metadata(context: Context, key: str, value: str) -> None:
+    context.sandbox_ref = SandboxRef(
+        sandbox_id="sb-meta",
+        plan_id="plan-meta",
+        resource_id="res-meta",
+        created_at=datetime.now(),
+        metadata={key: value},
+    )
+
+
+@then('the SandboxRef metadata should contain key "{key}"')
+def step_sandbox_ref_meta_key(context: Context, key: str) -> None:
+    assert key in context.sandbox_ref.metadata
+
+
+# ---------------------------------------------------------------------------
+# DiffEntry / DiffView
+# ---------------------------------------------------------------------------
+
+
+@given('a DiffEntry with path "{path}" and operation "{op}"')
+def step_create_diff_entry(context: Context, path: str, op: str) -> None:
+    context.diff_entry = DiffEntry(path=path, operation=op)
+
+
+@then('the DiffEntry path should be "{expected}"')
+def step_diff_entry_path(context: Context, expected: str) -> None:
+    assert context.diff_entry.path == expected
+
+
+@then('the DiffEntry operation should be "{expected}"')
+def step_diff_entry_op(context: Context, expected: str) -> None:
+    assert context.diff_entry.operation == expected
+
+
+@given("a DiffView with {count:d} entries")
+def step_create_diff_view(context: Context, count: int) -> None:
+    entries = [
+        DiffEntry(path=f"file{i}.txt", operation="modified") for i in range(count)
+    ]
+    context.diff_view = DiffView(
+        sandbox_id="dv-001",
+        entries=entries,
+    )
+
+
+@then("the DiffView should have {count:d} entries")
+def step_diff_view_count(context: Context, count: int) -> None:
+    assert len(context.diff_view.entries) == count
+
+
+@then("the DiffView should have a sandbox_id")
+def step_diff_view_sid(context: Context) -> None:
+    assert context.diff_view.sandbox_id
+
+
+@when("I attempt to create a DiffEntry with empty path")
+def step_create_diff_entry_empty(context: Context) -> None:
+    context.sandbox_error = None
+    try:
+        DiffEntry(path="", operation="modified")
+    except ValidationError as exc:
+        context.sandbox_error = exc
+
+
+@then("a sandbox strategy validation error should be raised")
+def step_sandbox_validation_error(context: Context) -> None:
+    assert context.sandbox_error is not None
+
+
+# ---------------------------------------------------------------------------
+# SandboxStrategyProtocol
+# ---------------------------------------------------------------------------
+
+
+@given("the SandboxStrategyProtocol is available")
+def step_protocol_available(context: Context) -> None:
+    context.protocol = SandboxStrategyProtocol
+
+
+@then("it should be a runtime_checkable Protocol")
+def step_protocol_runtime_checkable(context: Context) -> None:
+    assert hasattr(context.protocol, "__protocol_attrs__") or hasattr(
+        context.protocol, "_is_runtime_protocol"
+    )
+
+
+class _FullMockStrategy:
+    """Mock satisfying all 9 SandboxStrategy methods."""
+
+    def create(self, plan_id: str, resource: Any) -> Any:
+        return None
+
+    def read(self, ref: Any, path: str) -> bytes:
+        return b""
+
+    def write(self, ref: Any, path: str, content: bytes) -> Any:
+        return None
+
+    def diff(self, ref: Any) -> Any:
+        return None
+
+    def commit(self, ref: Any) -> None:
+        pass
+
+    def rollback(self, ref: Any) -> None:
+        pass
+
+    def checkpoint(self, ref: Any, checkpoint_id: str) -> None:
+        pass
+
+    def restore_checkpoint(self, ref: Any, checkpoint_id: str) -> None:
+        pass
+
+    def cleanup(self, ref: Any) -> None:
+        pass
+
+
+class _PartialMockStrategy:
+    """Missing checkpoint method."""
+
+    def create(self, plan_id: str, resource: Any) -> Any:
+        return None
+
+    def read(self, ref: Any, path: str) -> bytes:
+        return b""
+
+    def write(self, ref: Any, path: str, content: bytes) -> Any:
+        return None
+
+    def diff(self, ref: Any) -> Any:
+        return None
+
+    def commit(self, ref: Any) -> None:
+        pass
+
+    def rollback(self, ref: Any) -> None:
+        pass
+
+    # Missing: checkpoint, restore_checkpoint
+
+    def cleanup(self, ref: Any) -> None:
+        pass
+
+
+@given("a mock class implementing all 9 SandboxStrategy methods")
+def step_full_mock(context: Context) -> None:
+    context.mock_cls = _FullMockStrategy
+
+
+@then("the mock class should satisfy SandboxStrategyProtocol")
+def step_mock_satisfies(context: Context) -> None:
+    instance = context.mock_cls()
+    assert isinstance(instance, SandboxStrategyProtocol)
+
+
+@given("a class missing the checkpoint method")
+def step_partial_mock(context: Context) -> None:
+    context.partial_cls = _PartialMockStrategy
+
+
+@then("the class should not satisfy SandboxStrategyProtocol")
+def step_partial_fails(context: Context) -> None:
+    instance = context.partial_cls()
+    assert not isinstance(instance, SandboxStrategyProtocol)
+
+
+# ---------------------------------------------------------------------------
+# SandboxStrategyRegistry
+# ---------------------------------------------------------------------------
+
+
+@given("a SandboxStrategyRegistry with test allowed prefixes")
+def step_registry_with_prefixes(context: Context) -> None:
+    context.strategy_registry = SandboxStrategyRegistry(
+        allowed_prefixes=("cleveragents.",)
+    )
+
+
+@given("an empty SandboxStrategyRegistry")
+def step_empty_registry(context: Context) -> None:
+    context.strategy_registry = SandboxStrategyRegistry(
+        allowed_prefixes=("cleveragents.",)
+    )
+
+
+@given(
+    "a valid custom strategy class in "
+    '"cleveragents.infrastructure.sandbox.strategy_adapter"'
+)
+def step_valid_strategy_class(context: Context) -> None:
+    context.strategy_module = "cleveragents.infrastructure.sandbox.strategy_adapter"
+    context.strategy_class = "BuiltInSandboxStrategyAdapter"
+
+
+@when('I register the strategy as "{name}"')
+def step_register_strategy(context: Context, name: str) -> None:
+    context.strategy_registry.register(
+        name, context.strategy_module, context.strategy_class
+    )
+
+
+@then('the registry should contain strategy "{name}"')
+def step_registry_contains(context: Context, name: str) -> None:
+    assert context.strategy_registry.has(name)
+
+
+@then('listing strategies should include "{name}"')
+def step_registry_list(context: Context, name: str) -> None:
+    assert name in context.strategy_registry.list_strategies()
+
+
+@when('I attempt to register a non-protocol class as "{name}"')
+def step_register_nonprotocol(context: Context, name: str) -> None:
+    context.sandbox_error = None
+    try:
+        context.strategy_registry.register(
+            name,
+            "cleveragents.infrastructure.sandbox.protocol",
+            "SandboxError",
+        )
+    except ProtocolMismatchError as exc:
+        context.sandbox_error = exc
+
+
+@then("a sandbox ProtocolMismatchError should be raised")
+def step_sandbox_protocol_mismatch(context: Context) -> None:
+    assert isinstance(context.sandbox_error, ProtocolMismatchError)
+
+
+@when("I attempt to register a strategy with empty name")
+def step_register_empty_name(context: Context) -> None:
+    context.sandbox_error = None
+    try:
+        context.strategy_registry.register(
+            "",
+            "cleveragents.infrastructure.sandbox.strategy_adapter",
+            "BuiltInSandboxStrategyAdapter",
+        )
+    except ValueError as exc:
+        context.sandbox_error = exc
+
+
+@then("a sandbox ValueError should be raised")
+def step_sandbox_value_error(context: Context) -> None:
+    assert isinstance(context.sandbox_error, (ValueError, type(None))) is False or (
+        context.sandbox_error is not None
+    )
+    assert context.sandbox_error is not None
+
+
+@when('I look up strategy "{name}"')
+def step_lookup_strategy(context: Context, name: str) -> None:
+    context.strategy_lookup_result = context.strategy_registry.get(name)
+
+
+@then("the strategy lookup result should be None")
+def step_strategy_result_none(context: Context) -> None:
+    assert context.strategy_lookup_result is None
+
+
+@given("a config dictionary with 2 custom strategies")
+def step_config_dict(context: Context) -> None:
+    context.config_dict = {
+        "adapter1": {
+            "module": "cleveragents.infrastructure.sandbox.strategy_adapter",
+            "class": "BuiltInSandboxStrategyAdapter",
+        },
+        "adapter2": {
+            "module": "cleveragents.infrastructure.sandbox.strategy_adapter",
+            "class": "BuiltInSandboxStrategyAdapter",
+        },
+    }
+
+
+@when("I register all from config")
+def step_register_all_config(context: Context) -> None:
+    context.registered = context.strategy_registry.register_all_from_config(
+        context.config_dict
+    )
+
+
+@then("{count:d} strategies should be registered")
+def step_count_registered(context: Context, count: int) -> None:
+    assert len(context.registered) == count
+
+
+@given("a SandboxStrategyRegistry with one registered strategy")
+def step_registry_one(context: Context) -> None:
+    context.strategy_registry = SandboxStrategyRegistry(
+        allowed_prefixes=("cleveragents.",)
+    )
+    context.strategy_registry.register(
+        "single",
+        "cleveragents.infrastructure.sandbox.strategy_adapter",
+        "BuiltInSandboxStrategyAdapter",
+    )
+
+
+@when("I clear the sandbox strategy registry")
+def step_clear_sandbox_registry(context: Context) -> None:
+    context.strategy_registry.clear()
+
+
+@then("the sandbox strategy registry should be empty")
+def step_sandbox_registry_empty(context: Context) -> None:
+    assert len(context.strategy_registry.list_strategies()) == 0
+
+
+# ---------------------------------------------------------------------------
+# BuiltInSandboxStrategyAdapter
+# ---------------------------------------------------------------------------
+
+
+@given('a BuiltInSandboxStrategyAdapter for "{strategy}" strategy')
+def step_adapter(context: Context, strategy: str) -> None:
+    context.adapter = BuiltInSandboxStrategyAdapter(strategy_name=strategy)  # type: ignore[arg-type]
+
+
+@given("a test Resource with location")
+def step_resource_with_location(context: Context) -> None:
+    context.tmpdir = tempfile.mkdtemp(prefix="ca-test-")
+    context.test_resource = _make_resource("test-res", location=context.tmpdir)
+
+
+@given("a test Resource with a temporary directory")
+def step_resource_with_tmpdir(context: Context) -> None:
+    context.tmpdir = tempfile.mkdtemp(prefix="ca-test-cow-")
+    # Create a seed file so CoW sandbox has something to copy
+    with open(os.path.join(context.tmpdir, "seed.txt"), "w") as f:
+        f.write("seed")
+    context.test_resource = Resource(
+        resource_id=_ulid("cow-res"),
+        name="cow-res",
+        resource_type_name="fs-directory",
+        classification=PhysVirt.PHYSICAL,
+        sandbox_strategy=SandboxStrategyEnum.COPY_ON_WRITE,
+        location=context.tmpdir,
+        capabilities=ResourceCapabilities(
+            readable=True, writable=True, sandboxable=True, checkpointable=False
+        ),
+    )
+
+
+@when('I call adapter.create with plan "{plan_id}"')
+def step_adapter_create(context: Context, plan_id: str) -> None:
+    context.adapter_ref = context.adapter.create(plan_id, context.test_resource)
+
+
+@given('the adapter sandbox is created for plan "{plan_id}"')
+def step_adapter_create_given(context: Context, plan_id: str) -> None:
+    context.adapter_ref = context.adapter.create(plan_id, context.test_resource)
+
+
+@then('I should get a SandboxRef with plan_id "{expected}"')
+def step_adapter_ref_check(context: Context, expected: str) -> None:
+    assert isinstance(context.adapter_ref, SandboxRef)
+    assert context.adapter_ref.plan_id == expected
+
+
+@when('I call adapter.write with path "{path}" and content "{content}"')
+def step_adapter_write(context: Context, path: str, content: str) -> None:
+    context.write_result = context.adapter.write(
+        context.adapter_ref, path, content.encode()
+    )
+
+
+@given('I write "{content}" to "{path}" via the adapter')
+def step_adapter_write_given(context: Context, content: str, path: str) -> None:
+    context.adapter.write(context.adapter_ref, path, content.encode())
+
+
+@then('I should get a DiffEntry with operation "{expected}"')
+def step_adapter_write_op(context: Context, expected: str) -> None:
+    assert isinstance(context.write_result, DiffEntry)
+    assert context.write_result.operation == expected
+
+
+@when('I call adapter.read for "{path}"')
+def step_adapter_read(context: Context, path: str) -> None:
+    context.read_result = context.adapter.read(context.adapter_ref, path)
+
+
+@then('I should get content "{expected}"')
+def step_adapter_read_content(context: Context, expected: str) -> None:
+    assert context.read_result == expected.encode()
+
+
+@when("I call adapter.diff")
+def step_adapter_diff(context: Context) -> None:
+    context.diff_result = context.adapter.diff(context.adapter_ref)
+
+
+@then("I should get a DiffView")
+def step_adapter_diff_check(context: Context) -> None:
+    assert isinstance(context.diff_result, DiffView)
+
+
+@given('I checkpoint as "{cp_id}"')
+def step_adapter_checkpoint(context: Context, cp_id: str) -> None:
+    context.adapter.checkpoint(context.adapter_ref, cp_id)
+
+
+@when('I restore checkpoint "{cp_id}"')
+def step_adapter_restore(context: Context, cp_id: str) -> None:
+    context.adapter.restore_checkpoint(context.adapter_ref, cp_id)
+
+
+@then('reading "{path}" should return "{expected}"')
+def step_adapter_read_after_restore(context: Context, path: str, expected: str) -> None:
+    data = context.adapter.read(context.adapter_ref, path)
+    assert data == expected.encode(), f"Expected {expected!r}, got {data!r}"
+
+
+@when("I call adapter.cleanup")
+def step_adapter_cleanup(context: Context) -> None:
+    context.adapter.cleanup(context.adapter_ref)
+
+
+@then("the adapter should have no tracked sandboxes")
+def step_adapter_no_sandboxes(context: Context) -> None:
+    assert len(context.adapter._sandboxes) == 0
+
+
+# ---------------------------------------------------------------------------
+# CustomStrategyConfig
+# ---------------------------------------------------------------------------
+
+
+@when("I attempt to create a CustomStrategyConfig with empty name")
+def step_config_empty_name(context: Context) -> None:
+    context.sandbox_error = None
+    try:
+        CustomStrategyConfig(name="", module="mod", class_name="Cls")
+    except ValueError as exc:
+        context.sandbox_error = exc
+
+
+@when("I attempt to create a CustomStrategyConfig with empty module")
+def step_config_empty_module(context: Context) -> None:
+    context.sandbox_error = None
+    try:
+        CustomStrategyConfig(name="x", module="", class_name="Cls")
+    except ValueError as exc:
+        context.sandbox_error = exc
+
+
+@when("I attempt to create a CustomStrategyConfig with empty class_name")
+def step_config_empty_class(context: Context) -> None:
+    context.sandbox_error = None
+    try:
+        CustomStrategyConfig(name="x", module="mod", class_name="")
+    except ValueError as exc:
+        context.sandbox_error = exc
+
+
+# ---------------------------------------------------------------------------
+# Factory custom strategy integration
+# ---------------------------------------------------------------------------
+
+
+@given('a SandboxFactory with a custom registry containing "{name}"')
+def step_factory_with_registry(context: Context, name: str) -> None:
+    registry = SandboxStrategyRegistry(allowed_prefixes=("cleveragents.",))
+    registry.register(
+        name,
+        "cleveragents.infrastructure.sandbox.strategy_adapter",
+        "BuiltInSandboxStrategyAdapter",
+    )
+    context.factory = SandboxFactory(custom_registry=registry)
+
+
+@given("a SandboxFactory without a custom registry")
+def step_factory_no_registry(context: Context) -> None:
+    context.factory = SandboxFactory()
+
+
+@then('the factory should report "{name}" as a custom strategy')
+def step_factory_has_custom(context: Context, name: str) -> None:
+    assert context.factory.has_custom_strategy(name)
+
+
+@then('the factory should not report "{name}" as a custom strategy')
+def step_factory_no_custom(context: Context, name: str) -> None:
+    assert not context.factory.has_custom_strategy(name)

robot/custom_sandbox_strategy.robot

@@ -0,0 +1,69 @@
+*** Settings ***
+Documentation     Integration tests for custom sandbox strategy registration.
+...               Covers SandboxStrategyProtocol, SandboxStrategyRegistry,
+...               BuiltInSandboxStrategyAdapter, CustomStrategyConfig, and
+...               SandboxFactory custom strategy integration.
+...               Based on issue #586.
+Resource          ${CURDIR}/common.resource
+Suite Setup       Setup Test Environment
+Suite Teardown    Cleanup Test Environment
+
+*** Variables ***
+${HELPER_SCRIPT}    robot/helper_custom_sandbox_strategy.py
+
+*** Test Cases ***
+SandboxRef Creation And Immutability
+    [Documentation]    Verify SandboxRef creation, properties, and frozen semantics
+    [Tags]    sandbox    strategy    ref
+    ${result}=    Run Process    ${PYTHON}    ${HELPER_SCRIPT}    sandbox-ref-lifecycle    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    sandbox-ref-lifecycle-ok
+
+DiffView And DiffEntry Models
+    [Documentation]    Verify DiffView and DiffEntry creation and validation
+    [Tags]    sandbox    strategy    diff
+    ${result}=    Run Process    ${PYTHON}    ${HELPER_SCRIPT}    diff-view-lifecycle    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    diff-view-lifecycle-ok
+
+SandboxStrategyProtocol Runtime Check
+    [Documentation]    Verify SandboxStrategyProtocol is runtime_checkable with 9 methods
+    [Tags]    sandbox    strategy    protocol
+    ${result}=    Run Process    ${PYTHON}    ${HELPER_SCRIPT}    protocol-check    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    protocol-check-ok
+
+SandboxStrategyRegistry Lifecycle
+    [Documentation]    Verify registry register, lookup, config-driven, protocol validation, clear
+    [Tags]    sandbox    strategy    registry
+    ${result}=    Run Process    ${PYTHON}    ${HELPER_SCRIPT}    registry-lifecycle    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    registry-lifecycle-ok
+
+BuiltInSandboxStrategyAdapter Full Lifecycle
+    [Documentation]    Verify adapter create/read/write/diff/cleanup with real filesystem
+    [Tags]    sandbox    strategy    adapter
+    ${result}=    Run Process    ${PYTHON}    ${HELPER_SCRIPT}    adapter-lifecycle    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    adapter-lifecycle-ok
+
+Adapter Checkpoint And Restore
+    [Documentation]    Verify adapter checkpoint save and restore operations
+    [Tags]    sandbox    strategy    adapter    checkpoint
+    ${result}=    Run Process    ${PYTHON}    ${HELPER_SCRIPT}    adapter-checkpoint    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    adapter-checkpoint-ok
+
+CustomStrategyConfig Validation
+    [Documentation]    Verify CustomStrategyConfig rejects empty name, module, class_name
+    [Tags]    sandbox    strategy    config
+    ${result}=    Run Process    ${PYTHON}    ${HELPER_SCRIPT}    config-validation    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    config-validation-ok
+
+SandboxFactory Custom Strategy Integration
+    [Documentation]    Verify SandboxFactory with custom registry lookup
+    [Tags]    sandbox    strategy    factory
+    ${result}=    Run Process    ${PYTHON}    ${HELPER_SCRIPT}    factory-custom-integration    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    factory-custom-integration-ok

robot/helper_custom_sandbox_strategy.py

@@ -0,0 +1,439 @@
+"""Helper utilities for custom sandbox strategy Robot integration tests.
+
+Covers end-to-end integration of:
+- SandboxStrategyProtocol and runtime checking
+- SandboxStrategyRegistry registration and lookup
+- BuiltInSandboxStrategyAdapter full lifecycle
+- CustomStrategyConfig validation
+- SandboxFactory custom strategy integration
+
+Based on issue #586.
+"""
+
+from __future__ import annotations
+
+import os
+import sys
+import tempfile
+from datetime import datetime
+
+from cleveragents.domain.models.core.resource import (
+    PhysVirt,
+    Resource,
+    ResourceCapabilities,
+)
+from cleveragents.domain.models.core.resource import (
+    SandboxStrategy as SandboxStrategyEnum,
+)
+from cleveragents.domain.models.core.sandbox_strategy import (
+    DiffEntry,
+    DiffView,
+    SandboxRef,
+    SandboxStrategyProtocol,
+)
+from cleveragents.infrastructure.plugins.exceptions import (
+    ProtocolMismatchError,
+)
+from cleveragents.infrastructure.sandbox.factory import SandboxFactory
+from cleveragents.infrastructure.sandbox.strategy_adapter import (
+    BuiltInSandboxStrategyAdapter,
+)
+from cleveragents.infrastructure.sandbox.strategy_registry import (
+    CustomStrategyConfig,
+    SandboxStrategyRegistry,
+)
+
+# ---------------------------------------------------------------------------
+# ULID helper
+# ---------------------------------------------------------------------------
+
+_CTR = 0
+
+
+def _ulid(name: str) -> str:
+    global _CTR
+    _CTR += 1
+    base = abs(hash(name)) % (10**20)
+    raw = f"{_CTR:06d}{base:020d}"
+    charset = "0123456789ABCDEFGHJKMNPQRSTVWXYZ"
+    result = ""
+    for ch in raw:
+        result += charset[int(ch)]
+    return (result + "0" * 26)[:26]
+
+
+def _make_resource(name: str, location: str) -> Resource:
+    return Resource(
+        resource_id=_ulid(name),
+        name=name,
+        resource_type_name="fs-directory",
+        classification=PhysVirt.PHYSICAL,
+        sandbox_strategy=SandboxStrategyEnum.NONE,
+        location=location,
+        capabilities=ResourceCapabilities(
+            readable=True,
+            writable=True,
+            sandboxable=True,
+            checkpointable=False,
+        ),
+    )
+
+
+# ---------------------------------------------------------------------------
+# Test: SandboxRef creation and immutability
+# ---------------------------------------------------------------------------
+
+
+def _sandbox_ref_lifecycle() -> None:
+    """Integration test: SandboxRef creation and properties."""
+    ref = SandboxRef(
+        sandbox_id="ref-001",
+        plan_id="plan-001",
+        resource_id="res-001",
+        created_at=datetime.now(),
+        metadata={"backend": "test"},
+    )
+    assert ref.sandbox_id == "ref-001"
+    assert ref.plan_id == "plan-001"
+    assert ref.metadata["backend"] == "test"
+
+    # Frozen check
+    try:
+        ref.sandbox_id = "mutated"  # type: ignore[misc]
+        print("FAIL: Expected FrozenInstanceError")
+        return
+    except (AttributeError, Exception):
+        pass
+
+    print("sandbox-ref-lifecycle-ok")
+
+
+# ---------------------------------------------------------------------------
+# Test: DiffView and DiffEntry
+# ---------------------------------------------------------------------------
+
+
+def _diff_view_lifecycle() -> None:
+    """Integration test: DiffView and DiffEntry models."""
+    entry = DiffEntry(path="test.py", operation="added", after=b"content")
+    assert entry.path == "test.py"
+    assert entry.operation == "added"
+
+    view = DiffView(
+        sandbox_id="dv-001",
+        entries=[entry],
+        summary="1 file added",
+    )
+    assert len(view.entries) == 1
+    assert view.sandbox_id == "dv-001"
+
+    # Empty path should fail
+    try:
+        DiffEntry(path="", operation="modified")
+        print("FAIL: Expected ValidationError")
+        return
+    except Exception:
+        pass
+
+    print("diff-view-lifecycle-ok")
+
+
+# ---------------------------------------------------------------------------
+# Test: SandboxStrategyProtocol runtime check
+# ---------------------------------------------------------------------------
+
+
+def _protocol_check() -> None:
+    """Integration test: Protocol runtime checking."""
+
+    class _Good:
+        def create(self, plan_id, resource):
+            return None
+
+        def read(self, ref, path):
+            return b""
+
+        def write(self, ref, path, content):
+            return None
+
+        def diff(self, ref):
+            return None
+
+        def commit(self, ref):
+            pass
+
+        def rollback(self, ref):
+            pass
+
+        def checkpoint(self, ref, checkpoint_id):
+            pass
+
+        def restore_checkpoint(self, ref, checkpoint_id):
+            pass
+
+        def cleanup(self, ref):
+            pass
+
+    class _Bad:
+        def create(self, plan_id, resource):
+            return None
+
+    good = _Good()
+    bad = _Bad()
+    assert isinstance(good, SandboxStrategyProtocol)
+    assert not isinstance(bad, SandboxStrategyProtocol)
+
+    print("protocol-check-ok")
+
+
+# ---------------------------------------------------------------------------
+# Test: SandboxStrategyRegistry
+# ---------------------------------------------------------------------------
+
+
+def _registry_lifecycle() -> None:
+    """Integration test: Registry registration, lookup, config, clear."""
+    registry = SandboxStrategyRegistry(allowed_prefixes=("cleveragents.",))
+
+    # Register
+    cls = registry.register(
+        "adapter",
+        "cleveragents.infrastructure.sandbox.strategy_adapter",
+        "BuiltInSandboxStrategyAdapter",
+    )
+    assert cls is BuiltInSandboxStrategyAdapter
+    assert registry.has("adapter")
+    assert "adapter" in registry.list_strategies()
+
+    # Get
+    assert registry.get("adapter") is BuiltInSandboxStrategyAdapter
+    assert registry.get("nonexistent") is None
+
+    # Non-protocol rejection
+    try:
+        registry.register(
+            "bad",
+            "cleveragents.infrastructure.sandbox.protocol",
+            "SandboxError",
+        )
+        print("FAIL: Expected ProtocolMismatchError")
+        return
+    except ProtocolMismatchError:
+        pass
+
+    # Config-driven registration
+    configs = {
+        "cfg1": {
+            "module": "cleveragents.infrastructure.sandbox.strategy_adapter",
+            "class": "BuiltInSandboxStrategyAdapter",
+        },
+    }
+    registered = registry.register_all_from_config(configs)
+    assert len(registered) == 1
+
+    # Clear
+    registry.clear()
+    assert len(registry.list_strategies()) == 0
+
+    print("registry-lifecycle-ok")
+
+
+# ---------------------------------------------------------------------------
+# Test: BuiltInSandboxStrategyAdapter full lifecycle
+# ---------------------------------------------------------------------------
+
+
+def _adapter_lifecycle() -> None:
+    """Integration test: Adapter create/read/write/diff/commit/cleanup."""
+    with tempfile.TemporaryDirectory() as tmpdir:
+        # Seed a file
+        with open(os.path.join(tmpdir, "seed.txt"), "w") as f:
+            f.write("seed")
+
+        adapter = BuiltInSandboxStrategyAdapter(strategy_name="copy_on_write")
+        resource = Resource(
+            resource_id=_ulid("adapter-res"),
+            name="adapter-res",
+            resource_type_name="fs-directory",
+            classification=PhysVirt.PHYSICAL,
+            sandbox_strategy=SandboxStrategyEnum.COPY_ON_WRITE,
+            location=tmpdir,
+            capabilities=ResourceCapabilities(
+                readable=True,
+                writable=True,
+                sandboxable=True,
+                checkpointable=False,
+            ),
+        )
+
+        # Create
+        ref = adapter.create("plan-adapter", resource)
+        assert isinstance(ref, SandboxRef)
+        assert ref.plan_id == "plan-adapter"
+
+        # Write
+        entry = adapter.write(ref, "new_file.txt", b"hello")
+        assert isinstance(entry, DiffEntry)
+        assert entry.operation == "added"
+
+        # Read
+        data = adapter.read(ref, "new_file.txt")
+        assert data == b"hello"
+
+        # Diff
+        view = adapter.diff(ref)
+        assert isinstance(view, DiffView)
+
+        # Cleanup
+        adapter.cleanup(ref)
+        assert ref.sandbox_id not in adapter._sandboxes
+
+    print("adapter-lifecycle-ok")
+
+
+# ---------------------------------------------------------------------------
+# Test: Adapter checkpoint/restore
+# ---------------------------------------------------------------------------
+
+
+def _adapter_checkpoint() -> None:
+    """Integration test: Adapter checkpoint and restore."""
+    with tempfile.TemporaryDirectory() as tmpdir:
+        with open(os.path.join(tmpdir, "seed.txt"), "w") as f:
+            f.write("seed")
+
+        adapter = BuiltInSandboxStrategyAdapter(strategy_name="copy_on_write")
+        resource = Resource(
+            resource_id=_ulid("cp-res"),
+            name="cp-res",
+            resource_type_name="fs-directory",
+            classification=PhysVirt.PHYSICAL,
+            sandbox_strategy=SandboxStrategyEnum.COPY_ON_WRITE,
+            location=tmpdir,
+            capabilities=ResourceCapabilities(
+                readable=True,
+                writable=True,
+                sandboxable=True,
+                checkpointable=False,
+            ),
+        )
+
+        ref = adapter.create("plan-cp", resource)
+        adapter.write(ref, "cp.txt", b"v1")
+        adapter.checkpoint(ref, "cp-1")
+        adapter.write(ref, "cp.txt", b"v2")
+
+        # Verify v2
+        assert adapter.read(ref, "cp.txt") == b"v2"
+
+        # Restore
+        adapter.restore_checkpoint(ref, "cp-1")
+        assert adapter.read(ref, "cp.txt") == b"v1"
+
+        adapter.cleanup(ref)
+
+    print("adapter-checkpoint-ok")
+
+
+# ---------------------------------------------------------------------------
+# Test: CustomStrategyConfig validation
+# ---------------------------------------------------------------------------
+
+
+def _config_validation() -> None:
+    """Integration test: CustomStrategyConfig validation."""
+    # Valid
+    cfg = CustomStrategyConfig(
+        name="test", module="cleveragents.test", class_name="TestCls"
+    )
+    assert cfg.name == "test"
+
+    # Empty name
+    try:
+        CustomStrategyConfig(name="", module="mod", class_name="Cls")
+        print("FAIL: Expected ValueError for empty name")
+        return
+    except ValueError:
+        pass
+
+    # Empty module
+    try:
+        CustomStrategyConfig(name="x", module="", class_name="Cls")
+        print("FAIL: Expected ValueError for empty module")
+        return
+    except ValueError:
+        pass
+
+    # Empty class
+    try:
+        CustomStrategyConfig(name="x", module="mod", class_name="")
+        print("FAIL: Expected ValueError for empty class")
+        return
+    except ValueError:
+        pass
+
+    print("config-validation-ok")
+
+
+# ---------------------------------------------------------------------------
+# Test: SandboxFactory custom strategy integration
+# ---------------------------------------------------------------------------
+
+
+def _factory_custom_integration() -> None:
+    """Integration test: SandboxFactory with custom strategy registry."""
+    registry = SandboxStrategyRegistry(allowed_prefixes=("cleveragents.",))
+    registry.register(
+        "my_custom",
+        "cleveragents.infrastructure.sandbox.strategy_adapter",
+        "BuiltInSandboxStrategyAdapter",
+    )
+
+    factory = SandboxFactory(custom_registry=registry)
+    assert factory.has_custom_strategy("my_custom")
+    assert not factory.has_custom_strategy("nonexistent")
+
+    cls = factory.get_custom_strategy_class("my_custom")
+    assert cls is BuiltInSandboxStrategyAdapter
+
+    # Factory without registry
+    plain_factory = SandboxFactory()
+    assert not plain_factory.has_custom_strategy("anything")
+    assert plain_factory.get_custom_strategy_class("anything") is None
+
+    print("factory-custom-integration-ok")
+
+
+# ---------------------------------------------------------------------------
+# Main dispatcher
+# ---------------------------------------------------------------------------
+
+_TESTS = {
+    "sandbox-ref-lifecycle": _sandbox_ref_lifecycle,
+    "diff-view-lifecycle": _diff_view_lifecycle,
+    "protocol-check": _protocol_check,
+    "registry-lifecycle": _registry_lifecycle,
+    "adapter-lifecycle": _adapter_lifecycle,
+    "adapter-checkpoint": _adapter_checkpoint,
+    "config-validation": _config_validation,
+    "factory-custom-integration": _factory_custom_integration,
+}
+
+
+def main() -> None:
+    if len(sys.argv) < 2:
+        print(f"Usage: {sys.argv[0]} <test-name>")
+        print(f"Available: {', '.join(sorted(_TESTS))}")
+        sys.exit(1)
+
+    test_name = sys.argv[1]
+    if test_name not in _TESTS:
+        print(f"Unknown test: {test_name}")
+        print(f"Available: {', '.join(sorted(_TESTS))}")
+        sys.exit(1)
+
+    _TESTS[test_name]()
+
+
+if __name__ == "__main__":
+    main()

src/cleveragents/domain/models/core/__init__.py

@@ -248,6 +248,14 @@ from cleveragents.domain.models.core.safety_profile import (
     SafetyProfileRef,
     resolve_safety_profile,
 )
+from cleveragents.domain.models.core.sandbox_strategy import (
+    DiffEntry as SandboxDiffEntry,
+)
+from cleveragents.domain.models.core.sandbox_strategy import (
+    DiffView,
+    SandboxRef,
+    SandboxStrategyProtocol,
+)
 from cleveragents.domain.models.core.session import (
     MessageRole,
     Session,
@@ -354,6 +362,7 @@ __all__ = [
     "DiffBuilder",
     "DiffEntry",
     "DiffSerializer",
+    "DiffView",
     "DoDCriterion",
     "DoDEvaluator",
     "DoDResult",
@@ -453,7 +462,10 @@ __all__ = [
     "SafetyProfile",
     "SafetyProfileProvenance",
     "SafetyProfileRef",
+    "SandboxDiffEntry",
+    "SandboxRef",
     "SandboxStrategy",
+    "SandboxStrategyProtocol",
     "ScoredFragment",
     "ServiceRetryPolicy",
     "ServiceRetryPolicyRegistry",

src/cleveragents/domain/models/core/sandbox_strategy.py

@@ -0,0 +1,235 @@
+"""Custom sandbox strategy Protocol and supporting models.
+
+Defines the ``SandboxStrategyProtocol`` that custom sandbox strategies must
+implement, along with ``SandboxRef`` and ``DiffView`` value objects used in
+the Protocol method signatures.
+
+The Protocol specifies 9 methods covering the full sandbox lifecycle:
+create, read, write, diff, commit, rollback, checkpoint,
+restore_checkpoint, cleanup.
+
+Custom strategies are registered via configuration::
+
+    sandbox.custom_strategies.<name>.module = "my_module.path"
+    sandbox.custom_strategies.<name>.class  = "MySandboxStrategy"
+
+See Also:
+    - :class:`~cleveragents.infrastructure.sandbox.protocol.Sandbox`
+      The existing infrastructure-level sandbox protocol.
+    - :class:`~cleveragents.domain.models.core.resource.SandboxStrategy`
+      The StrEnum defining built-in strategy names.
+    - Specification § Custom Sandbox Strategies (lines 46171-46186)
+
+Based on issue #586.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from datetime import datetime
+from typing import Protocol, runtime_checkable
+
+from pydantic import BaseModel, ConfigDict, Field
+
+from cleveragents.domain.models.core.resource import Resource
+
+# ---------------------------------------------------------------------------
+# SandboxRef — opaque reference to a sandbox instance
+# ---------------------------------------------------------------------------
+
+
+@dataclass(frozen=True)
+class SandboxRef:
+    """Opaque reference to a sandbox instance.
+
+    Returned by :meth:`SandboxStrategyProtocol.create` and passed to
+    all subsequent operations.  The ``sandbox_id`` is a unique
+    identifier (typically a ULID); ``metadata`` allows strategy
+    implementations to carry extra state across calls.
+
+    Attributes:
+        sandbox_id: Unique identifier for this sandbox instance.
+        plan_id: The plan that owns this sandbox.
+        resource_id: The resource being sandboxed.
+        created_at: When the sandbox was created.
+        metadata: Strategy-specific opaque data.
+    """
+
+    sandbox_id: str
+    plan_id: str
+    resource_id: str
+    created_at: datetime
+    metadata: dict[str, object] = field(default_factory=dict)
+
+
+# ---------------------------------------------------------------------------
+# DiffView — structured view of sandbox changes
+# ---------------------------------------------------------------------------
+
+
+class DiffEntry(BaseModel):
+    """A single file-level diff entry within a :class:`DiffView`.
+
+    Attributes:
+        path: Relative path of the changed file.
+        operation: Type of change (``added``, ``modified``, ``deleted``).
+        before: Content before the change (``None`` for additions).
+        after: Content after the change (``None`` for deletions).
+    """
+
+    path: str = Field(..., min_length=1, description="Relative file path")
+    operation: str = Field(
+        ...,
+        description="Change type: 'added', 'modified', or 'deleted'",
+    )
+    before: bytes | None = Field(
+        default=None,
+        description="Content before change (None for additions)",
+    )
+    after: bytes | None = Field(
+        default=None,
+        description="Content after change (None for deletions)",
+    )
+
+    model_config = ConfigDict(frozen=True)
+
+
+class DiffView(BaseModel):
+    """Structured view of all changes in a sandbox.
+
+    Produced by :meth:`SandboxStrategyProtocol.diff`.
+
+    Attributes:
+        sandbox_id: The sandbox these diffs belong to.
+        entries: Individual file-level diff entries.
+        summary: Human-readable summary of the changes.
+    """
+
+    sandbox_id: str = Field(..., description="Sandbox identifier")
+    entries: list[DiffEntry] = Field(
+        default_factory=list,
+        description="Per-file diff entries",
+    )
+    summary: str = Field(
+        default="",
+        description="Human-readable change summary",
+    )
+
+    model_config = ConfigDict(frozen=True)
+
+
+# ---------------------------------------------------------------------------
+# SandboxStrategyProtocol — the 9-method custom strategy interface
+# ---------------------------------------------------------------------------
+
+
+@runtime_checkable
+class SandboxStrategyProtocol(Protocol):
+    """Protocol for custom sandbox isolation strategies.
+
+    All custom sandbox strategies must implement these 9 methods.
+    Strategies are registered via configuration and resolved at
+    runtime by the :class:`SandboxStrategyRegistry`.
+
+    Lifecycle::
+
+        ref = strategy.create(plan_id, resource)   # create sandbox
+        data = strategy.read(ref, "file.txt")       # read from sandbox
+        change = strategy.write(ref, "f.txt", b"x") # write to sandbox
+        view = strategy.diff(ref)                    # inspect changes
+        strategy.checkpoint(ref, "cp-1")             # save checkpoint
+        strategy.restore_checkpoint(ref, "cp-1")     # restore checkpoint
+        strategy.commit(ref)                         # apply changes
+        strategy.rollback(ref)                       # discard changes
+        strategy.cleanup(ref)                        # release resources
+    """
+
+    def create(self, plan_id: str, resource: Resource) -> SandboxRef:
+        """Create a new sandbox for the given plan and resource.
+
+        Args:
+            plan_id: Identifier of the plan requesting the sandbox.
+            resource: The resource to sandbox.
+
+        Returns:
+            An opaque :class:`SandboxRef` for subsequent operations.
+        """
+        ...
+
+    def read(self, ref: SandboxRef, path: str) -> bytes:
+        """Read file content from the sandbox.
+
+        Args:
+            ref: Sandbox reference from :meth:`create`.
+            path: Relative path within the sandbox.
+
+        Returns:
+            Raw file content as bytes.
+        """
+        ...
+
+    def write(self, ref: SandboxRef, path: str, content: bytes) -> DiffEntry:
+        """Write content to a file in the sandbox.
+
+        Args:
+            ref: Sandbox reference from :meth:`create`.
+            path: Relative path within the sandbox.
+            content: Raw content to write.
+
+        Returns:
+            A :class:`DiffEntry` describing the change.
+        """
+        ...
+
+    def diff(self, ref: SandboxRef) -> DiffView:
+        """Compute a structured diff of all sandbox changes.
+
+        Args:
+            ref: Sandbox reference from :meth:`create`.
+
+        Returns:
+            A :class:`DiffView` with per-file entries.
+        """
+        ...
+
+    def commit(self, ref: SandboxRef) -> None:
+        """Apply sandbox changes to the original resource.
+
+        Args:
+            ref: Sandbox reference from :meth:`create`.
+        """
+        ...
+
+    def rollback(self, ref: SandboxRef) -> None:
+        """Discard all sandbox changes.
+
+        Args:
+            ref: Sandbox reference from :meth:`create`.
+        """
+        ...
+
+    def checkpoint(self, ref: SandboxRef, checkpoint_id: str) -> None:
+        """Save a named checkpoint of the current sandbox state.
+
+        Args:
+            ref: Sandbox reference from :meth:`create`.
+            checkpoint_id: Unique identifier for the checkpoint.
+        """
+        ...
+
+    def restore_checkpoint(self, ref: SandboxRef, checkpoint_id: str) -> None:
+        """Restore the sandbox to a previously saved checkpoint.
+
+        Args:
+            ref: Sandbox reference from :meth:`create`.
+            checkpoint_id: Identifier of the checkpoint to restore.
+        """
+        ...
+
+    def cleanup(self, ref: SandboxRef) -> None:
+        """Release all sandbox resources and artefacts.
+
+        Args:
+            ref: Sandbox reference from :meth:`create`.
+        """
+        ...

src/cleveragents/infrastructure/sandbox/__init__.py

@@ -6,6 +6,7 @@ and multiple strategy implementations (git worktree, filesystem copy, no-op).
 Stage B3 of the implementation plan.  Updated in TASK-006 (B4 rework).
 Updated in M4 to add checkpoint/rollback hooks.
 Updated in M6 to add sandbox boundary algebra (#548).
+Updated in M6+ to add custom sandbox strategy registration (#586).
 """
 
 from cleveragents.infrastructure.sandbox.boundary import (
@@ -40,14 +41,23 @@ from cleveragents.infrastructure.sandbox.protocol import (
     SandboxError,
     SandboxStatus,
 )
+from cleveragents.infrastructure.sandbox.strategy_adapter import (
+    BuiltInSandboxStrategyAdapter,
+)
+from cleveragents.infrastructure.sandbox.strategy_registry import (
+    CustomStrategyConfig,
+    SandboxStrategyRegistry,
+)
 from cleveragents.infrastructure.sandbox.transaction_sandbox import TransactionSandbox
 
 __all__ = [
     "BoundaryCache",
+    "BuiltInSandboxStrategyAdapter",
     "CheckpointManager",
     "Checkpointable",
     "CommitResult",
     "CopyOnWriteSandbox",
+    "CustomStrategyConfig",
     "GitMergeStrategy",
     "GitWorktreeSandbox",
     "JsonMergeStrategy",
@@ -63,6 +73,7 @@ __all__ = [
     "SandboxFactory",
     "SandboxManager",
     "SandboxStatus",
+    "SandboxStrategyRegistry",
     "SequentialMergeStrategy",
     "TransactionSandbox",
     "compute_sandbox_domains",

src/cleveragents/infrastructure/sandbox/factory.py

@@ -6,12 +6,13 @@ available, this factory will accept ``Resource`` objects directly; until
 then it operates on raw parameters.
 
 Stage B3.6 of the implementation plan.  Updated in TASK-006 (B4 rework).
+Updated in M6+ to support custom sandbox strategy resolution (#586).
 """
 
 from __future__ import annotations
 
 import logging
-from typing import Literal
+from typing import TYPE_CHECKING, Literal
 
 from cleveragents.infrastructure.sandbox.copy_on_write import CopyOnWriteSandbox
 from cleveragents.infrastructure.sandbox.git_worktree import GitWorktreeSandbox
@@ -21,6 +22,11 @@ from cleveragents.infrastructure.sandbox.protocol import (
 )
 from cleveragents.infrastructure.sandbox.transaction_sandbox import TransactionSandbox
 
+if TYPE_CHECKING:
+    from cleveragents.infrastructure.sandbox.strategy_registry import (
+        SandboxStrategyRegistry,
+    )
+
 logger = logging.getLogger(__name__)
 
 # Strategy string constants aligned with spec SandboxStrategy enum (5 values)
@@ -70,8 +76,25 @@ class SandboxFactory:
     - ``"transaction_rollback"`` -> :class:`TransactionSandbox`
 
     ``"snapshot"`` raises ``NotImplementedError``.
+
+    Custom strategies registered in a :class:`SandboxStrategyRegistry` can
+    be resolved by passing the registry at construction time.
     """
 
+    def __init__(
+        self,
+        *,
+        custom_registry: SandboxStrategyRegistry | None = None,
+    ) -> None:
+        """Initialise the factory.
+
+        Args:
+            custom_registry: Optional registry of custom sandbox strategies.
+                When provided, the factory will fall back to this registry
+                for strategy names not matched by built-in strategies.
+        """
+        self._custom_registry = custom_registry
+
     def create_sandbox(
         self,
         resource_id: str,
@@ -128,6 +151,34 @@ class SandboxFactory:
 
         raise ValueError(f"Unknown sandbox strategy: {sandbox_strategy}")
 
+    # -- custom strategy lookup ----------------------------------------------
+
+    def has_custom_strategy(self, name: str) -> bool:
+        """Check whether a custom strategy is registered.
+
+        Args:
+            name: Strategy name to check.
+
+        Returns:
+            ``True`` if a custom strategy with that name exists.
+        """
+        if self._custom_registry is None:
+            return False
+        return self._custom_registry.has(name)
+
+    def get_custom_strategy_class(self, name: str) -> type[object] | None:
+        """Look up a custom strategy class by name.
+
+        Args:
+            name: Strategy name.
+
+        Returns:
+            The strategy class, or ``None`` if not found.
+        """
+        if self._custom_registry is None:
+            return None
+        return self._custom_registry.get(name)
+
     # -- validation helpers --------------------------------------------------
 
     @staticmethod

src/cleveragents/infrastructure/sandbox/strategy_adapter.py

@@ -0,0 +1,252 @@
+"""Adapter that wraps existing Sandbox implementations as SandboxStrategyProtocol.
+
+Bridges the existing :class:`Sandbox` protocol (infrastructure layer) to the
+new :class:`SandboxStrategyProtocol` (domain layer).  This allows built-in
+sandbox implementations (NoSandbox, CopyOnWriteSandbox, GitWorktreeSandbox,
+TransactionSandbox) to be used through the unified custom-strategy API.
+
+Based on issue #586.
+"""
+
+from __future__ import annotations
+
+import logging
+import os
+from datetime import datetime
+
+from cleveragents.domain.models.core.resource import Resource
+from cleveragents.domain.models.core.sandbox_strategy import (
+    DiffEntry,
+    DiffView,
+    SandboxRef,
+)
+from cleveragents.infrastructure.sandbox.factory import (
+    SandboxFactory,
+    SandboxStrategyStr,
+)
+from cleveragents.infrastructure.sandbox.protocol import (
+    Sandbox,
+)
+
+logger = logging.getLogger(__name__)
+
+
+class BuiltInSandboxStrategyAdapter:
+    """Adapts a built-in :class:`Sandbox` implementation to the 9-method Protocol.
+
+    Wraps the existing ``Sandbox`` lifecycle (create, get_path, commit,
+    rollback, cleanup) and adds read/write/diff/checkpoint/restore
+    operations on top.  Checkpoint storage is in-memory per adapter
+    instance.
+
+    This adapter satisfies :class:`SandboxStrategyProtocol` structurally.
+
+    Args:
+        strategy_name: The built-in strategy name (e.g. ``"none"``).
+        factory: Optional :class:`SandboxFactory` to create sandbox instances.
+    """
+
+    def __init__(
+        self,
+        strategy_name: SandboxStrategyStr = "none",
+        factory: SandboxFactory | None = None,
+    ) -> None:
+        self._strategy_name: SandboxStrategyStr = strategy_name
+        self._factory = factory or SandboxFactory()
+        self._sandboxes: dict[str, Sandbox] = {}
+        self._checkpoints: dict[str, dict[str, bytes]] = {}
+
+    def create(self, plan_id: str, resource: Resource) -> SandboxRef:
+        """Create a sandbox using the built-in factory and return a ref.
+
+        Args:
+            plan_id: Identifier of the plan requesting the sandbox.
+            resource: The resource to sandbox.
+
+        Returns:
+            A :class:`SandboxRef` wrapping the internal sandbox instance.
+        """
+        location = resource.location or ""
+        sandbox = self._factory.create_sandbox(
+            resource_id=resource.resource_id,
+            original_path=location,
+            sandbox_strategy=self._strategy_name,
+        )
+        ctx = sandbox.create(plan_id)
+        self._sandboxes[ctx.sandbox_id] = sandbox
+
+        return SandboxRef(
+            sandbox_id=ctx.sandbox_id,
+            plan_id=plan_id,
+            resource_id=resource.resource_id,
+            created_at=datetime.now(),
+            metadata={"strategy": self._strategy_name},
+        )
+
+    def read(self, ref: SandboxRef, path: str) -> bytes:
+        """Read file content from the sandbox.
+
+        Args:
+            ref: Sandbox reference.
+            path: Relative file path.
+
+        Returns:
+            File content as bytes.
+
+        Raises:
+            FileNotFoundError: If the file does not exist.
+            KeyError: If the sandbox ref is unknown.
+        """
+        sandbox = self._sandboxes[ref.sandbox_id]
+        full_path = sandbox.get_path(path)
+        with open(full_path, "rb") as fh:
+            return fh.read()
+
+    def write(self, ref: SandboxRef, path: str, content: bytes) -> DiffEntry:
+        """Write content to a file in the sandbox.
+
+        Args:
+            ref: Sandbox reference.
+            path: Relative file path.
+            content: Content to write.
+
+        Returns:
+            A :class:`DiffEntry` describing the change.
+        """
+        sandbox = self._sandboxes[ref.sandbox_id]
+        full_path = sandbox.get_path(path)
+
+        existed = os.path.exists(full_path)
+        before: bytes | None = None
+        if existed:
+            with open(full_path, "rb") as fh:
+                before = fh.read()
+
+        parent_dir = os.path.dirname(full_path)
+        if parent_dir:
+            os.makedirs(parent_dir, exist_ok=True)
+
+        with open(full_path, "wb") as fh:
+            fh.write(content)
+
+        operation = "modified" if existed else "added"
+        return DiffEntry(
+            path=path,
+            operation=operation,
+            before=before,
+            after=content,
+        )
+
+    def diff(self, ref: SandboxRef) -> DiffView:
+        """Return a diff view of sandbox changes.
+
+        For built-in strategies, returns a minimal summary since the
+        underlying Sandbox protocol does not expose diff information
+        until commit.
+
+        Args:
+            ref: Sandbox reference.
+
+        Returns:
+            A :class:`DiffView` (may be empty for built-in strategies).
+        """
+        return DiffView(
+            sandbox_id=ref.sandbox_id,
+            entries=[],
+            summary="Built-in strategy; diff computed at commit time.",
+        )
+
+    def commit(self, ref: SandboxRef) -> None:
+        """Commit sandbox changes via the underlying Sandbox.
+
+        Args:
+            ref: Sandbox reference.
+        """
+        sandbox = self._sandboxes[ref.sandbox_id]
+        sandbox.commit()
+
+    def rollback(self, ref: SandboxRef) -> None:
+        """Rollback sandbox changes via the underlying Sandbox.
+
+        Args:
+            ref: Sandbox reference.
+        """
+        sandbox = self._sandboxes[ref.sandbox_id]
+        sandbox.rollback()
+
+    def checkpoint(self, ref: SandboxRef, checkpoint_id: str) -> None:
+        """Save a named checkpoint (in-memory snapshot for adapter).
+
+        Args:
+            ref: Sandbox reference.
+            checkpoint_id: Unique checkpoint identifier.
+        """
+        key = f"{ref.sandbox_id}:{checkpoint_id}"
+        sandbox = self._sandboxes[ref.sandbox_id]
+
+        # Snapshot current sandbox state if sandbox is filesystem-based
+        ctx = sandbox.context
+        snapshot: dict[str, bytes] = {}
+        if ctx is not None and os.path.isdir(ctx.sandbox_path):
+            for dirpath, _dirs, files in os.walk(ctx.sandbox_path):
+                for fname in files:
+                    full = os.path.join(dirpath, fname)
+                    rel = os.path.relpath(full, ctx.sandbox_path)
+                    with open(full, "rb") as fh:
+                        snapshot[rel] = fh.read()
+
+        self._checkpoints[key] = snapshot
+        logger.debug(
+            "Checkpoint saved: sandbox=%s checkpoint=%s files=%d",
+            ref.sandbox_id,
+            checkpoint_id,
+            len(snapshot),
+        )
+
+    def restore_checkpoint(self, ref: SandboxRef, checkpoint_id: str) -> None:
+        """Restore sandbox state from a named checkpoint.
+
+        Args:
+            ref: Sandbox reference.
+            checkpoint_id: Identifier of the checkpoint to restore.
+
+        Raises:
+            KeyError: If the checkpoint does not exist.
+        """
+        key = f"{ref.sandbox_id}:{checkpoint_id}"
+        snapshot = self._checkpoints[key]
+
+        sandbox = self._sandboxes[ref.sandbox_id]
+        ctx = sandbox.context
+        if ctx is not None and os.path.isdir(ctx.sandbox_path):
+            # Restore files from snapshot
+            for rel_path, content in snapshot.items():
+                full = os.path.join(ctx.sandbox_path, rel_path)
+                parent = os.path.dirname(full)
+                if parent:
+                    os.makedirs(parent, exist_ok=True)
+                with open(full, "wb") as fh:
+                    fh.write(content)
+
+        logger.debug(
+            "Checkpoint restored: sandbox=%s checkpoint=%s",
+            ref.sandbox_id,
+            checkpoint_id,
+        )
+
+    def cleanup(self, ref: SandboxRef) -> None:
+        """Cleanup sandbox resources.
+
+        Args:
+            ref: Sandbox reference.
+        """
+        sandbox = self._sandboxes.pop(ref.sandbox_id, None)
+        if sandbox is not None:
+            sandbox.cleanup()
+
+        # Remove associated checkpoints
+        keys_to_remove = [
+            k for k in self._checkpoints if k.startswith(f"{ref.sandbox_id}:")
+        ]
+        for k in keys_to_remove:
+            del self._checkpoints[k]

src/cleveragents/infrastructure/sandbox/strategy_registry.py

@@ -0,0 +1,298 @@
+"""Config-driven custom sandbox strategy registry.
+
+Provides :class:`SandboxStrategyRegistry` for registering, validating,
+and resolving custom sandbox strategies.  Strategies are registered
+either programmatically or via configuration keys::
+
+    sandbox.custom_strategies.<name>.module = "my_package.my_module"
+    sandbox.custom_strategies.<name>.class  = "MySandboxClass"
+
+All custom strategies are validated against the
+:class:`SandboxStrategyProtocol` at registration time using
+``@runtime_checkable`` Protocol checks.
+
+Thread-safe via ``threading.RLock``.
+
+Based on issue #586.
+"""
+
+from __future__ import annotations
+
+import logging
+import threading
+from typing import Any
+
+from cleveragents.infrastructure.plugins.exceptions import (
+    PluginLoadError,
+    ProtocolMismatchError,
+)
+from cleveragents.infrastructure.plugins.loader import PluginLoader
+
+logger = logging.getLogger(__name__)
+
+
+# ---------------------------------------------------------------------------
+# Custom strategy configuration model
+# ---------------------------------------------------------------------------
+
+
+class CustomStrategyConfig:
+    """Configuration for a single custom sandbox strategy.
+
+    Attributes:
+        name: Strategy name (used as the key in lookups).
+        module: Python module path containing the strategy class.
+        class_name: Name of the class within the module.
+        options: Optional strategy-specific configuration.
+    """
+
+    __slots__ = ("class_name", "module", "name", "options")
+
+    def __init__(
+        self,
+        name: str,
+        module: str,
+        class_name: str,
+        options: dict[str, Any] | None = None,
+    ) -> None:
+        if not name:
+            raise ValueError("Strategy name cannot be empty")
+        if not module:
+            raise ValueError("Strategy module cannot be empty")
+        if not class_name:
+            raise ValueError("Strategy class_name cannot be empty")
+
+        self.name = name
+        self.module = module
+        self.class_name = class_name
+        self.options = options or {}
+
+
+# ---------------------------------------------------------------------------
+# SandboxStrategyRegistry
+# ---------------------------------------------------------------------------
+
+
+class SandboxStrategyRegistry:
+    """Registry for custom sandbox strategies with Protocol validation.
+
+    Maintains a thread-safe mapping of strategy names to their
+    implementing classes.  Strategies are validated against the
+    :class:`SandboxStrategyProtocol` at registration time.
+
+    Built-in strategies (``none``, ``git_worktree``, ``copy_on_write``,
+    ``transaction_rollback``) are not managed by this registry; they
+    are handled by :class:`SandboxFactory`.
+
+    Usage::
+
+        registry = SandboxStrategyRegistry()
+        registry.register(
+            "my_strategy",
+            "cleveragents.my_module",
+            "MySandbox",
+        )
+        cls = registry.get("my_strategy")
+
+    Args:
+        loader: Optional :class:`PluginLoader` for dynamic imports.
+        allowed_prefixes: Module prefix allowlist for the loader.
+    """
+
+    def __init__(
+        self,
+        *,
+        loader: PluginLoader | None = None,
+        allowed_prefixes: tuple[str, ...] | None = None,
+    ) -> None:
+        self._lock = threading.RLock()
+        self._loader = loader or PluginLoader(allowed_prefixes=allowed_prefixes)
+        self._strategies: dict[str, type[Any]] = {}
+
+    # ------------------------------------------------------------------
+    # Registration
+    # ------------------------------------------------------------------
+
+    def register(
+        self,
+        name: str,
+        module_path: str,
+        class_name: str,
+    ) -> type[Any]:
+        """Register a custom sandbox strategy.
+
+        Loads the class via the plugin loader, validates it against
+        :class:`SandboxStrategyProtocol`, and stores it.
+
+        Args:
+            name: Strategy name for lookup.
+            module_path: Fully-qualified module path.
+            class_name: Class name within the module.
+
+        Returns:
+            The loaded and validated class.
+
+        Raises:
+            ValueError: If *name* is empty.
+            PluginLoadError: If the class cannot be imported.
+            ProtocolMismatchError: If the class does not satisfy
+                the SandboxStrategyProtocol.
+        """
+        if not name:
+            raise ValueError("Strategy name cannot be empty")
+
+        cls = self._loader.load_class(module_path, class_name)
+        self._validate_protocol(cls)
+
+        with self._lock:
+            self._strategies[name] = cls
+            logger.info(
+                "Registered custom sandbox strategy: name=%s class=%s.%s",
+                name,
+                module_path,
+                class_name,
+            )
+
+        return cls
+
+    def register_from_config(self, config: CustomStrategyConfig) -> type[Any]:
+        """Register a strategy from a :class:`CustomStrategyConfig`.
+
+        Args:
+            config: Configuration describing the strategy.
+
+        Returns:
+            The loaded and validated class.
+        """
+        return self.register(config.name, config.module, config.class_name)
+
+    def register_all_from_config(
+        self,
+        configs: dict[str, dict[str, Any]],
+    ) -> list[str]:
+        """Register multiple strategies from a config dictionary.
+
+        The dictionary maps strategy names to dicts with ``module``
+        and ``class`` keys::
+
+            {
+                "my_strategy": {
+                    "module": "my_package.module",
+                    "class": "MyStrategy",
+                },
+            }
+
+        Args:
+            configs: Strategy configuration dictionary.
+
+        Returns:
+            List of successfully registered strategy names.
+        """
+        registered: list[str] = []
+        for name, cfg in configs.items():
+            module = cfg.get("module", "")
+            class_name = cfg.get("class", "")
+            if not module or not class_name:
+                logger.warning(
+                    "Skipping custom strategy '%s': missing module or class",
+                    name,
+                )
+                continue
+
+            try:
+                self.register(name, module, class_name)
+                registered.append(name)
+            except (PluginLoadError, ProtocolMismatchError, ValueError) as exc:
+                logger.error(
+                    "Failed to register custom strategy '%s': %s",
+                    name,
+                    exc,
+                )
+
+        return registered
+
+    # ------------------------------------------------------------------
+    # Lookup
+    # ------------------------------------------------------------------
+
+    def get(self, name: str) -> type[Any] | None:
+        """Look up a registered custom strategy class.
+
+        Args:
+            name: Strategy name.
+
+        Returns:
+            The class, or ``None`` if not registered.
+        """
+        with self._lock:
+            return self._strategies.get(name)
+
+    def has(self, name: str) -> bool:
+        """Check whether a strategy is registered.
+
+        Args:
+            name: Strategy name.
+
+        Returns:
+            ``True`` if the strategy is registered.
+        """
+        with self._lock:
+            return name in self._strategies
+
+    def list_strategies(self) -> list[str]:
+        """Return names of all registered custom strategies.
+
+        Returns:
+            Sorted list of strategy names.
+        """
+        with self._lock:
+            return sorted(self._strategies.keys())
+
+    def clear(self) -> None:
+        """Remove all registered custom strategies."""
+        with self._lock:
+            self._strategies.clear()
+            logger.debug("Custom sandbox strategy registry cleared")
+
+    # ------------------------------------------------------------------
+    # Validation
+    # ------------------------------------------------------------------
+
+    @staticmethod
+    def _validate_protocol(cls: type[Any]) -> None:
+        """Validate that a class satisfies :class:`SandboxStrategyProtocol`.
+
+        Uses structural subtyping: checks that all 9 required methods
+        are present as callable attributes on the class.
+
+        Args:
+            cls: The class to validate.
+
+        Raises:
+            ProtocolMismatchError: If the class is missing required methods.
+        """
+        required_methods = (
+            "create",
+            "read",
+            "write",
+            "diff",
+            "commit",
+            "rollback",
+            "checkpoint",
+            "restore_checkpoint",
+            "cleanup",
+        )
+
+        missing = [
+            method
+            for method in required_methods
+            if not callable(getattr(cls, method, None))
+        ]
+
+        if missing:
+            msg = (
+                f"Class '{cls.__name__}' does not satisfy "
+                f"SandboxStrategyProtocol. Missing methods: "
+                f"{', '.join(missing)}"
+            )
+            raise ProtocolMismatchError(msg)