feat(sandbox): implement sandbox boundary algebra and domain computation

Implement sandbox_boundary(r) function that walks up containment edges
in the resource DAG to the nearest sandboxable ancestor, enabling
resources sharing a boundary to share one sandbox instance.

Changes:
- Add boundary.py: is_sandbox_boundary(), sandbox_boundary(),
  compute_sandbox_domains(), BoundaryCache (thread-safe, per-execution)
- Update SandboxManager: resolve_sandbox_key() and
  get_or_create_sandbox_for_resource() key by (plan_id, boundary_id)
  instead of (plan_id, resource_id); boundary cache lifecycle methods
- Define "sandboxable" via ResourceCapabilities.sandboxable + non-none
  sandbox_strategy as per specification section 24659-24674
- Export new symbols from sandbox __init__.py
- Add vulture whitelist entries for new public API

Tests:
- 26 Behave BDD scenarios (features/sandbox_boundary_algebra.feature)
- 5 Robot Framework integration tests (robot/sandbox_boundary_algebra.robot)
- ASV benchmarks for boundary walk, domain grouping, and cache performance

ISSUES CLOSED: #548

benchmarks/sandbox_boundary_bench.py

@@ -0,0 +1,196 @@
+"""ASV benchmarks for sandbox boundary algebra.
+
+Measures the performance of:
+- ``is_sandbox_boundary()`` predicate
+- ``sandbox_boundary()`` walk for shallow and deep DAGs
+- ``compute_sandbox_domains()`` grouping
+- ``BoundaryCache`` lookup hit and miss
+"""
+
+from __future__ import annotations
+
+import importlib
+import sys
+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,
+)
+from cleveragents.infrastructure.sandbox.boundary import (  # noqa: E402
+    BoundaryCache,
+    compute_sandbox_domains,
+    is_sandbox_boundary,
+    sandbox_boundary,
+)
+
+# ---------------------------------------------------------------------------
+# Test data builders
+# ---------------------------------------------------------------------------
+
+_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 _resource(
+    name: str,
+    sandboxable: bool = False,
+    strategy: str = "none",
+    parents: list[str] | None = None,
+) -> Resource:
+    return Resource(
+        resource_id=_ulid(name),
+        name=name,
+        resource_type_name="git-checkout" if sandboxable else "fs-file",
+        classification=PhysVirt.PHYSICAL,
+        sandbox_strategy=SandboxStrategy(strategy),
+        capabilities=ResourceCapabilities(
+            readable=True,
+            writable=True,
+            sandboxable=sandboxable,
+            checkpointable=False,
+        ),
+        parents=parents or [],
+    )
+
+
+# ---------------------------------------------------------------------------
+# Pre-built test fixtures
+# ---------------------------------------------------------------------------
+
+
+def _build_shallow_dag() -> tuple[Resource, dict[str, Resource]]:
+    """Build a shallow DAG: boundary -> 10 files."""
+    boundary = _resource("bench-co", sandboxable=True, strategy="git_worktree")
+    registry = {boundary.resource_id: boundary}
+    leaf = boundary
+    for i in range(10):
+        child = _resource(f"bench-f{i}", parents=[boundary.resource_id])
+        registry[child.resource_id] = child
+        leaf = child
+    return leaf, registry
+
+
+def _build_deep_dag(depth: int = 50) -> tuple[Resource, dict[str, Resource]]:
+    """Build a deep linear DAG: boundary -> dir1 -> dir2 -> ... -> file."""
+    boundary = _resource("bench-deep-co", sandboxable=True, strategy="git_worktree")
+    registry = {boundary.resource_id: boundary}
+    current = boundary
+    for i in range(depth):
+        child = _resource(f"bench-d{i}", parents=[current.resource_id])
+        registry[child.resource_id] = child
+        current = child
+    return current, registry
+
+
+_SHALLOW_LEAF, _SHALLOW_REGISTRY = _build_shallow_dag()
+_DEEP_LEAF, _DEEP_REGISTRY = _build_deep_dag()
+
+
+# ---------------------------------------------------------------------------
+# is_sandbox_boundary benchmarks
+# ---------------------------------------------------------------------------
+
+
+class TimeSandboxBoundaryCheck:
+    """Benchmark is_sandbox_boundary() predicate."""
+
+    def setup(self) -> None:
+        self.boundary = _resource("bm-co", sandboxable=True, strategy="git_worktree")
+        self.non_boundary = _resource("bm-f", sandboxable=False)
+
+    def time_is_boundary_true(self) -> None:
+        is_sandbox_boundary(self.boundary)
+
+    def time_is_boundary_false(self) -> None:
+        is_sandbox_boundary(self.non_boundary)
+
+
+# ---------------------------------------------------------------------------
+# sandbox_boundary() walk benchmarks
+# ---------------------------------------------------------------------------
+
+
+class TimeSandboxBoundaryWalk:
+    """Benchmark sandbox_boundary() for various DAG shapes."""
+
+    def time_self_boundary(self) -> None:
+        """Walk when resource is itself the boundary (O(1))."""
+        boundary = _SHALLOW_REGISTRY[_SHALLOW_LEAF.parents[0]]
+        sandbox_boundary(boundary, _SHALLOW_REGISTRY)
+
+    def time_shallow_walk(self) -> None:
+        """Walk one hop from leaf to boundary."""
+        sandbox_boundary(_SHALLOW_LEAF, _SHALLOW_REGISTRY)
+
+    def time_deep_walk(self) -> None:
+        """Walk 50 hops from leaf to boundary."""
+        sandbox_boundary(_DEEP_LEAF, _DEEP_REGISTRY)
+
+
+# ---------------------------------------------------------------------------
+# compute_sandbox_domains benchmarks
+# ---------------------------------------------------------------------------
+
+
+class TimeSandboxDomains:
+    """Benchmark compute_sandbox_domains() grouping."""
+
+    def setup(self) -> None:
+        self.boundary = _resource("dom-co", sandboxable=True, strategy="git_worktree")
+        self.resources = []
+        self.registry = {self.boundary.resource_id: self.boundary}
+        for i in range(100):
+            r = _resource(f"dom-f{i}", parents=[self.boundary.resource_id])
+            self.resources.append(r)
+            self.registry[r.resource_id] = r
+
+    def time_domains_100_resources(self) -> None:
+        compute_sandbox_domains(self.resources, self.registry)
+
+
+# ---------------------------------------------------------------------------
+# BoundaryCache benchmarks
+# ---------------------------------------------------------------------------
+
+
+class TimeBoundaryCache:
+    """Benchmark BoundaryCache hit and miss paths."""
+
+    def setup(self) -> None:
+        self.cache = BoundaryCache()
+        self.leaf = _SHALLOW_LEAF
+        self.registry = _SHALLOW_REGISTRY
+
+    def time_cache_miss(self) -> None:
+        """First lookup (cache miss)."""
+        self.cache.clear()
+        self.cache.get_boundary(self.leaf, self.registry)
+
+    def time_cache_hit(self) -> None:
+        """Second lookup (cache hit)."""
+        self.cache.get_boundary(self.leaf, self.registry)

features/sandbox_boundary_algebra.feature

@@ -0,0 +1,185 @@
+Feature: Sandbox Boundary Algebra
+  As a plan execution engine
+  I want to compute sandbox boundaries by walking the resource DAG
+  So that resources sharing a boundary share one sandbox instance
+
+  Background:
+    Given a resource registry for boundary algebra tests
+
+  # -- sandbox_boundary(r) basic resolution --
+
+  Scenario: A resource that is itself a sandbox boundary returns itself
+    Given a sandboxable resource "checkout-01" with strategy "git_worktree"
+    When I compute the sandbox boundary for "checkout-01"
+    Then the boundary should be "checkout-01"
+
+  Scenario: A child resource resolves to its sandboxable parent
+    Given a sandboxable resource "checkout-01" with strategy "git_worktree"
+    And a non-sandboxable resource "file-01" with parent "checkout-01"
+    When I compute the sandbox boundary for "file-01"
+    Then the boundary should be "checkout-01"
+
+  Scenario: A deeply nested resource resolves to a distant boundary
+    Given a sandboxable resource "checkout-01" with strategy "git_worktree"
+    And a non-sandboxable resource "dir-01" with parent "checkout-01"
+    And a non-sandboxable resource "subdir-01" with parent "dir-01"
+    And a non-sandboxable resource "file-deep" with parent "subdir-01"
+    When I compute the sandbox boundary for "file-deep"
+    Then the boundary should be "checkout-01"
+
+  Scenario: A resource with no sandboxable ancestor returns None
+    Given a non-sandboxable resource "orphan-01" with no parents
+    When I compute the sandbox boundary for "orphan-01"
+    Then the boundary should be None
+
+  Scenario: A resource whose parent is not in the registry returns None
+    Given a non-sandboxable resource "dangling-01" with missing parent "ghost-99"
+    When I compute the sandbox boundary for "dangling-01"
+    Then the boundary should be None
+
+  # -- is_sandbox_boundary checks --
+
+  Scenario: A resource with sandboxable capability and non-none strategy is a boundary
+    Given a sandboxable resource "boundary-res" with strategy "copy_on_write"
+    When I check if "boundary-res" is a sandbox boundary
+    Then it should be a sandbox boundary
+
+  Scenario: A resource with sandboxable capability but none strategy is not a boundary
+    Given a resource "non-boundary" with sandboxable true and strategy "none"
+    When I check if "non-boundary" is a sandbox boundary
+    Then it should not be a sandbox boundary
+
+  Scenario: A resource with no sandboxable capability is not a boundary
+    Given a non-sandboxable resource "no-cap" with no parents
+    When I check if "no-cap" is a sandbox boundary
+    Then it should not be a sandbox boundary
+
+  Scenario: Checking is_sandbox_boundary with None raises ValueError
+    When I check is_sandbox_boundary with None resource
+    Then the error should be a ValueError with text "resource cannot be None"
+
+  # -- domain grouping --
+
+  Scenario: Resources with the same boundary are grouped into one domain
+    Given a sandboxable resource "checkout-01" with strategy "git_worktree"
+    And a non-sandboxable resource "file-a" with parent "checkout-01"
+    And a non-sandboxable resource "file-b" with parent "checkout-01"
+    When I compute sandbox domains for "file-a" and "file-b"
+    Then both resources should be in the domain of "checkout-01"
+
+  Scenario: Multiple boundaries produce separate domains
+    Given a sandboxable resource "checkout-01" with strategy "git_worktree"
+    And a sandboxable resource "dir-root" with strategy "copy_on_write"
+    And a non-sandboxable resource "file-git" with parent "checkout-01"
+    And a non-sandboxable resource "file-fs" with parent "dir-root"
+    When I compute sandbox domains for "file-git" and "file-fs"
+    Then "file-git" should be in the domain of "checkout-01"
+    And "file-fs" should be in the domain of "dir-root"
+
+  Scenario: Unsandboxable resources are grouped under the special key
+    Given a non-sandboxable resource "orphan-a" with no parents
+    And a non-sandboxable resource "orphan-b" with no parents
+    When I compute sandbox domains for "orphan-a" and "orphan-b"
+    Then both resources should be in the unsandboxable domain
+
+  # -- boundary caching --
+
+  Scenario: Boundary results are cached across lookups
+    Given a sandboxable resource "checkout-01" with strategy "git_worktree"
+    And a non-sandboxable resource "file-cached" with parent "checkout-01"
+    And a fresh boundary cache
+    When I look up the boundary for "file-cached" via the cache
+    And I look up the boundary for "file-cached" via the cache again
+    Then the cache should contain 1 entry
+    And the cached boundary for "file-cached" should be "checkout-01"
+
+  Scenario: Cache can be cleared
+    Given a sandboxable resource "checkout-01" with strategy "git_worktree"
+    And a non-sandboxable resource "file-c" with parent "checkout-01"
+    And a fresh boundary cache
+    When I look up the boundary for "file-c" via the cache
+    And I clear the boundary cache
+    Then the cache should contain 0 entries
+
+  Scenario: Cache returns None for unsandboxable resources
+    Given a non-sandboxable resource "orphan-cached" with no parents
+    And a fresh boundary cache
+    When I look up the boundary for "orphan-cached" via the cache
+    Then the cached boundary result should be None
+    And the cache should contain 1 entry
+
+  # -- SandboxManager boundary-aware keying --
+
+  Scenario: Manager resolves sandbox key via boundary algebra
+    Given a located sandboxable resource "checkout-01" strategy "git_worktree" location "/tmp/repo"
+    And a non-sandboxable resource "file-01" with parent "checkout-01"
+    And a sandbox factory instance
+    And a sandbox manager with the factory
+    When I resolve the sandbox key for plan "plan-001" resource "file-01"
+    Then the sandbox key should be "checkout-01"
+
+  Scenario: Manager creates sandbox keyed by boundary ID
+    Given a located sandboxable resource "checkout-01" strategy "git_worktree" location "/tmp/repo"
+    And a non-sandboxable resource "file-01" with parent "checkout-01"
+    And a non-sandboxable resource "file-02" with parent "checkout-01"
+    And a mock sandbox factory
+    And a sandbox manager with the mock factory
+    When I get or create a boundary sandbox for plan "plan-001" resource "file-01"
+    And I get or create a boundary sandbox for plan "plan-001" resource "file-02"
+    Then both resources should share the same sandbox instance
+
+  Scenario: Manager raises NoSandboxBoundaryError for orphan resource
+    Given a non-sandboxable resource "orphan-mgr" with no parents
+    And a sandbox factory instance
+    And a sandbox manager with the factory
+    When I try to resolve the sandbox key for plan "plan-001" resource "orphan-mgr"
+    Then a NoSandboxBoundaryError should be raised
+
+  Scenario: Manager boundary cache is cleared on demand
+    Given a located sandboxable resource "checkout-01" strategy "git_worktree" location "/tmp/repo"
+    And a non-sandboxable resource "file-01" with parent "checkout-01"
+    And a sandbox factory instance
+    And a sandbox manager with the factory
+    When I resolve the sandbox key for plan "plan-001" resource "file-01"
+    Then the manager boundary cache should have 1 entry
+    When I clear the manager boundary cache
+    Then the manager boundary cache should have 0 entries
+
+  # -- argument validation --
+
+  Scenario: sandbox_boundary with None resource raises ValueError
+    When I call sandbox_boundary with None resource
+    Then the error should be a ValueError with text "resource cannot be None"
+
+  Scenario: sandbox_boundary with None registry raises ValueError
+    Given a non-sandboxable resource "dummy" with no parents
+    When I call sandbox_boundary with None registry for "dummy"
+    Then the error should be a ValueError with text "resource_registry cannot be None"
+
+  Scenario: compute_sandbox_domains with None resources raises ValueError
+    When I call compute_sandbox_domains with None resources
+    Then the error should be a ValueError with text "resources cannot be None"
+
+  Scenario: compute_sandbox_domains with None registry raises ValueError
+    Given a non-sandboxable resource "dummy" with no parents
+    When I call compute_sandbox_domains with None registry
+    Then the error should be a ValueError with text "resource_registry cannot be None"
+
+  Scenario: BoundaryCache with None resource raises ValueError
+    Given a fresh boundary cache
+    When I call boundary cache get_boundary with None resource
+    Then the error should be a ValueError with text "resource cannot be None"
+
+  Scenario: BoundaryCache with None registry raises ValueError
+    Given a non-sandboxable resource "dummy" with no parents
+    And a fresh boundary cache
+    When I call boundary cache get_boundary with None registry for "dummy"
+    Then the error should be a ValueError with text "resource_registry cannot be None"
+
+  Scenario: Boundary walk skips already-visited parent IDs
+    Given a sandboxable resource "checkout-01" with strategy "git_worktree"
+    And a non-sandboxable resource "dir-a" with parent "checkout-01"
+    And a non-sandboxable resource "dir-b" with parent "checkout-01"
+    And a non-sandboxable resource "file-shared" with parents "dir-a" and "dir-b"
+    When I compute the sandbox boundary for "file-shared"
+    Then the boundary should be "checkout-01"

features/steps/sandbox_boundary_algebra_steps.py

@@ -0,0 +1,534 @@
+"""Step definitions for sandbox boundary algebra BDD tests.
+
+Covers:
+- ``sandbox_boundary(r)`` function
+- ``is_sandbox_boundary(r)`` predicate
+- ``compute_sandbox_domains()`` grouping
+- ``BoundaryCache`` caching behaviour
+- ``SandboxManager`` boundary-aware keying
+"""
+
+from __future__ import annotations
+
+from datetime import datetime
+from typing import Any
+from unittest.mock import MagicMock
+
+from behave import given, then, when
+
+from cleveragents.domain.models.core.resource import (
+    PhysVirt,
+    Resource,
+    ResourceCapabilities,
+    SandboxStrategy,
+)
+from cleveragents.infrastructure.sandbox.boundary import (
+    BoundaryCache,
+    NoSandboxBoundaryError,
+    compute_sandbox_domains,
+    is_sandbox_boundary,
+    sandbox_boundary,
+)
+from cleveragents.infrastructure.sandbox.factory import SandboxFactory
+from cleveragents.infrastructure.sandbox.manager import SandboxManager
+from cleveragents.infrastructure.sandbox.protocol import (
+    CommitResult,
+    SandboxContext,
+    SandboxStatus,
+)
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+# Counter for generating unique ULIDs in tests.
+_ULID_COUNTER: int = 0
+
+
+def _make_ulid(name: str) -> str:
+    """Generate a deterministic ULID-like ID from a name.
+
+    Uses Crockford base-32 charset (0-9, A-H, J-K, M-N, P-T, V-Z)
+    to satisfy the ULID pattern ``^[0-9A-HJKMNP-TV-Z]{26}$``.
+    """
+    global _ULID_COUNTER
+    _ULID_COUNTER += 1
+    # Pad the counter + hash of name into 26 chars of Crockford base32.
+    base = abs(hash(name)) % (10**20)
+    raw = f"{_ULID_COUNTER:06d}{base:020d}"
+    # Map digits to Crockford charset.
+    charset = "0123456789ABCDEFGHJKMNPQRSTVWXYZ"
+    result = ""
+    for ch in raw:
+        result += charset[int(ch)]
+    # Pad/truncate to 26.
+    return (result + "0" * 26)[:26]
+
+
+def _make_resource(
+    name: str,
+    sandboxable: bool = False,
+    strategy: str = "none",
+    parents: list[str] | None = None,
+    location: str | None = None,
+) -> Resource:
+    """Create a test resource with the given properties."""
+    rid = _make_ulid(name)
+    return Resource(
+        resource_id=rid,
+        name=name,
+        resource_type_name="fs-file" if not sandboxable else "git-checkout",
+        classification=PhysVirt.PHYSICAL,
+        sandbox_strategy=SandboxStrategy(strategy),
+        capabilities=ResourceCapabilities(
+            readable=True,
+            writable=True,
+            sandboxable=sandboxable,
+            checkpointable=False,
+        ),
+        parents=parents or [],
+        location=location,
+    )
+
+
+# ---------------------------------------------------------------------------
+# Background
+# ---------------------------------------------------------------------------
+
+
+@given("a resource registry for boundary algebra tests")
+def step_given_resource_registry(context: Any) -> None:
+    """Initialise a fresh resource registry (dict) for this scenario."""
+    context.resource_registry = {}
+    context.resources_by_name = {}
+    context.boundary_result = None
+    context.boundary_error = None
+    context.is_boundary_result = None
+    context.domain_result = None
+    context.boundary_cache = None
+    context.sandbox_key_result = None
+    context.boundary_sandbox_results = []
+    context.error = None
+
+
+# ---------------------------------------------------------------------------
+# Given steps - resource construction
+# ---------------------------------------------------------------------------
+
+
+@given('a sandboxable resource "{name}" with strategy "{strategy}"')
+def step_given_sandboxable_resource(context: Any, name: str, strategy: str) -> None:
+    res = _make_resource(name, sandboxable=True, strategy=strategy)
+    context.resource_registry[res.resource_id] = res
+    context.resources_by_name[name] = res
+
+
+@given('a located sandboxable resource "{name}" strategy "{strategy}" location "{loc}"')
+def step_given_sandboxable_resource_with_location(
+    context: Any, name: str, strategy: str, loc: str
+) -> None:
+    res = _make_resource(name, sandboxable=True, strategy=strategy, location=loc)
+    context.resource_registry[res.resource_id] = res
+    context.resources_by_name[name] = res
+
+
+@given('a non-sandboxable resource "{name}" with parent "{parent_name}"')
+def step_given_nonsandboxable_with_parent(
+    context: Any, name: str, parent_name: str
+) -> None:
+    parent = context.resources_by_name[parent_name]
+    res = _make_resource(name, sandboxable=False, parents=[parent.resource_id])
+    context.resource_registry[res.resource_id] = res
+    context.resources_by_name[name] = res
+
+
+@given('a non-sandboxable resource "{name}" with no parents')
+def step_given_nonsandboxable_orphan(context: Any, name: str) -> None:
+    res = _make_resource(name, sandboxable=False)
+    context.resource_registry[res.resource_id] = res
+    context.resources_by_name[name] = res
+
+
+@given('a non-sandboxable resource "{name}" with missing parent "{missing_parent}"')
+def step_given_nonsandboxable_dangling(
+    context: Any, name: str, missing_parent: str
+) -> None:
+    # Parent ID is not registered in the registry.
+    fake_parent_id = _make_ulid(missing_parent)
+    res = _make_resource(name, sandboxable=False, parents=[fake_parent_id])
+    context.resource_registry[res.resource_id] = res
+    context.resources_by_name[name] = res
+
+
+@given('a resource "{name}" with sandboxable true and strategy "{strategy}"')
+def step_given_sandboxable_with_strategy(
+    context: Any, name: str, strategy: str
+) -> None:
+    res = _make_resource(name, sandboxable=True, strategy=strategy)
+    context.resource_registry[res.resource_id] = res
+    context.resources_by_name[name] = res
+
+
+@given("a fresh boundary cache")
+def step_given_boundary_cache(context: Any) -> None:
+    context.boundary_cache = BoundaryCache()
+
+
+def _make_mock_sandbox(
+    resource_id: str = "res-mock",
+    plan_id: str = "plan-mock",
+    original_path: str = "/tmp/mock",
+) -> MagicMock:
+    """Create a mock sandbox satisfying the Sandbox protocol."""
+    sb = MagicMock()
+    sandbox_id = f"sb-{resource_id}"
+    sb.sandbox_id = sandbox_id
+    sb.status = SandboxStatus.CREATED
+    sb.context = SandboxContext(
+        sandbox_id=sandbox_id,
+        sandbox_path=original_path,
+        original_path=original_path,
+        resource_id=resource_id,
+        plan_id=plan_id,
+        created_at=datetime.now(),
+    )
+    sb.create.return_value = sb.context
+    sb.commit.return_value = CommitResult(
+        sandbox_id=sandbox_id,
+        success=True,
+        timestamp=datetime.now(),
+    )
+    sb.rollback.return_value = None
+    sb.cleanup.return_value = None
+    return sb
+
+
+@given("a mock sandbox factory")
+def step_given_mock_factory(context: Any) -> None:
+    """Create a mock factory that returns mock sandboxes.
+
+    Mocks are cached by resource_id so the same mock is returned for
+    repeated calls with the same resource_id (mimics real dedup).
+    """
+    factory = MagicMock(spec=SandboxFactory)
+    mock_cache: dict[str, MagicMock] = {}
+
+    def _create_sandbox(
+        resource_id: str,
+        original_path: str,
+        sandbox_strategy: str,
+    ) -> MagicMock:
+        if resource_id not in mock_cache:
+            mock_cache[resource_id] = _make_mock_sandbox(
+                resource_id=resource_id,
+                original_path=original_path,
+            )
+        return mock_cache[resource_id]
+
+    factory.create_sandbox.side_effect = _create_sandbox
+    context.factory = factory
+
+
+@given("a sandbox manager with the mock factory")
+def step_given_manager_mock_factory(context: Any) -> None:
+    context.manager = SandboxManager(factory=context.factory, cleanup_on_exit=False)
+    context.error = None
+
+
+# ---------------------------------------------------------------------------
+# When steps
+# ---------------------------------------------------------------------------
+
+
+@when('I compute the sandbox boundary for "{name}"')
+def step_when_compute_boundary(context: Any, name: str) -> None:
+    res = context.resources_by_name[name]
+    context.boundary_result = sandbox_boundary(res, context.resource_registry)
+
+
+@when('I check if "{name}" is a sandbox boundary')
+def step_when_check_is_boundary(context: Any, name: str) -> None:
+    res = context.resources_by_name[name]
+    context.is_boundary_result = is_sandbox_boundary(res)
+
+
+@when("I check is_sandbox_boundary with None resource")
+def step_when_check_is_boundary_none(context: Any) -> None:
+    try:
+        is_sandbox_boundary(None)  # type: ignore[arg-type]
+    except ValueError as exc:
+        context.error = exc
+
+
+@when('I compute sandbox domains for "{name_a}" and "{name_b}"')
+def step_when_compute_domains(context: Any, name_a: str, name_b: str) -> None:
+    res_a = context.resources_by_name[name_a]
+    res_b = context.resources_by_name[name_b]
+    context.domain_result = compute_sandbox_domains(
+        [res_a, res_b], context.resource_registry
+    )
+
+
+@when('I look up the boundary for "{name}" via the cache')
+def step_when_cache_lookup(context: Any, name: str) -> None:
+    res = context.resources_by_name[name]
+    context.boundary_result = context.boundary_cache.get_boundary(
+        res, context.resource_registry
+    )
+
+
+@when('I look up the boundary for "{name}" via the cache again')
+def step_when_cache_lookup_again(context: Any, name: str) -> None:
+    # Same as above; exercises cache hit path.
+    res = context.resources_by_name[name]
+    context.boundary_result = context.boundary_cache.get_boundary(
+        res, context.resource_registry
+    )
+
+
+@when("I clear the boundary cache")
+def step_when_clear_cache(context: Any) -> None:
+    if context.boundary_cache is not None:
+        context.boundary_cache.clear()
+    else:
+        context.manager.clear_boundary_cache()
+
+
+@when('I resolve the sandbox key for plan "{plan_id}" resource "{name}"')
+def step_when_resolve_sandbox_key(context: Any, plan_id: str, name: str) -> None:
+    res = context.resources_by_name[name]
+    context.sandbox_key_result = context.manager.resolve_sandbox_key(
+        res, context.resource_registry
+    )
+
+
+@when('I try to resolve the sandbox key for plan "{plan_id}" resource "{name}"')
+def step_when_try_resolve_sandbox_key(context: Any, plan_id: str, name: str) -> None:
+    res = context.resources_by_name[name]
+    try:
+        context.manager.resolve_sandbox_key(res, context.resource_registry)
+    except NoSandboxBoundaryError as exc:
+        context.error = exc
+
+
+@when('I get or create a boundary sandbox for plan "{plan_id}" resource "{name}"')
+def step_when_get_or_create_boundary_sandbox(
+    context: Any, plan_id: str, name: str
+) -> None:
+    res = context.resources_by_name[name]
+    sb = context.manager.get_or_create_sandbox_for_resource(
+        plan_id=plan_id,
+        resource=res,
+        resource_registry=context.resource_registry,
+    )
+    context.boundary_sandbox_results.append(sb)
+
+
+@when("I clear the manager boundary cache")
+def step_when_clear_manager_cache(context: Any) -> None:
+    context.manager.clear_boundary_cache()
+
+
+@given('a non-sandboxable resource "{name}" with parents "{parent_a}" and "{parent_b}"')
+def step_given_nonsandboxable_multi_parent(
+    context: Any, name: str, parent_a: str, parent_b: str
+) -> None:
+    pa = context.resources_by_name[parent_a]
+    pb = context.resources_by_name[parent_b]
+    res = _make_resource(
+        name, sandboxable=False, parents=[pa.resource_id, pb.resource_id]
+    )
+    context.resource_registry[res.resource_id] = res
+    context.resources_by_name[name] = res
+
+
+@when("I call boundary cache get_boundary with None resource")
+def step_when_cache_none_resource(context: Any) -> None:
+    try:
+        context.boundary_cache.get_boundary(None, context.resource_registry)  # type: ignore[arg-type]
+    except ValueError as exc:
+        context.error = exc
+
+
+@when('I call boundary cache get_boundary with None registry for "{name}"')
+def step_when_cache_none_registry(context: Any, name: str) -> None:
+    res = context.resources_by_name[name]
+    try:
+        context.boundary_cache.get_boundary(res, None)  # type: ignore[arg-type]
+    except ValueError as exc:
+        context.error = exc
+
+
+@when("I call sandbox_boundary with None resource")
+def step_when_sandbox_boundary_none_resource(context: Any) -> None:
+    try:
+        sandbox_boundary(None, context.resource_registry)  # type: ignore[arg-type]
+    except ValueError as exc:
+        context.error = exc
+
+
+@when('I call sandbox_boundary with None registry for "{name}"')
+def step_when_sandbox_boundary_none_registry(context: Any, name: str) -> None:
+    res = context.resources_by_name[name]
+    try:
+        sandbox_boundary(res, None)  # type: ignore[arg-type]
+    except ValueError as exc:
+        context.error = exc
+
+
+@when("I call compute_sandbox_domains with None resources")
+def step_when_domains_none_resources(context: Any) -> None:
+    try:
+        compute_sandbox_domains(None, context.resource_registry)  # type: ignore[arg-type]
+    except ValueError as exc:
+        context.error = exc
+
+
+@when("I call compute_sandbox_domains with None registry")
+def step_when_domains_none_registry(context: Any) -> None:
+    res = context.resources_by_name.get("dummy")
+    try:
+        compute_sandbox_domains([res] if res else [], None)  # type: ignore[arg-type]
+    except ValueError as exc:
+        context.error = exc
+
+
+# ---------------------------------------------------------------------------
+# Then steps
+# ---------------------------------------------------------------------------
+
+
+@then('the boundary should be "{expected_name}"')
+def step_then_boundary_is(context: Any, expected_name: str) -> None:
+    expected = context.resources_by_name[expected_name]
+    assert context.boundary_result is not None, (
+        f"Expected boundary '{expected_name}' but got None"
+    )
+    assert context.boundary_result.resource_id == expected.resource_id, (
+        f"Expected boundary {expected.resource_id}, "
+        f"got {context.boundary_result.resource_id}"
+    )
+
+
+@then("the boundary should be None")
+def step_then_boundary_is_none(context: Any) -> None:
+    assert context.boundary_result is None, (
+        f"Expected None boundary, got {context.boundary_result}"
+    )
+
+
+@then("it should be a sandbox boundary")
+def step_then_is_boundary(context: Any) -> None:
+    assert context.is_boundary_result is True, (
+        "Expected resource to be a sandbox boundary"
+    )
+
+
+@then("it should not be a sandbox boundary")
+def step_then_is_not_boundary(context: Any) -> None:
+    assert context.is_boundary_result is False, (
+        "Expected resource NOT to be a sandbox boundary"
+    )
+
+
+@then('the error should be a ValueError with text "{msg}"')
+def step_then_value_error_containing(context: Any, msg: str) -> None:
+    assert context.error is not None, "Expected a ValueError but no error was raised"
+    assert isinstance(context.error, ValueError), (
+        f"Expected ValueError, got {type(context.error).__name__}: {context.error}"
+    )
+    assert msg in str(context.error), f"Expected '{msg}' in '{context.error}'"
+    assert msg in str(context.error), f"Expected '{msg}' in '{context.error}'"
+
+
+@then('both resources should be in the domain of "{boundary_name}"')
+def step_then_both_in_domain(context: Any, boundary_name: str) -> None:
+    boundary = context.resources_by_name[boundary_name]
+    domain = context.domain_result.get(boundary.resource_id, [])
+    assert len(domain) == 2, (
+        f"Expected 2 resources in domain of '{boundary_name}', got {len(domain)}"
+    )
+
+
+@then('"{name}" should be in the domain of "{boundary_name}"')
+def step_then_resource_in_domain(context: Any, name: str, boundary_name: str) -> None:
+    boundary = context.resources_by_name[boundary_name]
+    res = context.resources_by_name[name]
+    domain = context.domain_result.get(boundary.resource_id, [])
+    ids = [r.resource_id for r in domain]
+    assert res.resource_id in ids, (
+        f"Resource '{name}' not found in domain of '{boundary_name}'"
+    )
+
+
+@then("both resources should be in the unsandboxable domain")
+def step_then_both_unsandboxable(context: Any) -> None:
+    domain = context.domain_result.get("__unsandboxable__", [])
+    assert len(domain) == 2, f"Expected 2 unsandboxable resources, got {len(domain)}"
+
+
+@then("the cache should contain {count:d} entry")
+def step_then_cache_size_singular(context: Any, count: int) -> None:
+    cache = context.boundary_cache or context.manager._boundary_cache
+    assert cache.size == count, f"Expected cache size {count}, got {cache.size}"
+
+
+@then("the cache should contain {count:d} entries")
+def step_then_cache_size(context: Any, count: int) -> None:
+    cache = context.boundary_cache or context.manager._boundary_cache
+    assert cache.size == count, f"Expected cache size {count}, got {cache.size}"
+
+
+@then('the cached boundary for "{name}" should be "{boundary_name}"')
+def step_then_cached_boundary(context: Any, name: str, boundary_name: str) -> None:
+    res = context.resources_by_name[name]
+    boundary = context.resources_by_name[boundary_name]
+    cache = context.boundary_cache
+    result = cache.get_boundary(res, context.resource_registry)
+    assert result is not None, f"Cached boundary for '{name}' is None"
+    assert result.resource_id == boundary.resource_id
+
+
+@then("the cached boundary result should be None")
+def step_then_cached_boundary_none(context: Any) -> None:
+    assert context.boundary_result is None
+
+
+@then('the sandbox key should be "{expected_name}"')
+def step_then_sandbox_key(context: Any, expected_name: str) -> None:
+    expected = context.resources_by_name[expected_name]
+    assert context.sandbox_key_result == expected.resource_id, (
+        f"Expected key {expected.resource_id}, got {context.sandbox_key_result}"
+    )
+
+
+@then("both resources should share the same sandbox instance")
+def step_then_same_sandbox_instance(context: Any) -> None:
+    assert len(context.boundary_sandbox_results) == 2
+    assert context.boundary_sandbox_results[0] is context.boundary_sandbox_results[1], (
+        "Expected both resources to share the same sandbox instance"
+    )
+
+
+@then("a NoSandboxBoundaryError should be raised")
+def step_then_no_boundary_error(context: Any) -> None:
+    assert context.error is not None, (
+        "Expected NoSandboxBoundaryError but no error was raised"
+    )
+    assert isinstance(context.error, NoSandboxBoundaryError), (
+        f"Expected NoSandboxBoundaryError, got {type(context.error).__name__}"
+    )
+
+
+@then("the manager boundary cache should have {count:d} entry")
+def step_then_manager_cache_singular(context: Any, count: int) -> None:
+    assert context.manager.boundary_cache_size == count, (
+        f"Expected {count} entry, got {context.manager.boundary_cache_size}"
+    )
+
+
+@then("the manager boundary cache should have {count:d} entries")
+def step_then_manager_cache(context: Any, count: int) -> None:
+    assert context.manager.boundary_cache_size == count, (
+        f"Expected {count} entries, got {context.manager.boundary_cache_size}"
+    )

robot/helper_sandbox_boundary.py

@@ -0,0 +1,281 @@
+"""Helper utilities for sandbox boundary algebra Robot integration tests.
+
+Covers the integration between:
+- sandbox_boundary() function
+- compute_sandbox_domains() grouping
+- BoundaryCache caching
+- SandboxManager boundary-aware keying
+"""
+
+from __future__ import annotations
+
+import sys
+
+from cleveragents.domain.models.core.resource import (
+    PhysVirt,
+    Resource,
+    ResourceCapabilities,
+    SandboxStrategy,
+)
+from cleveragents.infrastructure.sandbox.boundary import (
+    BoundaryCache,
+    NoSandboxBoundaryError,
+    compute_sandbox_domains,
+    is_sandbox_boundary,
+    sandbox_boundary,
+)
+from cleveragents.infrastructure.sandbox.factory import SandboxFactory
+from cleveragents.infrastructure.sandbox.manager import SandboxManager
+
+_COUNTER = 0
+
+
+def _ulid(name: str) -> str:
+    """Generate a deterministic ULID for test resources."""
+    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 _resource(
+    name: str,
+    sandboxable: bool = False,
+    strategy: str = "none",
+    parents: list[str] | None = None,
+    location: str | None = None,
+) -> Resource:
+    return Resource(
+        resource_id=_ulid(name),
+        name=name,
+        resource_type_name="git-checkout" if sandboxable else "fs-file",
+        classification=PhysVirt.PHYSICAL,
+        sandbox_strategy=SandboxStrategy(strategy),
+        capabilities=ResourceCapabilities(
+            readable=True,
+            writable=True,
+            sandboxable=sandboxable,
+            checkpointable=False,
+        ),
+        parents=parents or [],
+        location=location,
+    )
+
+
+def _test_boundary_walk() -> None:
+    """Integration: boundary walks up to nearest sandboxable ancestor."""
+    checkout = _resource("checkout", sandboxable=True, strategy="git_worktree")
+    directory = _resource("dir", sandboxable=False, parents=[checkout.resource_id])
+    file_res = _resource("file", sandboxable=False, parents=[directory.resource_id])
+    registry = {
+        checkout.resource_id: checkout,
+        directory.resource_id: directory,
+        file_res.resource_id: file_res,
+    }
+
+    assert is_sandbox_boundary(checkout)
+    assert not is_sandbox_boundary(directory)
+    assert not is_sandbox_boundary(file_res)
+
+    boundary = sandbox_boundary(file_res, registry)
+    assert boundary is not None
+    assert boundary.resource_id == checkout.resource_id
+
+    # Self-boundary
+    self_boundary = sandbox_boundary(checkout, registry)
+    assert self_boundary is not None
+    assert self_boundary.resource_id == checkout.resource_id
+
+    print("boundary-walk-ok")
+
+
+def _test_domain_grouping() -> None:
+    """Integration: domain grouping partitions resources by boundary."""
+    co1 = _resource("co1", sandboxable=True, strategy="git_worktree")
+    co2 = _resource("co2", sandboxable=True, strategy="copy_on_write")
+    f1 = _resource("f1", parents=[co1.resource_id])
+    f2 = _resource("f2", parents=[co1.resource_id])
+    f3 = _resource("f3", parents=[co2.resource_id])
+    orphan = _resource("orphan")
+    registry = {r.resource_id: r for r in [co1, co2, f1, f2, f3, orphan]}
+
+    domains = compute_sandbox_domains([f1, f2, f3, orphan], registry)
+
+    assert co1.resource_id in domains
+    assert len(domains[co1.resource_id]) == 2
+    assert co2.resource_id in domains
+    assert len(domains[co2.resource_id]) == 1
+    assert "__unsandboxable__" in domains
+    assert len(domains["__unsandboxable__"]) == 1
+
+    print("domain-grouping-ok")
+
+
+def _test_boundary_cache() -> None:
+    """Integration: boundary cache stores and retrieves results."""
+    checkout = _resource("co-cache", sandboxable=True, strategy="git_worktree")
+    file_res = _resource("f-cache", parents=[checkout.resource_id])
+    registry = {
+        checkout.resource_id: checkout,
+        file_res.resource_id: file_res,
+    }
+    cache = BoundaryCache()
+
+    assert cache.size == 0
+    result = cache.get_boundary(file_res, registry)
+    assert result is not None
+    assert result.resource_id == checkout.resource_id
+    assert cache.size == 1
+
+    # Cache hit
+    result2 = cache.get_boundary(file_res, registry)
+    assert result2 is not None
+    assert result2.resource_id == checkout.resource_id
+    assert cache.size == 1
+
+    cache.clear()
+    assert cache.size == 0
+
+    print("boundary-cache-ok")
+
+
+def _test_manager_boundary_keying() -> None:
+    """Integration: SandboxManager resolves keys via boundary algebra.
+
+    Uses ``resolve_sandbox_key`` (no real sandbox creation) and then
+    ``get_or_create_sandbox`` with the resolved boundary to demonstrate
+    that two child resources share the same sandbox.
+    """
+    checkout = _resource(
+        "co-mgr",
+        sandboxable=True,
+        strategy="git_worktree",
+        location="/tmp",
+    )
+    f1 = _resource("f1-mgr", parents=[checkout.resource_id])
+    f2 = _resource("f2-mgr", parents=[checkout.resource_id])
+    registry = {
+        checkout.resource_id: checkout,
+        f1.resource_id: f1,
+        f2.resource_id: f2,
+    }
+
+    factory = SandboxFactory()
+    manager = SandboxManager(factory=factory, cleanup_on_exit=False)
+
+    key1 = manager.resolve_sandbox_key(f1, registry)
+    key2 = manager.resolve_sandbox_key(f2, registry)
+    assert key1 == checkout.resource_id
+    assert key2 == checkout.resource_id
+
+    # Both files resolve to the same key — verify they would share a sandbox
+    # by using get_or_create_sandbox with the resolved key and "none" strategy
+    # (to avoid needing a real git repo).
+    sb1 = manager.get_or_create_sandbox(
+        plan_id="plan-1",
+        resource_id=key1,
+        original_path="/tmp",
+        sandbox_strategy="none",
+    )
+    sb2 = manager.get_or_create_sandbox(
+        plan_id="plan-1",
+        resource_id=key2,
+        original_path="/tmp",
+        sandbox_strategy="none",
+    )
+    assert sb1 is sb2
+
+    # Orphan raises NoSandboxBoundaryError.
+    orphan = _resource("orphan-mgr")
+    registry[orphan.resource_id] = orphan
+    try:
+        manager.resolve_sandbox_key(orphan, registry)
+        print("FAIL: expected NoSandboxBoundaryError")
+        return
+    except NoSandboxBoundaryError:
+        pass
+
+    manager.cleanup_all("plan-1")
+    print("manager-boundary-keying-ok")
+
+
+def _test_multi_resource_plan() -> None:
+    """Integration: multi-resource plan with separate sandbox domains.
+
+    Uses resolve_sandbox_key to verify two different boundaries produce
+    different keys, then creates sandboxes using the keys directly.
+    """
+    co_git = _resource(
+        "co-git",
+        sandboxable=True,
+        strategy="git_worktree",
+        location="/tmp",
+    )
+    co_fs = _resource(
+        "co-fs",
+        sandboxable=True,
+        strategy="copy_on_write",
+        location="/tmp",
+    )
+    f_git = _resource("f-git", parents=[co_git.resource_id])
+    f_fs = _resource("f-fs", parents=[co_fs.resource_id])
+    registry = {r.resource_id: r for r in [co_git, co_fs, f_git, f_fs]}
+
+    factory = SandboxFactory()
+    manager = SandboxManager(factory=factory, cleanup_on_exit=False)
+
+    key_git = manager.resolve_sandbox_key(f_git, registry)
+    key_fs = manager.resolve_sandbox_key(f_fs, registry)
+    assert key_git != key_fs
+    assert key_git == co_git.resource_id
+    assert key_fs == co_fs.resource_id
+
+    # Create sandboxes via resolved keys using "none" strategy (test env).
+    sb_git = manager.get_or_create_sandbox(
+        plan_id="plan-2",
+        resource_id=key_git,
+        original_path="/tmp",
+        sandbox_strategy="none",
+    )
+    sb_fs = manager.get_or_create_sandbox(
+        plan_id="plan-2",
+        resource_id=key_fs,
+        original_path="/tmp",
+        sandbox_strategy="none",
+    )
+
+    # Different boundaries → different sandboxes.
+    assert sb_git is not sb_fs
+
+    sandboxes = manager.list_sandboxes("plan-2")
+    assert len(sandboxes) == 2
+
+    manager.cleanup_all("plan-2")
+    print("multi-resource-plan-ok")
+
+
+_COMMANDS = {
+    "boundary-walk": _test_boundary_walk,
+    "domain-grouping": _test_domain_grouping,
+    "boundary-cache": _test_boundary_cache,
+    "manager-boundary-keying": _test_manager_boundary_keying,
+    "multi-resource-plan": _test_multi_resource_plan,
+}
+
+if __name__ == "__main__":
+    if len(sys.argv) < 2:
+        print(f"Usage: {sys.argv[0]} <command>")
+        print(f"Commands: {', '.join(_COMMANDS)}")
+        sys.exit(1)
+
+    cmd = sys.argv[1]
+    fn = _COMMANDS.get(cmd)
+    if fn is None:
+        print(f"Unknown command: {cmd}")
+        sys.exit(1)
+    fn()

robot/sandbox_boundary_algebra.robot

@@ -0,0 +1,47 @@
+*** Settings ***
+Documentation     Integration tests for sandbox boundary algebra: sandbox_boundary(),
+...               compute_sandbox_domains(), BoundaryCache, and SandboxManager
+...               boundary-aware keying.
+...               Covers M6 sandbox boundary algebra (#548).
+Resource          ${CURDIR}/common.resource
+Suite Setup       Setup Test Environment
+Suite Teardown    Cleanup Test Environment
+
+*** Variables ***
+${HELPER_SCRIPT}    robot/helper_sandbox_boundary.py
+
+*** Test Cases ***
+Sandbox Boundary Walk To Nearest Sandboxable Ancestor
+    [Documentation]    Verify sandbox_boundary() walks containment edges to nearest sandboxable resource
+    [Tags]    sandbox    boundary    walk
+    ${result}=    Run Process    ${PYTHON}    ${HELPER_SCRIPT}    boundary-walk    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    boundary-walk-ok
+
+Domain Grouping Partitions Resources By Boundary
+    [Documentation]    Verify compute_sandbox_domains() groups resources by sandbox boundary
+    [Tags]    sandbox    boundary    domains
+    ${result}=    Run Process    ${PYTHON}    ${HELPER_SCRIPT}    domain-grouping    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    domain-grouping-ok
+
+Boundary Cache Stores And Retrieves Results
+    [Documentation]    Verify BoundaryCache caches boundary lookups for plan execution
+    [Tags]    sandbox    boundary    cache
+    ${result}=    Run Process    ${PYTHON}    ${HELPER_SCRIPT}    boundary-cache    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    boundary-cache-ok
+
+Manager Keys Sandboxes By Boundary ID
+    [Documentation]    Verify SandboxManager resolves keys via boundary algebra so child resources share sandbox
+    [Tags]    sandbox    boundary    manager
+    ${result}=    Run Process    ${PYTHON}    ${HELPER_SCRIPT}    manager-boundary-keying    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    manager-boundary-keying-ok
+
+Multi Resource Plan With Separate Sandbox Domains
+    [Documentation]    Verify multi-resource plan with different boundaries creates separate sandboxes
+    [Tags]    sandbox    boundary    multi
+    ${result}=    Run Process    ${PYTHON}    ${HELPER_SCRIPT}    multi-resource-plan    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    multi-resource-plan-ok

src/cleveragents/infrastructure/sandbox/__init__.py

@@ -5,8 +5,17 @@ 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).
 """
 
+from cleveragents.infrastructure.sandbox.boundary import (
+    BoundaryCache,
+    NoSandboxBoundaryError,
+    SandboxBoundaryError,
+    compute_sandbox_domains,
+    is_sandbox_boundary,
+    sandbox_boundary,
+)
 from cleveragents.infrastructure.sandbox.checkpoint import (
     Checkpointable,
     CheckpointManager,
@@ -33,6 +42,7 @@ from cleveragents.infrastructure.sandbox.protocol import (
 )
 
 __all__ = [
+    "BoundaryCache",
     "CheckpointManager",
     "Checkpointable",
     "CommitResult",
@@ -43,7 +53,9 @@ __all__ = [
     "MergeResult",
     "MergeStrategy",
     "NoSandbox",
+    "NoSandboxBoundaryError",
     "Sandbox",
+    "SandboxBoundaryError",
     "SandboxCheckpoint",
     "SandboxContext",
     "SandboxError",
@@ -51,4 +63,7 @@ __all__ = [
     "SandboxManager",
     "SandboxStatus",
     "SequentialMergeStrategy",
+    "compute_sandbox_domains",
+    "is_sandbox_boundary",
+    "sandbox_boundary",
 ]

src/cleveragents/infrastructure/sandbox/boundary.py

@@ -0,0 +1,228 @@
+"""Sandbox boundary algebra for CleverAgents.
+
+Implements the ``sandbox_boundary(r)`` function defined in the specification:
+walk up containment edges in the resource DAG to the nearest sandboxable
+ancestor.  All resources sharing the same boundary share one sandbox instance.
+
+**Definitions** (from spec §24659-24674):
+
+* **Sandbox boundary**: A resource ``b`` where
+  ``b.capabilities.sandboxable is True`` and
+  ``b.sandbox_strategy != "none"``.
+
+* **sandbox_boundary(r)**: The nearest ancestor of ``r`` (inclusive)
+  along ``contains`` edges that is a sandbox boundary.  If ``r`` itself
+  is a boundary, returns ``r``.  If no boundary exists in ``r``'s
+  ancestor chain, returns ``None`` (unsandboxable).
+
+* **Sandbox domain**: The set ``{r : sandbox_boundary(r) == b}`` —
+  resources whose nearest sandbox boundary is ``b``.
+
+Stage M6 - Sandbox Boundary Algebra (#548).
+"""
+
+from __future__ import annotations
+
+import logging
+import threading
+from collections.abc import Iterable, Mapping
+
+from cleveragents.domain.models.core.resource import Resource, SandboxStrategy
+
+logger = logging.getLogger(__name__)
+
+# Maximum DAG depth to prevent infinite loops from cyclic parent references.
+_MAX_DAG_DEPTH = 200
+
+
+class SandboxBoundaryError(Exception):
+    """Raised when sandbox boundary computation encounters an error."""
+
+
+class NoSandboxBoundaryError(SandboxBoundaryError):
+    """Raised when a resource has no sandboxable ancestor in its chain."""
+
+
+def is_sandbox_boundary(resource: Resource) -> bool:
+    """Check whether *resource* is a sandbox boundary.
+
+    A resource is a sandbox boundary when it declares itself sandboxable
+    **and** has a non-``none`` sandbox strategy.
+
+    Args:
+        resource: The resource to check.
+
+    Returns:
+        ``True`` if the resource is a sandbox boundary.
+
+    Raises:
+        ValueError: If *resource* is ``None``.
+    """
+    if resource is None:
+        raise ValueError("resource cannot be None")
+
+    return (
+        resource.capabilities.sandboxable
+        and resource.sandbox_strategy is not None
+        and resource.sandbox_strategy != SandboxStrategy.NONE
+    )
+
+
+def sandbox_boundary(
+    resource: Resource,
+    resource_registry: Mapping[str, Resource],
+) -> Resource | None:
+    """Walk up containment edges to the nearest sandboxable ancestor.
+
+    If *resource* itself is a sandbox boundary, it is returned immediately.
+    Otherwise we walk the ``parents`` list (containment edges) upward
+    until a boundary is found or the root is reached.
+
+    Args:
+        resource: Starting resource in the DAG.
+        resource_registry: Mapping of ``resource_id`` → ``Resource`` for
+            the entire DAG (or at least the ancestor chain).
+
+    Returns:
+        The nearest sandbox-boundary ancestor, or ``None`` if none exists.
+
+    Raises:
+        ValueError: If *resource* or *resource_registry* is ``None``.
+    """
+    if resource is None:
+        raise ValueError("resource cannot be None")
+    if resource_registry is None:
+        raise ValueError("resource_registry cannot be None")
+
+    # Self-check first (inclusive).
+    if is_sandbox_boundary(resource):
+        return resource
+
+    visited: set[str] = {resource.resource_id}
+    frontier: list[str] = list(resource.parents)
+    depth = 0
+
+    while frontier and depth < _MAX_DAG_DEPTH:
+        parent_id = frontier.pop(0)
+        if parent_id in visited:
+            continue
+        visited.add(parent_id)
+        depth += 1
+
+        parent = resource_registry.get(parent_id)
+        if parent is None:
+            logger.warning(
+                "Parent resource %s not found in registry while computing "
+                "sandbox boundary for %s",
+                parent_id,
+                resource.resource_id,
+            )
+            continue
+
+        if is_sandbox_boundary(parent):
+            return parent
+
+        # Continue walking upward.
+        for grandparent_id in parent.parents:
+            if grandparent_id not in visited:
+                frontier.append(grandparent_id)
+
+    return None
+
+
+def compute_sandbox_domains(
+    resources: Iterable[Resource],
+    resource_registry: Mapping[str, Resource],
+) -> dict[str, list[Resource]]:
+    """Group resources by their sandbox boundary.
+
+    Returns a mapping of ``boundary_resource_id`` → list of resources
+    in that domain.  Resources with no sandbox boundary are collected
+    under the key ``"__unsandboxable__"``.
+
+    Args:
+        resources: Iterable of resources to partition.
+        resource_registry: Full resource registry for ancestor lookup.
+
+    Returns:
+        Domain mapping (boundary_id → resource list).
+
+    Raises:
+        ValueError: If *resources* or *resource_registry* is ``None``.
+    """
+    if resources is None:
+        raise ValueError("resources cannot be None")
+    if resource_registry is None:
+        raise ValueError("resource_registry cannot be None")
+
+    domains: dict[str, list[Resource]] = {}
+    for res in resources:
+        boundary = sandbox_boundary(res, resource_registry)
+        key = boundary.resource_id if boundary is not None else "__unsandboxable__"
+        domains.setdefault(key, []).append(res)
+
+    return domains
+
+
+class BoundaryCache:
+    """Thread-safe cache for sandbox boundary lookups.
+
+    The resource DAG does not change during a plan execution, so boundary
+    results can be cached for the duration of a plan run.  Call
+    :meth:`clear` at the start of each execution or when the DAG changes.
+
+    Thread-safe: all mutable state is protected by ``_lock``.
+    """
+
+    def __init__(self) -> None:
+        self._cache: dict[str, str | None] = {}
+        self._lock: threading.RLock = threading.RLock()
+
+    def get_boundary(
+        self,
+        resource: Resource,
+        resource_registry: Mapping[str, Resource],
+    ) -> Resource | None:
+        """Return the sandbox boundary for *resource*, using the cache.
+
+        Args:
+            resource: The resource to look up.
+            resource_registry: Full resource registry for ancestor walk.
+
+        Returns:
+            The boundary resource, or ``None`` if unsandboxable.
+
+        Raises:
+            ValueError: If *resource* or *resource_registry* is ``None``.
+        """
+        if resource is None:
+            raise ValueError("resource cannot be None")
+        if resource_registry is None:
+            raise ValueError("resource_registry cannot be None")
+
+        with self._lock:
+            if resource.resource_id in self._cache:
+                cached_id = self._cache[resource.resource_id]
+                if cached_id is None:
+                    return None
+                return resource_registry.get(cached_id)
+
+        # Compute (outside lock to avoid holding it during DAG walk).
+        boundary = sandbox_boundary(resource, resource_registry)
+        boundary_id = boundary.resource_id if boundary is not None else None
+
+        with self._lock:
+            self._cache[resource.resource_id] = boundary_id
+
+        return boundary
+
+    def clear(self) -> None:
+        """Clear all cached boundary lookups."""
+        with self._lock:
+            self._cache.clear()
+
+    @property
+    def size(self) -> int:
+        """Return the number of cached entries."""
+        with self._lock:
+            return len(self._cache)

src/cleveragents/infrastructure/sandbox/manager.py

@@ -4,6 +4,10 @@ Manages the creation, tracking, and cleanup of sandbox instances across
 plan executions.  Thread-safe via an internal reentrant lock.
 
 Stage B3.7 of the implementation plan.
+Updated in M6 to support sandbox boundary algebra (#548):
+  - ``get_or_create_sandbox_for_resource`` keys by ``(plan_id, boundary_id)``
+  - ``resolve_sandbox_key`` computes the boundary key for a resource
+  - Boundary cache per plan execution
 """
 
 from __future__ import annotations
@@ -12,8 +16,14 @@ import atexit
 import contextlib
 import logging
 import threading
+from collections.abc import Mapping
 from datetime import datetime
 
+from cleveragents.domain.models.core.resource import Resource
+from cleveragents.infrastructure.sandbox.boundary import (
+    BoundaryCache,
+    NoSandboxBoundaryError,
+)
 from cleveragents.infrastructure.sandbox.factory import (
     SandboxFactory,
     SandboxStrategyStr,
@@ -74,6 +84,7 @@ class SandboxManager:
         self._active_sandboxes: dict[str, dict[str, Sandbox]] = {}
         self._lock: threading.RLock = threading.RLock()
         self._cleanup_on_exit: bool = cleanup_on_exit
+        self._boundary_cache: BoundaryCache = BoundaryCache()
 
         if cleanup_on_exit:
             atexit.register(self._cleanup_on_exit_handler)
@@ -339,6 +350,109 @@ class SandboxManager:
 
         return cleaned
 
+    # -- boundary-aware API (M6 #548) ----------------------------------------
+
+    def resolve_sandbox_key(
+        self,
+        resource: Resource,
+        resource_registry: Mapping[str, Resource],
+    ) -> str:
+        """Compute the sandbox key for a resource via boundary algebra.
+
+        The key is the ``resource_id`` of the resource's sandbox boundary.
+        If the resource itself is a boundary, its own ID is returned.
+
+        Args:
+            resource: The resource to resolve.
+            resource_registry: Full resource registry for DAG traversal.
+
+        Returns:
+            The ``resource_id`` of the sandbox boundary.
+
+        Raises:
+            ValueError: If *resource* or *resource_registry* is ``None``.
+            NoSandboxBoundaryError: If no sandbox boundary exists
+                in the resource's ancestor chain.
+        """
+        if resource is None:
+            raise ValueError("resource cannot be None")
+        if resource_registry is None:
+            raise ValueError("resource_registry cannot be None")
+
+        boundary = self._boundary_cache.get_boundary(resource, resource_registry)
+        if boundary is None:
+            raise NoSandboxBoundaryError(
+                f"Resource {resource.resource_id} has no sandboxable ancestor"
+            )
+        return boundary.resource_id
+
+    def get_or_create_sandbox_for_resource(
+        self,
+        plan_id: str,
+        resource: Resource,
+        resource_registry: Mapping[str, Resource],
+    ) -> Sandbox:
+        """Return a sandbox for the resource, keyed by sandbox boundary.
+
+        Uses the boundary algebra to determine which sandbox boundary
+        governs this resource, then delegates to
+        :meth:`get_or_create_sandbox` with the boundary's ID and
+        properties.  Multiple resources sharing the same boundary will
+        share the same sandbox instance.
+
+        Args:
+            plan_id: Identifier of the plan requesting the sandbox.
+            resource: The resource that needs sandboxing.
+            resource_registry: Full resource registry for DAG traversal.
+
+        Returns:
+            A :class:`Sandbox` instance for the resource's boundary.
+
+        Raises:
+            ValueError: If any required argument is empty/``None``.
+            NoSandboxBoundaryError: If no sandbox boundary exists.
+            SandboxError: If sandbox creation or initialisation fails.
+        """
+        if not plan_id:
+            raise ValueError("plan_id cannot be empty")
+        if resource is None:
+            raise ValueError("resource cannot be None")
+        if resource_registry is None:
+            raise ValueError("resource_registry cannot be None")
+
+        boundary_id = self.resolve_sandbox_key(resource, resource_registry)
+        boundary_resource = resource_registry[boundary_id]
+
+        # Determine strategy from the boundary resource.
+        strategy: SandboxStrategyStr = "none"
+        if boundary_resource.sandbox_strategy is not None:
+            strategy = boundary_resource.sandbox_strategy  # type: ignore[assignment]
+
+        # Determine location from the boundary resource.
+        original_path = boundary_resource.location or ""
+        if not original_path:
+            raise ValueError(f"Boundary resource {boundary_id} has no location")
+
+        return self.get_or_create_sandbox(
+            plan_id=plan_id,
+            resource_id=boundary_id,
+            original_path=original_path,
+            sandbox_strategy=strategy,
+        )
+
+    def clear_boundary_cache(self) -> None:
+        """Clear the boundary cache.
+
+        Should be called at the start of each plan execution or when
+        the resource DAG is modified.
+        """
+        self._boundary_cache.clear()
+
+    @property
+    def boundary_cache_size(self) -> int:
+        """Return the number of cached boundary lookups."""
+        return self._boundary_cache.size
+
     # -- internal ------------------------------------------------------------
 
     def _cleanup_on_exit_handler(self) -> None:

vulture_whitelist.py

@@ -708,3 +708,16 @@ LLM_TOTAL_COST_USD  # noqa: B018, F821
 LLM_AVG_LATENCY_MS  # noqa: B018, F821
 SUBPLAN_COUNT  # noqa: B018, F821
 _build_trace_service  # noqa: B018, F821
+
+# Sandbox boundary algebra — public API (M6, issue #548)
+SandboxBoundaryError  # noqa: B018, F821
+NoSandboxBoundaryError  # noqa: B018, F821
+BoundaryCache  # noqa: B018, F821
+is_sandbox_boundary  # noqa: B018, F821
+sandbox_boundary  # noqa: B018, F821
+compute_sandbox_domains  # noqa: B018, F821
+resolve_sandbox_key  # noqa: B018, F821
+get_or_create_sandbox_for_resource  # noqa: B018, F821
+clear_boundary_cache  # noqa: B018, F821
+boundary_cache_size  # noqa: B018, F821
+_MAX_DAG_DEPTH  # noqa: B018, F821