features/sandbox_manager_strategy_cast.feature

@@ -0,0 +1,18 @@
+Feature: Sandbox Manager strategy cast path
+  Exercises the cast("SandboxStrategyStr", boundary_resource.sandbox_strategy)
+  line in get_or_create_sandbox_for_resource, which is reached whenever the
+  boundary resource has a non-None sandbox_strategy.
+
+  Background:
+    Given smcast a sandbox factory instance
+    And smcast a sandbox manager with the factory
+
+  Scenario: get_or_create_sandbox_for_resource succeeds with SandboxStrategy.NONE
+    Given smcast a boundary resource with strategy "NONE" and location "/tmp/smcast-none"
+    When smcast I call get_or_create_sandbox_for_resource for plan "plan-smcast-1"
+    Then smcast a sandbox is returned without error
+
+  Scenario: get_or_create_sandbox_for_resource succeeds with SandboxStrategy.NONE on a second plan
+    Given smcast a boundary resource with strategy "NONE" and location "/tmp/smcast-none-2"
+    When smcast I call get_or_create_sandbox_for_resource for plan "plan-smcast-2"
+    Then smcast a sandbox is returned without error

features/steps/sandbox_manager_strategy_cast_steps.py

@@ -0,0 +1,165 @@
+"""Step definitions for sandbox manager strategy cast coverage.
+
+Exercises the cast("SandboxStrategyStr", boundary_resource.sandbox_strategy)
+line in get_or_create_sandbox_for_resource, which is reached whenever the
+boundary resource has a non-None sandbox_strategy.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+from unittest.mock import patch
+
+from behave import given, then, when
+
+from cleveragents.domain.models.core.resource import (
+    PhysVirt,
+    Resource,
+    ResourceCapabilities,
+    SandboxStrategy,
+)
+from cleveragents.infrastructure.sandbox.factory import SandboxFactory
+from cleveragents.infrastructure.sandbox.manager import SandboxManager
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+# Valid ULID strings for test resources
+_ULID_CHILD = "01ARZ3NDEKTSV4RRFFQ69G5FAV"
+_ULID_BOUNDARY = "01BRZ3NDEKTSV4RRFFQ69G5FAV"
+
+_STRATEGY_MAP: dict[str, SandboxStrategy] = {
+    "NONE": SandboxStrategy.NONE,
+    "GIT_WORKTREE": SandboxStrategy.GIT_WORKTREE,
+    "COPY_ON_WRITE": SandboxStrategy.COPY_ON_WRITE,
+    "TRANSACTION_ROLLBACK": SandboxStrategy.TRANSACTION_ROLLBACK,
+    "SNAPSHOT": SandboxStrategy.SNAPSHOT,
+    "OVERLAY": SandboxStrategy.OVERLAY,
+}
+
+
+def _make_resource(
+    resource_id: str = _ULID_CHILD,
+    location: str | None = "/tmp/repo",
+    sandbox_strategy: SandboxStrategy | None = SandboxStrategy.GIT_WORKTREE,
+    sandboxable: bool = True,
+    parents: list[str] | None = None,
+) -> Resource:
+    """Create a Resource domain object for testing."""
+    return Resource(
+        resource_id=resource_id,
+        resource_type_name="git-checkout",
+        classification=PhysVirt.PHYSICAL,
+        location=location,
+        sandbox_strategy=sandbox_strategy,
+        capabilities=ResourceCapabilities(sandboxable=sandboxable),
+        parents=parents or [],
+    )
+
+
+# ---------------------------------------------------------------------------
+# Background
+# ---------------------------------------------------------------------------
+
+
+@given("smcast a sandbox factory instance")
+def step_smcast_given_factory(context: Any) -> None:
+    context.smcast_factory = SandboxFactory()
+
+
+@given("smcast a sandbox manager with the factory")
+def step_smcast_given_manager(context: Any) -> None:
+    context.smcast_manager = SandboxManager(
+        factory=context.smcast_factory, cleanup_on_exit=False
+    )
+    context.smcast_error = None
+    context.smcast_sandbox = None
+    context.smcast_boundary_resource = None
+    context.smcast_child_resource = None
+    context.smcast_registry = None
+
+
+# ---------------------------------------------------------------------------
+# Given steps
+# ---------------------------------------------------------------------------
+
+
+@given(
+    'smcast a boundary resource with strategy "{strategy}" and location "{location}"'
+)
+def step_smcast_given_boundary_resource(
+    context: Any, strategy: str, location: str
+) -> None:
+    """Set up a boundary resource with the given strategy and location."""
+    sandbox_strategy = _STRATEGY_MAP[strategy]
+
+    boundary_resource = _make_resource(
+        resource_id=_ULID_BOUNDARY,
+        location=location,
+        sandbox_strategy=sandbox_strategy,
+        sandboxable=True,
+    )
+
+    child_resource = _make_resource(
+        resource_id=_ULID_CHILD,
+        location="/tmp/smcast-child",
+        sandbox_strategy=None,
+        sandboxable=False,
+        parents=[_ULID_BOUNDARY],
+    )
+
+    context.smcast_boundary_resource = boundary_resource
+    context.smcast_child_resource = child_resource
+    context.smcast_registry = {
+        _ULID_BOUNDARY: boundary_resource,
+        _ULID_CHILD: child_resource,
+    }
+
+
+# ---------------------------------------------------------------------------
+# When steps
+# ---------------------------------------------------------------------------
+
+
+@when('smcast I call get_or_create_sandbox_for_resource for plan "{plan_id}"')
+def step_smcast_when_call_for_resource(context: Any, plan_id: str) -> None:
+    """Call get_or_create_sandbox_for_resource with the prepared boundary resource.
+
+    The boundary cache is patched to return the boundary resource directly,
+    bypassing DAG traversal.
+    """
+    context.smcast_error = None
+    context.smcast_sandbox = None
+
+    patcher = patch.object(
+        context.smcast_manager._boundary_cache,
+        "get_boundary",
+        return_value=context.smcast_boundary_resource,
+    )
+    patcher.start()
+    context.add_cleanup(patcher.stop)
+
+    try:
+        context.smcast_sandbox = (
+            context.smcast_manager.get_or_create_sandbox_for_resource(
+                plan_id=plan_id,
+                resource=context.smcast_child_resource,
+                resource_registry=context.smcast_registry,
+            )
+        )
+    except Exception as exc:
+        context.smcast_error = exc
+
+
+# ---------------------------------------------------------------------------
+# Then steps
+# ---------------------------------------------------------------------------
+
+
+@then("smcast a sandbox is returned without error")
+def step_smcast_then_sandbox_returned(context: Any) -> None:
+    assert context.smcast_error is None, f"Unexpected error: {context.smcast_error}"
+    assert context.smcast_sandbox is not None, (
+        "Expected a sandbox to be returned, but got None"
+    )

src/cleveragents/infrastructure/sandbox/manager.py

@@ -18,6 +18,7 @@ import logging
 import threading
 from collections.abc import Mapping
 from datetime import datetime
+from typing import cast
 
 from cleveragents.domain.models.core.resource import Resource
 from cleveragents.infrastructure.sandbox.boundary import (
@@ -615,7 +616,7 @@ class SandboxManager:
         # 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]
+            strategy = cast("SandboxStrategyStr", boundary_resource.sandbox_strategy)
 
         # Determine location from the boundary resource.
         original_path = boundary_resource.location or ""