cleveragents-core/robot/helper_sandbox_boundary.py

"""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()