Conduit

The durable key/value rendezvous between host and sandbox — the only channel between the two halves. store_from_env selects a concrete backend from the environment; every backend implements the same three-operation Conduit interface. ObjectInfo describes a listed object.

resoluto.sandbox.Conduit

Bases: ABC

Durable key/value rendezvous (localfs, S3, GCS).

copy_prefix async

copy_prefix(src_prefix, dst_prefix, *, exclude_segments=())

Copy every object under src_prefix to dst_prefix, returning the count copied. exclude_segments: relative-path SEGMENTS to skip (e.g. ("steps",) drops every .../steps/... subtree — a resumed run must not inherit the prior run's step chunk indexes, or the telemetry reader mistakes the re-run pod for silent).

Source code in src/resoluto/sandbox/contracts.py

async def copy_prefix(
    self, src_prefix: str, dst_prefix: str, *, exclude_segments: tuple[str, ...] = ()
) -> int:
    """Copy every object under src_prefix to dst_prefix, returning the count copied.
    `exclude_segments`: relative-path SEGMENTS to skip (e.g. ("steps",) drops every
    `.../steps/...` subtree — a resumed run must not inherit the prior run's step
    chunk indexes, or the telemetry reader mistakes the re-run pod for silent)."""
    src, dst = src_prefix.rstrip("/"), dst_prefix.rstrip("/")
    n = 0
    for o in await self.list_prefix(src):
        rel = o.key[len(src) :].lstrip("/")
        if exclude_segments and set(rel.split("/")) & set(exclude_segments):
            continue
        await self.put(f"{dst}/{rel}", await self.get(o.key))
        n += 1
    return n

aclose async

aclose()

Release any cached client/session. Default no-op; override where there's something to release (a cached HTTP session, connection pool, etc). One name across every Conduit.

Source code in src/resoluto/sandbox/contracts.py

async def aclose(self) -> None:
    """Release any cached client/session. Default no-op; override where there's something to
    release (a cached HTTP session, connection pool, etc). One name across every Conduit."""

delete_prefix async

delete_prefix(prefix)

Delete every object under prefix, returning the count deleted. Retention/GC is the CALLER's policy; the conduit supplies only the mechanism.

Source code in src/resoluto/sandbox/contracts.py

async def delete_prefix(self, prefix: str) -> int:
    """Delete every object under `prefix`, returning the count deleted. Retention/GC is
    the CALLER's policy; the conduit supplies only the mechanism."""
    raise NotImplementedError(f"{type(self).__name__} does not implement delete_prefix")

resoluto.sandbox.ObjectInfo

Bases: BaseModel

resoluto.sandbox.conduit.factory.store_from_env

store_from_env(env=None)

Build a Conduit from environment variables. Inputs: optional env dict (defaults to os.environ). Output: a concrete Conduit for the requested RESOLUTO_STORE_KIND.

Source code in src/resoluto/sandbox/conduit/factory.py

def store_from_env(env: dict[str, str] | None = None) -> Conduit:
    """Build a Conduit from environment variables. Inputs: optional env dict (defaults
    to os.environ). Output: a concrete Conduit for the requested RESOLUTO_STORE_KIND."""
    env = env if env is not None else os.environ
    kind = env["RESOLUTO_STORE_KIND"]
    if kind == "stdout":
        from resoluto.sandbox.conduit.stdout import StdoutConduit

        return StdoutConduit()
    if kind == "localfs":
        from resoluto.sandbox.conduit import LocalConduit

        return LocalConduit(env["RESOLUTO_STORE_ROOT"])
    if kind == "s3":
        from resoluto.sandbox.conduit.s3 import S3Conduit

        write_token = env.get("RESOLUTO_STORE_WRITE_TOKEN")
        if write_token:
            tok = json.loads(write_token)
            return S3Conduit(
                tok["bucket"],
                endpoint_url=tok.get("endpoint_url"),
                region_name=tok.get("region", "us-east-1"),
                aws_access_key_id=tok["access_key_id"],
                aws_secret_access_key=tok["secret_access_key"],
                aws_session_token=tok.get("session_token"),
            )
        return S3Conduit(
            env["RESOLUTO_STORE_BUCKET"],
            endpoint_url=env.get("RESOLUTO_STORE_ENDPOINT") or None,
            region_name=env.get("RESOLUTO_STORE_REGION", "us-east-1"),
            aws_access_key_id=env.get("AWS_ACCESS_KEY_ID"),
            aws_secret_access_key=env.get("AWS_SECRET_ACCESS_KEY"),
        )
    if kind == "gcs":
        from resoluto.sandbox.conduit.gcs import GcsConduit

        if env.get("RESOLUTO_STORE_WRITE_TOKEN"):
            raise RuntimeError(
                "RESOLUTO_STORE_KIND=gcs cannot honor a prefix-scoped RESOLUTO_STORE_WRITE_TOKEN "
                "(that is the s3 STS path) — GcsConduit is a single-tenant host-side store. "
                "Refusing rather than silently granting whole-service-account access."
            )
        return GcsConduit(
            env["RESOLUTO_STORE_BUCKET"],
            service_file=env.get("RESOLUTO_GCS_SERVICE_FILE"),
        )
    raise RuntimeError(f"unknown RESOLUTO_STORE_KIND={kind!r}")

resoluto.sandbox.LocalConduit

LocalConduit(root, *, world_writable=False)

Bases: Conduit

Source code in src/resoluto/sandbox/conduit/local.py

def __init__(self, root: str | Path, *, world_writable: bool = False) -> None:
    self._root = Path(root)
    self._world_writable = world_writable
    self._root.mkdir(parents=True, exist_ok=True)
    if world_writable:
        self._chmod_world(self._root)

resoluto.sandbox.StdoutConduit

StdoutConduit(*, sink=None)

Bases: Conduit

Source code in src/resoluto/sandbox/conduit/stdout.py

def __init__(self, *, sink: IO[str] | None = None) -> None:
    self._sink = sink if sink is not None else sys.stdout

resoluto.sandbox.conduit.s3.S3Conduit

S3Conduit(
    bucket,
    *,
    endpoint_url=None,
    region_name=None,
    aws_access_key_id=None,
    aws_secret_access_key=None,
    aws_session_token=None,
)

Bases: Conduit

Source code in src/resoluto/sandbox/conduit/s3.py

def __init__(
    self,
    bucket: str,
    *,
    endpoint_url: str | None = None,
    region_name: str | None = None,
    aws_access_key_id: str | None = None,
    aws_secret_access_key: str | None = None,
    aws_session_token: str | None = None,
) -> None:
    self._bucket = bucket
    self._client_kwargs = {
        "endpoint_url": endpoint_url,
        "region_name": region_name,
        "aws_access_key_id": aws_access_key_id,
        "aws_secret_access_key": aws_secret_access_key,
        "aws_session_token": aws_session_token,
    }
    self._session = None

aclose async

aclose()

Drop the cached session so a stale one isn't reused after teardown.

Source code in src/resoluto/sandbox/conduit/s3.py

async def aclose(self) -> None:
    """Drop the cached session so a stale one isn't reused after teardown."""
    self._session = None

ensure_bucket async

ensure_bucket()

Create the bucket if absent.

Source code in src/resoluto/sandbox/conduit/s3.py

async def ensure_bucket(self) -> None:
    """Create the bucket if absent."""
    from botocore.exceptions import ClientError

    async with self._client() as c:
        try:
            await c.head_bucket(Bucket=self._bucket)
        except ClientError:
            await c.create_bucket(Bucket=self._bucket)

resoluto.sandbox.conduit.gcs.GcsConduit

GcsConduit(bucket, *, service_file=None)

Bases: Conduit

Source code in src/resoluto/sandbox/conduit/gcs.py

def __init__(self, bucket: str, *, service_file: str | None = None) -> None:
    self._bucket = bucket
    self._service_file = service_file
    self._storage = None