Storage API Reference

cs_copilot.storage

Cs_copilot Storage Module

Provides unified storage abstraction for local files and S3/MinIO. Handles session management and file operations transparently.

Main Components:

• S3: Storage client class for file operations
• SESSION_ID: Current session identifier
• S3Config: Configuration dataclass for S3 settings

Usage:

from cs_copilot.storage import S3

# Read a file
with S3.open("data/compounds.csv", "r") as f:
    df = pd.read_csv(f)

# Write a file
with S3.open("results/output.csv", "w") as f:
    df.to_csv(f)

# Get S3 path
path = S3.path("results/model.pkl.gz")

S3

Unified storage client for S3/MinIO and local file operations.

All file operations are scoped to the current session by default, unless an absolute S3 URL or local path is provided.

Class Attributes:

prefix : str Session-scoped prefix for all relative paths

Methods:

path(rel: str) -> str Convert a relative path to an S3 URL

open(rel: str, mode: str = "rb") Open a file for reading or writing

Examples:

Write to session-scoped S3 path

with S3.open("results.csv", "w") as f: ... f.write("data")

Read from absolute S3 URL

with S3.open("s3://bucket/key.csv", "r") as f: ... data = f.read()

Read from local file

with S3.open("/tmp/local.csv", "r") as f: ... data = f.read()

Source code in src/cs_copilot/storage/client.py

class S3(metaclass=_S3Meta):
    """
    Unified storage client for S3/MinIO and local file operations.

    All file operations are scoped to the current session by default,
    unless an absolute S3 URL or local path is provided.

    Class Attributes:
    -----------------
    prefix : str
        Session-scoped prefix for all relative paths

    Methods:
    --------
    path(rel: str) -> str
        Convert a relative path to an S3 URL

    open(rel: str, mode: str = "rb")
        Open a file for reading or writing

    Examples:
    ---------
    >>> # Write to session-scoped S3 path
    >>> with S3.open("results.csv", "w") as f:
    ...     f.write("data")

    >>> # Read from absolute S3 URL
    >>> with S3.open("s3://bucket/key.csv", "r") as f:
    ...     data = f.read()

    >>> # Read from local file
    >>> with S3.open("/tmp/local.csv", "r") as f:
    ...     data = f.read()
    """

    _fallback_prefix = _DEFAULT_PREFIX
    logger.info(f"Initialized S3 client with SESSION_ID: {SESSION_ID}")

    @classmethod
    def set_session_prefix(cls, prefix: str) -> None:
        """Set the storage prefix for the current execution context.

        Chainlit serves multiple chat sessions in one Python process.  A plain
        class-level prefix is shared across concurrent chats, so relative
        storage paths can leak into another user's session.  The context-local
        prefix isolates async tasks while keeping ``S3.prefix`` as a fallback
        for CLI/tests/backwards compatibility.
        """
        normalized = _normalize_storage_prefix(prefix)
        cls._fallback_prefix = normalized
        _SESSION_PREFIX.set(normalized)

    @classmethod
    def current_prefix(cls) -> str:
        """Return the active context-local prefix, falling back to ``S3.prefix``."""
        return _SESSION_PREFIX.get() or cls._fallback_prefix

    @classmethod
    @contextmanager
    def scoped_session_prefix(cls, prefix: str) -> Iterator[None]:
        """Bind a session prefix for this execution context only.

        Unlike :meth:`set_session_prefix`, the process-wide fallback is left
        untouched, so concurrent chats in one process cannot leak into each
        other through it.
        """

        token = _SESSION_PREFIX.set(_normalize_storage_prefix(prefix))
        try:
            yield
        finally:
            _SESSION_PREFIX.reset(token)

    @classmethod
    @contextmanager
    def observe_writes(
        cls,
        boundary: str,
        *,
        protected_paths: Iterable[str] = (),
    ) -> Iterator[WriteObservation]:
        """Record, without enforcing, how confined writes would treat each write.

        While active and no :meth:`confine_writes` boundary is set, every write
        through :meth:`open` is classified against ``boundary`` and the
        protected paths, then performed exactly as it would be without the
        observer. Classification errors are recorded, never raised.
        """

        observation = WriteObservation(
            boundary=cls._normalize_relative_path(boundary, allow_empty=True),
            protected=frozenset(str(path) for path in protected_paths),
        )
        token = _WRITE_OBSERVATION.set(observation)
        try:
            yield observation
        finally:
            _WRITE_OBSERVATION.reset(token)

    @classmethod
    def _observe_write(cls, observation: WriteObservation, rel: str, mode: str) -> None:
        try:
            entry = cls._classify_observed_write(observation, rel, mode)
        except Exception:  # noqa: BLE001 - observation must never change a write
            logger.debug("Could not classify an observed write to %r", rel, exc_info=True)
            entry = {"path": str(rel), "verdict": "unclassified", "created": False}
        observation.record({**entry, "mode": mode})

    @classmethod
    def _classify_observed_write(
        cls,
        observation: WriteObservation,
        rel: str,
        mode: str,
    ) -> dict[str, Any]:
        candidate = str(rel).strip()
        actual = cls._verified_read_key(candidate)
        boundary = PurePosixPath(observation.boundary) if observation.boundary else PurePosixPath()
        inside = False
        if actual is not None:
            try:
                PurePosixPath(actual).relative_to(boundary)
                inside = True
            except ValueError:
                inside = False
        existed = cls._storage_key_exists(actual) if inside and actual is not None else False
        token = _WRITE_PROTECTED_PATHS.set(observation.protected)
        try:
            cls._validate_scoped_write(candidate, boundary=observation.boundary)
        except ScopedWriteDenied as denied:
            verdict = denied.verdict
        except (PermissionError, ValueError):
            verdict = "unsafe_path"
        else:
            if "a" in mode or "+" in mode:
                verdict = "append_existing" if existed else "create_new"
            else:
                verdict = "overwrite_existing" if existed else "create_new"
        finally:
            _WRITE_PROTECTED_PATHS.reset(token)
        return {
            "path": actual or candidate,
            "verdict": verdict,
            # Runtime metadata and registered artifacts are never new evidence.
            "created": bool(
                inside and not existed and verdict not in {"runtime_metadata", "protected_artifact"}
            ),
        }

    @classmethod
    def _storage_key_exists(cls, key: str) -> bool:
        if not is_s3_enabled():
            return cls._local_session_path(key).exists()
        filesystem, filesystem_path = fsspec.core.url_to_fs(
            cls.path(key),
            **get_s3_config().to_storage_options(),
        )
        return bool(filesystem.exists(filesystem_path))

    @classmethod
    @contextmanager
    def confine_writes(
        cls,
        boundary: str,
        *,
        protected_paths: Iterable[str] = (),
        publication_receipt: dict[str, dict[str, Any]] | None = None,
        commit_policy: str = "on_exit",
    ) -> Iterator[None]:
        """Confine relative writes to one session subtree for this execution context.

        The MCP adapter activates this guard while a session-scoped tool runs.
        It is context-local, propagates through ``asyncio.to_thread``, and is
        enforced again by the worker process. Existing registered artifacts
        can be supplied as protected paths so a tool cannot mutate immutable
        scientific evidence. Reads remain unaffected.

        ``commit_policy="on_exit"`` (default) keeps every write private until
        the scope exits cleanly. ``"on_close"`` publishes each file as soon as
        it is closed, so code that reads its own outputs through other APIs
        keeps working; the files are still recorded in ``publication_receipt``
        and rolled back if the scope exits with an exception.
        """

        if commit_policy == "on_close":
            with cls._confine_writes_on_close(
                boundary,
                protected_paths=protected_paths,
                publication_receipt=publication_receipt,
            ):
                yield
            return
        if commit_policy != "on_exit":
            raise ValueError(f"unknown commit policy: {commit_policy!r}")

        normalized = cls._normalize_relative_path(boundary, allow_empty=True)
        boundary_path = PurePosixPath(normalized) if normalized else PurePosixPath()
        protected: set[str] = set()
        for path in protected_paths:
            protected_path = cls._normalize_relative_path(path)
            try:
                PurePosixPath(protected_path).relative_to(boundary_path)
            except ValueError as exc:
                raise ValueError(
                    "protected write path must remain inside the active boundary"
                ) from exc
            protected.add(protected_path)
        normalized_protected = frozenset(protected)
        outer_pending = _PENDING_ARTIFACT_WRITES.get()
        pending: list[Any] = outer_pending if outer_pending is not None else []
        owns_transaction = outer_pending is None
        outer_staged = _STAGED_ARTIFACT_WRITES.get()
        staged: dict[str, Any] = outer_staged if outer_staged is not None else {}
        boundary_token = _WRITE_BOUNDARY.set(normalized)
        protected_token = _WRITE_PROTECTED_PATHS.set(normalized_protected)
        pending_token = _PENDING_ARTIFACT_WRITES.set(pending)
        staged_token = _STAGED_ARTIFACT_WRITES.set(staged)
        try:
            yield
        except BaseException:
            if owns_transaction:
                for writer in reversed(pending):
                    writer._abort()
            raise
        else:
            if owns_transaction:
                committed: list[Any] = []
                prepared_publications: dict[str, dict[str, Any]] = {}
                try:
                    if publication_receipt is not None:
                        for writer in pending:
                            staged_key = getattr(writer, "_staged_key", None)
                            if staged_key is None:
                                continue
                            prepared_publications[staged_key] = cls._pending_writer_metadata(
                                writer, staged_key
                            )
                    for writer in pending:
                        writer._commit()
                        committed.append(writer)
                except BaseException:
                    cleanup_errors: list[BaseException] = []
                    for writer in reversed(pending[len(committed) :]):
                        try:
                            writer._abort()
                        except BaseException as exc:
                            cleanup_errors.append(exc)
                            logger.warning(
                                "Could not abort a pending artifact publication",
                                exc_info=True,
                            )
                    for writer in reversed(committed):
                        try:
                            writer._rollback_published()
                        except BaseException as exc:
                            cleanup_errors.append(exc)
                            logger.warning(
                                "Refused or failed to roll back a published artifact",
                                exc_info=True,
                            )
                    if cleanup_errors:
                        raise RuntimeError(
                            "artifact transaction failed and could not be fully rolled back"
                        ) from cleanup_errors[0]
                    raise
                else:
                    if publication_receipt is not None:
                        publication_receipt.update(prepared_publications)
        finally:
            _STAGED_ARTIFACT_WRITES.reset(staged_token)
            _PENDING_ARTIFACT_WRITES.reset(pending_token)
            _WRITE_PROTECTED_PATHS.reset(protected_token)
            _WRITE_BOUNDARY.reset(boundary_token)

    @classmethod
    @contextmanager
    def _confine_writes_on_close(
        cls,
        boundary: str,
        *,
        protected_paths: Iterable[str],
        publication_receipt: dict[str, dict[str, Any]] | None,
    ) -> Iterator[None]:
        if _PENDING_ARTIFACT_WRITES.get() is not None:
            raise RuntimeError("on-close confinement cannot nest inside a deferred transaction")
        normalized = cls._normalize_relative_path(boundary, allow_empty=True)
        boundary_path = PurePosixPath(normalized) if normalized else PurePosixPath()
        protected: set[str] = set()
        for path in protected_paths:
            protected_path = cls._normalize_relative_path(path)
            try:
                PurePosixPath(protected_path).relative_to(boundary_path)
            except ValueError as exc:
                raise ValueError(
                    "protected write path must remain inside the active boundary"
                ) from exc
            protected.add(protected_path)
        committed: list[Any] = []
        boundary_token = _WRITE_BOUNDARY.set(normalized)
        protected_token = _WRITE_PROTECTED_PATHS.set(frozenset(protected))
        committed_token = _COMMITTED_ON_CLOSE.set(committed)
        try:
            yield
        except BaseException:
            cleanup_errors: list[BaseException] = []
            for writer in reversed(committed):
                try:
                    writer._rollback_published()
                except BaseException as exc:
                    cleanup_errors.append(exc)
                    logger.warning(
                        "Refused or failed to roll back a published artifact",
                        exc_info=True,
                    )
            if cleanup_errors:
                raise RuntimeError(
                    "artifact transaction failed and could not be fully rolled back"
                ) from cleanup_errors[0]
            raise
        else:
            if publication_receipt is not None:
                for writer in committed:
                    key = getattr(writer, "_staged_key", None)
                    if key is None or writer._expected_sha256 is None:
                        continue
                    publication_receipt[key] = {
                        "staged_path": key,
                        "sha256": writer._expected_sha256,
                        "size_bytes": writer._expected_size,
                    }
        finally:
            _COMMITTED_ON_CLOSE.reset(committed_token)
            _WRITE_PROTECTED_PATHS.reset(protected_token)
            _WRITE_BOUNDARY.reset(boundary_token)

    @classmethod
    def exists(cls, rel: str) -> bool:
        """Return whether a storage path (relative, local, or S3 URL) exists."""

        candidate = str(rel).strip()
        if cls._is_explicit_local_path(candidate):
            local = (
                unquote(urlsplit(candidate).path) if candidate.startswith("file://") else candidate
            )
            return Path(local).exists()
        if not cls._is_s3_url(candidate) and not is_s3_enabled():
            return cls._local_session_path(candidate).exists()
        url = candidate if cls._is_s3_url(candidate) else cls.path(candidate)
        filesystem, filesystem_path = fsspec.core.url_to_fs(
            url, **get_s3_config().to_storage_options()
        )
        filesystem.invalidate_cache(filesystem_path.rsplit("/", 1)[0])
        return bool(filesystem.exists(filesystem_path))

    @classmethod
    def first_free_path(cls, rel: str) -> str:
        """Return ``rel``, or its first free ``<stem>-v<N><suffix>`` sibling.

        Workflow outputs are immutable once registered and confined writes are
        create-only, so re-running a step writes a new version of each output
        instead of overwriting the previous one. The same spelling (relative,
        local, or S3 URL) is returned.
        """

        if not cls.exists(rel):
            return rel
        version = 2
        while cls.exists(_versioned_name(rel, version)):
            version += 1
        return _versioned_name(rel, version)

    @classmethod
    def latest_version_path(cls, rel: str) -> str:
        """Return the newest existing version of ``rel`` written by :meth:`first_free_path`.

        Readers that locate an output by its deterministic name use this to
        find the most recent re-run. ``rel`` itself is returned when it has no
        later version (or does not exist).
        """

        latest = rel
        if not cls.exists(rel):
            return rel
        version = 2
        while cls.exists(_versioned_name(rel, version)):
            latest = _versioned_name(rel, version)
            version += 1
        return latest

    @classmethod
    def session_relative(cls, path: str) -> str | None:
        """Return the session-relative key of a path in the active session, if any."""

        return cls._verified_read_key(path)

    @staticmethod
    def _pending_writer_metadata(writer: Any, final_key: str) -> dict[str, Any]:
        """Checksum a private transaction object before it is published."""

        digest = hashlib.sha256()
        size = 0
        with writer._open_staged_read("rb") as source:
            while True:
                chunk = source.read(1024 * 1024)
                if not chunk:
                    break
                if isinstance(chunk, str):
                    chunk = chunk.encode("utf-8")
                digest.update(chunk)
                size += len(chunk)
        return {
            "staged_path": final_key,
            "sha256": digest.hexdigest(),
            "size_bytes": size,
        }

    @staticmethod
    def current_write_boundary() -> str | None:
        """Return the active write boundary, or ``None`` outside guarded execution."""

        return _WRITE_BOUNDARY.get()

    @staticmethod
    def current_write_protected_paths() -> frozenset[str]:
        """Return immutable session-relative paths for the active tool call."""

        return _WRITE_PROTECTED_PATHS.get()

    @classmethod
    @contextmanager
    def stage_publications(cls, staging_id: str) -> Iterator[None]:
        """Redirect scoped writes to a worker-private publication namespace."""

        normalized_id = cls._normalize_relative_path(staging_id)
        if len(PurePosixPath(normalized_id).parts) != 1:
            raise ValueError("staging_id must be one safe path component")
        publications: dict[str, str] = {}
        id_token = _EXTERNAL_STAGING_ID.set(normalized_id)
        publications_token = _EXTERNAL_STAGED_PUBLICATIONS.set(publications)
        try:
            yield
        finally:
            _EXTERNAL_STAGED_PUBLICATIONS.reset(publications_token)
            _EXTERNAL_STAGING_ID.reset(id_token)

    @staticmethod
    def current_staged_publications() -> dict[str, str]:
        """Return final-to-staged keys created by the active worker invocation."""

        return dict(_EXTERNAL_STAGED_PUBLICATIONS.get() or {})

    @classmethod
    def staged_publication_metadata(cls) -> dict[str, dict[str, Any]]:
        """Checksum committed worker staging objects through a no-follow read."""

        metadata: dict[str, dict[str, Any]] = {}
        for final_key, staged_key in cls.current_staged_publications().items():
            digest = hashlib.sha256()
            size = 0
            if is_s3_enabled():
                config = get_s3_config()
                source = fsspec.open(
                    cls.path(staged_key),
                    mode="rb",
                    **config.to_storage_options(),
                ).open()
            else:
                source = cls._open_local_verified_source(staged_key)
            with source:
                while True:
                    chunk = source.read(1024 * 1024)
                    if not chunk:
                        break
                    if isinstance(chunk, str):
                        chunk = chunk.encode("utf-8")
                    digest.update(chunk)
                    size += len(chunk)
            metadata[final_key] = {
                "staged_path": staged_key,
                "sha256": digest.hexdigest(),
                "size_bytes": size,
            }
        return metadata

    @classmethod
    def promote_staged_publications(
        cls,
        publications: Mapping[str, Mapping[str, Any]],
    ) -> None:
        """Promote complete worker outputs to final create-only artifact keys."""

        promoted: dict[str, Mapping[str, Any]] = {}
        try:
            for final_key, metadata in publications.items():
                final = cls._normalize_relative_path(final_key)
                staged = cls._normalize_relative_path(str(metadata["staged_path"]))
                source = cls._open_verified_snapshot(
                    staged,
                    "rb",
                    expected_sha256=str(metadata["sha256"]),
                    expected_size=int(metadata["size_bytes"]),
                )
                with source:
                    with cls.open_atomic(final, "xb") as destination:
                        while True:
                            chunk = source.read(1024 * 1024)
                            if not chunk:
                                break
                            destination.write(chunk)
                promoted[final] = metadata
            for metadata in publications.values():
                cls._remove_storage_key(cls._normalize_relative_path(str(metadata["staged_path"])))
        except BaseException:
            cls.rollback_promoted_publications(promoted)
            raise

    @classmethod
    def rollback_promoted_publications(
        cls,
        publications: Mapping[str, Mapping[str, Any]],
    ) -> None:
        """Remove only finals that still match the worker's promoted bytes."""

        for final_key, metadata in reversed(tuple(publications.items())):
            final = cls._normalize_relative_path(final_key)
            try:
                cls._remove_storage_key_verified(
                    final,
                    expected_sha256=str(metadata["sha256"]),
                    expected_size=int(metadata["size_bytes"]),
                )
            except FileNotFoundError:
                continue
            except Exception:
                logger.warning(
                    "Refusing to roll back changed worker publication %s",
                    final,
                    exc_info=True,
                )
                continue

    @classmethod
    def discard_staging_prefix(cls, boundary: str, staging_id: str) -> None:
        """Remove one exact worker staging subtree after failure or cancellation."""

        normalized_boundary = cls._normalize_relative_path(boundary)
        normalized_id = cls._normalize_relative_path(staging_id)
        if len(PurePosixPath(normalized_id).parts) != 1:
            raise ValueError("staging_id must be one safe path component")
        prefix = PurePosixPath(
            normalized_boundary,
            ".staging",
            normalized_id,
        ).as_posix()
        if is_s3_enabled():
            config = get_s3_config()
            fs, fs_path = fsspec.core.url_to_fs(
                cls.path(prefix),
                **config.to_storage_options(),
            )
            if fs.exists(fs_path):
                fs.rm(fs_path, recursive=True)
            return
        _secure_local_rmtree(cls._local_session_path(prefix).absolute())

    @classmethod
    def _remove_storage_key(cls, key: str) -> None:
        if is_s3_enabled():
            config = get_s3_config()
            fs, fs_path = fsspec.core.url_to_fs(
                cls.path(key),
                **config.to_storage_options(),
            )
            if fs.exists(fs_path):
                fs.rm(fs_path)
            return
        _secure_local_unlink(cls._local_session_path(key).absolute())

    @classmethod
    def _remove_storage_key_verified(
        cls,
        key: str,
        *,
        expected_sha256: str,
        expected_size: int,
    ) -> bool:
        """Remove one exact publication only while its registered bytes match."""

        if is_s3_enabled():
            config = get_s3_config()
            fs, fs_path = fsspec.core.url_to_fs(
                cls.path(key),
                **config.to_storage_options(),
            )
            return _remove_fsspec_path_verified(
                fs,
                fs_path,
                expected_sha256=expected_sha256,
                expected_size=expected_size,
            )
        return _secure_local_unlink_verified(
            cls._local_session_path(key).absolute(),
            expected_sha256=expected_sha256,
            expected_size=expected_size,
        )

    @classmethod
    @contextmanager
    def confine_artifact_reads(
        cls,
        artifacts: Mapping[str, tuple[str, int]],
    ) -> Iterator[None]:
        """Pin registered artifact reads to checksum-verified byte snapshots."""

        inherited = dict(_VERIFIED_ARTIFACT_READS.get() or {})
        for path, expected in artifacts.items():
            normalized = cls._normalize_relative_path(path)
            if (
                not isinstance(expected, tuple)
                or len(expected) != 2
                or not isinstance(expected[0], str)
                or not isinstance(expected[1], int)
                or expected[1] < 0
            ):
                raise ValueError("verified artifact metadata must be (sha256, size_bytes)")
            inherited[normalized] = (expected[0], expected[1])
        token = _VERIFIED_ARTIFACT_READS.set(inherited)
        try:
            yield
        finally:
            _VERIFIED_ARTIFACT_READS.reset(token)

    @staticmethod
    def current_verified_artifact_reads() -> dict[str, tuple[str, int]]:
        """Return a detached copy of invocation-pinned artifact read metadata."""

        return dict(_VERIFIED_ARTIFACT_READS.get() or {})

    @classmethod
    def path(cls, rel: str) -> str:
        """
        Convert a relative path to an S3 URL or local session path.

        Explicit S3 and local paths are returned unchanged. Relative paths are
        resolved against the active backend.

        Args:
            rel: Relative path or absolute S3 URL

        Returns:
            str: Full S3 URL or local path

        Examples:
        ---------
        >>> S3.path("data.csv")
        's3://chatbot-assets/sessions/20250121-123456-abc123/data.csv'

        >>> S3.path("s3://mybucket/data.csv")
        's3://mybucket/data.csv'
        """
        if cls._is_s3_url(rel) or cls._is_explicit_local_path(rel):
            return rel

        config = get_s3_config()
        if not is_s3_enabled():
            return os.fspath(cls._local_session_path(rel))

        key = PurePosixPath(cls.current_prefix().strip("/")) / str(rel).lstrip("/")
        return f"s3://{config.bucket_name}/{key.as_posix()}"

    @classmethod
    def open(cls, rel: str, mode: str = "rb") -> Any:
        """
        Open a file for reading or writing.

        Supports three types of paths:
        1. Absolute S3 URLs (s3://...)
        2. Local absolute paths (/ or file://)
        3. Relative paths (scoped to current session)

        Args:
            rel: File path (relative, absolute, or S3 URL)
            mode: File mode ('r', 'w', 'rb', 'wb', etc.)

        Returns:
            File-like object opened with the specified mode

        Examples:
        ---------
        >>> # Read from session-scoped S3 path
        >>> with S3.open("data.csv", "r") as f:
        ...     df = pd.read_csv(f)

        >>> # Write to session-scoped S3 path
        >>> with S3.open("output.csv", "w") as f:
        ...     df.to_csv(f)

        >>> # Read from absolute S3 URL
        >>> with S3.open("s3://bucket/data.csv", "r") as f:
        ...     df = pd.read_csv(f)

        >>> # Read from local file
        >>> with S3.open("/tmp/data.csv", "r") as f:
        ...     df = pd.read_csv(f)
        """
        config = get_s3_config()
        write_mode = cls._is_write_mode(mode)
        if not write_mode:
            read_key = cls._verified_read_key(rel)
            staged = (
                _STAGED_ARTIFACT_WRITES.get().get(read_key)
                if read_key is not None and _STAGED_ARTIFACT_WRITES.get() is not None
                else None
            )
            if staged is not None:
                return staged._open_staged_read(mode)
            expected_reads = _VERIFIED_ARTIFACT_READS.get() or {}
            expected = expected_reads.get(read_key) if read_key else None
            if read_key is not None and expected is not None:
                return cls._open_verified_snapshot(
                    read_key,
                    mode,
                    expected_sha256=expected[0],
                    expected_size=expected[1],
                )

        write_boundary = cls.current_write_boundary() if write_mode else None
        if write_mode and write_boundary is None:
            observation = _WRITE_OBSERVATION.get()
            if observation is not None:
                cls._observe_write(observation, rel, mode)
        scoped_write_key: str | None = None
        if write_boundary is not None:
            scoped_write_key = cls._validate_scoped_write(rel, boundary=write_boundary)
        storage_target = rel
        staging_id = _EXTERNAL_STAGING_ID.get()
        publications = _EXTERNAL_STAGED_PUBLICATIONS.get()
        if scoped_write_key is not None and staging_id is not None and publications is not None:
            relative_to_boundary = PurePosixPath(scoped_write_key).relative_to(
                PurePosixPath(write_boundary or "")
            )
            staged_key = PurePosixPath(
                write_boundary or "",
                ".staging",
                staging_id,
                relative_to_boundary,
            ).as_posix()
            publications[scoped_write_key] = staged_key
            storage_target = staged_key

        if cls._is_s3_url(storage_target):
            if scoped_write_key is None:
                return fsspec.open(
                    storage_target,
                    mode=mode,
                    **config.to_storage_options(),
                )
            return cls._open_confined_s3_path(
                storage_target,
                mode,
                storage_options=config.to_storage_options(),
                public_write_key=scoped_write_key,
            )

        if cls._is_explicit_local_path(storage_target):
            if storage_target.startswith("file://"):
                return fsspec.open(storage_target, mode=mode)
            return cls._open_local_path(Path(storage_target), mode)

        if not is_s3_enabled():
            return cls._open_local_path(
                cls._local_session_path(storage_target),
                mode,
                scoped_write_key=scoped_write_key,
            )

        if scoped_write_key is None:
            return fsspec.open(
                cls.path(storage_target),
                mode=mode,
                **config.to_storage_options(),
            )
        return cls._open_confined_s3_path(
            cls.path(storage_target),
            mode,
            storage_options=config.to_storage_options(),
            public_write_key=scoped_write_key,
        )

    @classmethod
    def _verified_read_key(cls, path: str) -> str | None:
        """Resolve a public storage spelling to a session-relative artifact key."""

        if not isinstance(path, str) or not path.strip():
            return None
        candidate = path.strip()
        try:
            if cls._is_s3_url(candidate):
                parsed = urlsplit(candidate)
                config = get_s3_config()
                prefix = cls.current_prefix().strip("/")
                object_path = unquote(parsed.path).lstrip("/")
                expected_prefix = f"{prefix}/"
                if (
                    parsed.query
                    or parsed.fragment
                    or parsed.netloc != config.bucket_name
                    or not object_path.startswith(expected_prefix)
                ):
                    return None
                return cls._normalize_relative_path(object_path[len(expected_prefix) :])
            if cls._is_explicit_local_path(candidate):
                local_value = (
                    unquote(urlsplit(candidate).path)
                    if candidate.startswith("file://")
                    else candidate
                )
                local_path = Path(local_value).absolute()
                session_root = cls._local_session_path("").absolute()
                return cls._normalize_relative_path(local_path.relative_to(session_root).as_posix())
            expanded_prefix = PurePosixPath(
                LOCAL_STORAGE_ROOT.as_posix(),
                cls.current_prefix(),
            )
            candidate_path = PurePosixPath(candidate)
            try:
                expanded_relative = candidate_path.relative_to(expanded_prefix)
            except ValueError:
                pass
            else:
                return cls._normalize_relative_path(expanded_relative.as_posix())
            return cls._normalize_relative_path(candidate)
        except (ValueError, OSError):
            return None

    @classmethod
    def _open_verified_snapshot(
        cls,
        key: str,
        mode: str,
        *,
        expected_sha256: str,
        expected_size: int,
    ):
        """Copy one artifact once, verify it, and return the pinned byte snapshot."""

        if mode not in {"r", "rt", "rb"}:
            raise PermissionError("verified artifacts are available only in read mode")
        if is_s3_enabled():
            config = get_s3_config()
            source = fsspec.open(
                cls.path(key),
                mode="rb",
                **config.to_storage_options(),
            ).open()
        else:
            source = cls._open_local_verified_source(key)

        snapshot = tempfile.SpooledTemporaryFile(max_size=16 * 1024 * 1024, mode="w+b")
        digest = hashlib.sha256()
        size = 0
        try:
            with source:
                while True:
                    chunk = source.read(1024 * 1024)
                    if not chunk:
                        break
                    if isinstance(chunk, str):
                        chunk = chunk.encode("utf-8")
                    size += len(chunk)
                    if size > expected_size:
                        raise PermissionError(
                            f"registered artifact {key!r} exceeded its expected size"
                        )
                    digest.update(chunk)
                    snapshot.write(chunk)
            if size != expected_size or digest.hexdigest() != expected_sha256:
                raise PermissionError(f"registered artifact {key!r} failed checksum verification")
            snapshot.seek(0)
            if "b" in mode:
                return snapshot
            return io.TextIOWrapper(snapshot, encoding="utf-8")
        except BaseException:
            snapshot.close()
            raise

    @classmethod
    def _open_local_verified_source(cls, key: str):
        """Descriptor-open a registered local artifact without following links."""

        if not _OPEN_SUPPORTS_DIR_FD or not _O_NOFOLLOW:
            raise RuntimeError("secure local reads require directory-relative no-follow support")
        relative = PurePosixPath(cls._normalize_relative_path(key))
        common_flags = os.O_RDONLY | getattr(os, "O_CLOEXEC", 0) | _O_NOFOLLOW
        directory_flags = common_flags | getattr(os, "O_DIRECTORY", 0)
        current_fd: int | None = None
        file_fd: int | None = None
        try:
            current_fd = _open_existing_directory_tree(cls._local_session_path("").absolute())
            for component in relative.parts[:-1]:
                next_fd = os.open(component, directory_flags, dir_fd=current_fd)
                os.close(current_fd)
                current_fd = next_fd
            file_fd = os.open(relative.parts[-1], common_flags, dir_fd=current_fd)
            metadata = os.fstat(file_fd)
            if not stat.S_ISREG(metadata.st_mode) or metadata.st_nlink != 1:
                raise PermissionError("registered artifact must be a regular, non-linked file")
            handle = os.fdopen(file_fd, "rb")
            file_fd = None
            return handle
        except OSError as exc:
            if exc.errno in {errno.ELOOP, errno.ENOTDIR}:
                raise PermissionError(
                    "registered artifact contains a symlink or non-directory component"
                ) from exc
            raise
        finally:
            if file_fd is not None:
                os.close(file_fd)
            if current_fd is not None:
                os.close(current_fd)

    @classmethod
    def open_atomic(cls, rel: str, mode: str = "xb"):
        """Create one session-relative object that is invisible until complete."""

        if not cls._is_write_mode(mode):
            raise ValueError("open_atomic requires a write-capable mode")
        if cls._is_s3_url(rel) or cls._is_explicit_local_path(rel):
            raise ValueError("open_atomic requires a session-relative storage path")
        normalized = cls._normalize_relative_path(rel)
        config = get_s3_config()
        if not is_s3_enabled():
            return cls._open_confined_local_path(
                cls._local_session_path(normalized),
                mode,
                boundary="",
                scoped_write_key=normalized,
                defer_transaction=False,
            )
        return cls._open_confined_s3_path(
            cls.path(normalized),
            mode,
            storage_options=config.to_storage_options(),
            defer_transaction=False,
            public_write_key=None,
        )

    @staticmethod
    def _is_s3_url(path: str) -> bool:
        return isinstance(path, str) and path.startswith("s3://")

    @staticmethod
    def _is_explicit_local_path(path: str) -> bool:
        return isinstance(path, str) and (path.startswith("/") or path.startswith("file://"))

    @classmethod
    def _local_session_path(cls, rel: str) -> Path:
        session_root = LOCAL_STORAGE_ROOT / Path(cls.current_prefix().strip("/"))
        candidate = Path(rel)
        try:
            candidate.relative_to(session_root)
        except ValueError:
            return session_root / candidate
        return candidate

    @staticmethod
    def _is_write_mode(mode: str) -> bool:
        return any(flag in mode for flag in ("w", "a", "x", "+"))

    @classmethod
    def _open_local_path(
        cls,
        path: Path,
        mode: str,
        *,
        scoped_write_key: str | None = None,
    ):
        if cls._is_write_mode(mode):
            boundary = cls.current_write_boundary()
            if boundary is not None:
                return cls._open_confined_local_path(
                    path,
                    mode,
                    boundary=boundary,
                    scoped_write_key=scoped_write_key,
                )
            path.parent.mkdir(parents=True, exist_ok=True)
        return builtins.open(path, mode)

    @classmethod
    def _validate_scoped_write(cls, rel: str, *, boundary: str) -> str:
        """Validate a write destination against the active session/run prefix."""

        if not isinstance(rel, str) or not rel.strip():
            raise PermissionError("write destination must be a non-empty storage path")
        candidate = rel.strip()
        if "\x00" in candidate or "\\" in candidate:
            raise PermissionError("write destination contains an unsafe path component")

        if cls._is_s3_url(candidate):
            parsed = urlsplit(candidate)
            expected = urlsplit(cls.path(boundary))
            decoded_path = unquote(parsed.path)
            expected_path = unquote(expected.path).rstrip("/")
            if (
                parsed.query
                or parsed.fragment
                or parsed.netloc != expected.netloc
                or parsed.scheme.lower() != "s3"
                or expected.scheme.lower() != "s3"
                or not decoded_path.startswith(f"{expected_path}/")
            ):
                raise ScopedWriteDenied(
                    "S3 write destination is outside the active boundary",
                    verdict="outside_boundary",
                )
            relative = decoded_path[len(expected_path) + 1 :]
            cls._normalize_relative_path(relative)
            cls._reject_reserved_run_path(relative)
            session_relative = (
                PurePosixPath(boundary, relative).as_posix() if boundary else relative
            )
            cls._reject_protected_write(session_relative)
            return session_relative

        if cls._is_explicit_local_path(candidate):
            raise ScopedWriteDenied(
                "absolute and file:// write destinations are forbidden during scoped execution",
                verdict="absolute_path",
            )

        normalized = cls._normalize_relative_path(candidate)
        normalized_path = PurePosixPath(normalized)
        boundary_path = PurePosixPath(boundary) if boundary else PurePosixPath()
        try:
            relative = normalized_path.relative_to(boundary_path)
        except ValueError as exc:
            raise ScopedWriteDenied(
                "write destination is outside the active boundary",
                verdict="outside_boundary",
            ) from exc
        cls._reject_reserved_run_path(relative.as_posix())
        cls._reject_protected_write(normalized)
        return normalized

    @staticmethod
    def _normalize_relative_path(value: str, *, allow_empty: bool = False) -> str:
        raw = str(value).strip()
        if allow_empty and not raw:
            return ""
        if raw.startswith("/"):
            raise ValueError("storage boundary must be a safe relative path")
        text = raw.rstrip("/") if allow_empty else raw
        candidate = PurePosixPath(text)
        if (
            not text
            or not candidate.parts
            or candidate.is_absolute()
            or "\\" in text
            or any(part in {"", ".", ".."} for part in candidate.parts)
        ):
            raise ValueError("storage boundary must be a safe relative path")
        return candidate.as_posix()

    @staticmethod
    def _reject_reserved_run_path(relative: str) -> None:
        path = PurePosixPath(relative)
        parts = path.parts
        if len(parts) >= 3 and parts[0] == "workflows":
            path = PurePosixPath(*parts[2:])
            parts = path.parts
        normalized = path.as_posix()
        if normalized in _RESERVED_RUN_FILES or (parts and parts[0] in {".staging", "events"}):
            raise ScopedWriteDenied(
                "write destination is reserved for workflow runtime metadata",
                verdict="runtime_metadata",
            )

    @staticmethod
    def _reject_protected_write(session_relative: str) -> None:
        if session_relative in _WRITE_PROTECTED_PATHS.get():
            raise ScopedWriteDenied(
                "write destination is an immutable registered workflow artifact",
                verdict="protected_artifact",
            )

    @classmethod
    def _open_confined_s3_path(
        cls,
        path: str,
        mode: str,
        *,
        storage_options: dict,
        defer_transaction: bool = True,
        public_write_key: str | None = None,
    ):
        """Open one scoped S3 artifact with create-once semantics."""

        selected_mode = cls._exclusive_creation_mode(mode)
        deferred = defer_transaction and _PENDING_ARTIFACT_WRITES.get() is not None
        try:
            filesystem, filesystem_path = fsspec.core.url_to_fs(
                path,
                **storage_options,
            )
            handle = filesystem.open(
                filesystem_path,
                mode=selected_mode,
                autocommit=not deferred,
            )
        except (FileExistsError, IsADirectoryError) as exc:
            raise PermissionError(
                "scoped artifact writes are create-only; destination already exists"
            ) from exc
        except OSError as exc:
            if exc.errno == errno.EEXIST:
                raise PermissionError(
                    "scoped artifact writes are create-only; destination already exists"
                ) from exc
            raise
        return _AtomicS3CreateFile(
            handle,
            deferred=deferred,
            mode=selected_mode,
            staged_key=public_write_key or cls._verified_read_key(path),
        )

    @staticmethod
    def _exclusive_creation_mode(mode: str) -> str:
        """Translate the first scoped S3 write into an exclusive create."""

        if not isinstance(mode, str):
            raise ValueError("file mode must be a string")
        bases = [character for character in "rwax" if character in mode]
        if len(bases) != 1:
            raise ValueError(f"invalid file mode: {mode!r}")
        base = bases[0]
        if base == "x":
            return mode
        if base in {"w", "a"}:
            return mode.replace(base, "x", 1)
        raise PermissionError("scoped update modes cannot modify a pre-existing artifact")

    @classmethod
    def _open_confined_local_path(
        cls,
        path: Path,
        mode: str,
        *,
        boundary: str,
        scoped_write_key: str | None,
        defer_transaction: bool = True,
    ):
        """Open a local write by descriptor without following any symlink."""

        if not _OPEN_SUPPORTS_DIR_FD or not _O_NOFOLLOW:
            raise RuntimeError("secure local writes require directory-relative no-follow support")

        session_root = cls._local_session_path("").absolute()
        boundary_root = session_root.joinpath(*PurePosixPath(boundary).parts).absolute()
        destination = path.absolute()
        try:
            relative = destination.relative_to(boundary_root)
        except ValueError as exc:
            raise ScopedWriteDenied(
                "write destination is outside the active boundary",
                verdict="outside_boundary",
            ) from exc
        if not relative.parts:
            raise PermissionError("write destination must identify a file")

        directory_flags = (
            os.O_RDONLY | getattr(os, "O_CLOEXEC", 0) | getattr(os, "O_DIRECTORY", 0) | _O_NOFOLLOW
        )
        current_fd: int | None = None
        file_fd: int | None = None
        try:
            current_fd = cls._open_or_create_directory_tree(
                boundary_root,
                flags=directory_flags,
            )
            for component in relative.parts[:-1]:
                try:
                    os.mkdir(component, mode=0o700, dir_fd=current_fd)
                except FileExistsError:
                    pass
                next_fd = os.open(component, directory_flags, dir_fd=current_fd)
                os.close(current_fd)
                current_fd = next_fd

            if scoped_write_key is not None:
                destination_name = relative.parts[-1]
                temporary_name = f".{destination_name}.{uuid.uuid4().hex}.pending"
                flags, truncate = cls._write_open_flags(mode, exclusive_create=True)
                file_fd = os.open(temporary_name, flags, 0o600, dir_fd=current_fd)
            else:
                destination_name = relative.parts[-1]
                temporary_name = ""
                flags, truncate = cls._write_open_flags(mode)
                file_fd = os.open(destination_name, flags, 0o600, dir_fd=current_fd)
            metadata = os.fstat(file_fd)
            if not stat.S_ISREG(metadata.st_mode) or metadata.st_nlink != 1:
                raise PermissionError(
                    "write destination must be a regular, non-linked artifact file"
                )
            if truncate:
                os.ftruncate(file_fd, 0)
            fdopen_mode = mode.replace("x", "w", 1) if "x" in mode else mode
            handle = os.fdopen(file_fd, fdopen_mode)
            file_fd = None
            if scoped_write_key is not None:
                owned_directory_fd = current_fd
                current_fd = None
                return _AtomicLocalCreateFile(
                    handle,
                    directory_fd=owned_directory_fd,
                    temporary_name=temporary_name,
                    destination_name=destination_name,
                    destination_path=destination,
                    staged_key=scoped_write_key,
                    deferred=(defer_transaction and _PENDING_ARTIFACT_WRITES.get() is not None),
                )
            return handle
        except OSError as exc:
            if exc.errno == errno.EEXIST:
                raise PermissionError(
                    "scoped artifact writes are create-only; destination already exists"
                ) from exc
            if exc.errno in {errno.ELOOP, errno.ENOTDIR}:
                raise PermissionError(
                    "write destination contains a symlink or non-directory component"
                ) from exc
            raise
        finally:
            if file_fd is not None:
                os.close(file_fd)
            if current_fd is not None:
                os.close(current_fd)

    @staticmethod
    def _open_or_create_directory_tree(path: Path, *, flags: int) -> int:
        """Return a descriptor for an absolute directory tree, creating it safely."""

        if not path.is_absolute():
            raise ValueError("secure directory traversal requires an absolute path")
        current_fd = os.open(path.anchor, flags)
        try:
            for component in path.parts[1:]:
                try:
                    os.mkdir(component, mode=0o700, dir_fd=current_fd)
                except FileExistsError:
                    pass
                next_fd = os.open(component, flags, dir_fd=current_fd)
                os.close(current_fd)
                current_fd = next_fd
            result = current_fd
            current_fd = -1
            return result
        finally:
            if current_fd >= 0:
                os.close(current_fd)

    @staticmethod
    def _write_open_flags(
        mode: str,
        *,
        exclusive_create: bool = False,
    ) -> tuple[int, bool]:
        """Translate a Python write mode into no-follow ``os.open`` flags."""

        if not isinstance(mode, str):
            raise ValueError("file mode must be a string")
        if any(character not in "rwaxbt+" for character in mode):
            raise ValueError(f"unsupported file mode: {mode!r}")
        bases = [character for character in "rwax" if character in mode]
        if len(bases) != 1 or ("b" in mode and "t" in mode):
            raise ValueError(f"invalid file mode: {mode!r}")

        base = bases[0]
        if base == "r" and "+" not in mode:
            raise ValueError("confined writer requires a write-capable file mode")
        access = os.O_RDWR if "+" in mode else os.O_WRONLY
        flags = access | getattr(os, "O_CLOEXEC", 0) | getattr(os, "O_NONBLOCK", 0) | _O_NOFOLLOW
        truncate = False
        if base == "w":
            flags |= os.O_CREAT
            if exclusive_create:
                flags |= os.O_EXCL
            truncate = True
        elif base == "a":
            flags |= os.O_CREAT | os.O_APPEND
            if exclusive_create:
                flags |= os.O_EXCL
        elif base == "x":
            flags |= os.O_CREAT | os.O_EXCL
        elif base == "r" and exclusive_create:
            raise PermissionError("scoped update modes cannot modify a pre-existing artifact")
        return flags, truncate

set_session_prefix(prefix) classmethod

Set the storage prefix for the current execution context.

Chainlit serves multiple chat sessions in one Python process. A plain class-level prefix is shared across concurrent chats, so relative storage paths can leak into another user's session. The context-local prefix isolates async tasks while keeping S3.prefix as a fallback for CLI/tests/backwards compatibility.

Source code in src/cs_copilot/storage/client.py

@classmethod
def set_session_prefix(cls, prefix: str) -> None:
    """Set the storage prefix for the current execution context.

    Chainlit serves multiple chat sessions in one Python process.  A plain
    class-level prefix is shared across concurrent chats, so relative
    storage paths can leak into another user's session.  The context-local
    prefix isolates async tasks while keeping ``S3.prefix`` as a fallback
    for CLI/tests/backwards compatibility.
    """
    normalized = _normalize_storage_prefix(prefix)
    cls._fallback_prefix = normalized
    _SESSION_PREFIX.set(normalized)

current_prefix() classmethod

Return the active context-local prefix, falling back to S3.prefix.

Source code in src/cs_copilot/storage/client.py

@classmethod
def current_prefix(cls) -> str:
    """Return the active context-local prefix, falling back to ``S3.prefix``."""
    return _SESSION_PREFIX.get() or cls._fallback_prefix

scoped_session_prefix(prefix) classmethod

Bind a session prefix for this execution context only.

Unlike :meth:set_session_prefix, the process-wide fallback is left untouched, so concurrent chats in one process cannot leak into each other through it.

Source code in src/cs_copilot/storage/client.py

@classmethod
@contextmanager
def scoped_session_prefix(cls, prefix: str) -> Iterator[None]:
    """Bind a session prefix for this execution context only.

    Unlike :meth:`set_session_prefix`, the process-wide fallback is left
    untouched, so concurrent chats in one process cannot leak into each
    other through it.
    """

    token = _SESSION_PREFIX.set(_normalize_storage_prefix(prefix))
    try:
        yield
    finally:
        _SESSION_PREFIX.reset(token)

observe_writes(boundary, *, protected_paths=()) classmethod

Record, without enforcing, how confined writes would treat each write.

While active and no :meth:confine_writes boundary is set, every write through :meth:open is classified against boundary and the protected paths, then performed exactly as it would be without the observer. Classification errors are recorded, never raised.

Source code in src/cs_copilot/storage/client.py

@classmethod
@contextmanager
def observe_writes(
    cls,
    boundary: str,
    *,
    protected_paths: Iterable[str] = (),
) -> Iterator[WriteObservation]:
    """Record, without enforcing, how confined writes would treat each write.

    While active and no :meth:`confine_writes` boundary is set, every write
    through :meth:`open` is classified against ``boundary`` and the
    protected paths, then performed exactly as it would be without the
    observer. Classification errors are recorded, never raised.
    """

    observation = WriteObservation(
        boundary=cls._normalize_relative_path(boundary, allow_empty=True),
        protected=frozenset(str(path) for path in protected_paths),
    )
    token = _WRITE_OBSERVATION.set(observation)
    try:
        yield observation
    finally:
        _WRITE_OBSERVATION.reset(token)

confine_writes(boundary, *, protected_paths=(), publication_receipt=None, commit_policy='on_exit') classmethod

Confine relative writes to one session subtree for this execution context.

The MCP adapter activates this guard while a session-scoped tool runs. It is context-local, propagates through asyncio.to_thread, and is enforced again by the worker process. Existing registered artifacts can be supplied as protected paths so a tool cannot mutate immutable scientific evidence. Reads remain unaffected.

commit_policy="on_exit" (default) keeps every write private until the scope exits cleanly. "on_close" publishes each file as soon as it is closed, so code that reads its own outputs through other APIs keeps working; the files are still recorded in publication_receipt and rolled back if the scope exits with an exception.

Source code in src/cs_copilot/storage/client.py

@classmethod
@contextmanager
def confine_writes(
    cls,
    boundary: str,
    *,
    protected_paths: Iterable[str] = (),
    publication_receipt: dict[str, dict[str, Any]] | None = None,
    commit_policy: str = "on_exit",
) -> Iterator[None]:
    """Confine relative writes to one session subtree for this execution context.

    The MCP adapter activates this guard while a session-scoped tool runs.
    It is context-local, propagates through ``asyncio.to_thread``, and is
    enforced again by the worker process. Existing registered artifacts
    can be supplied as protected paths so a tool cannot mutate immutable
    scientific evidence. Reads remain unaffected.

    ``commit_policy="on_exit"`` (default) keeps every write private until
    the scope exits cleanly. ``"on_close"`` publishes each file as soon as
    it is closed, so code that reads its own outputs through other APIs
    keeps working; the files are still recorded in ``publication_receipt``
    and rolled back if the scope exits with an exception.
    """

    if commit_policy == "on_close":
        with cls._confine_writes_on_close(
            boundary,
            protected_paths=protected_paths,
            publication_receipt=publication_receipt,
        ):
            yield
        return
    if commit_policy != "on_exit":
        raise ValueError(f"unknown commit policy: {commit_policy!r}")

    normalized = cls._normalize_relative_path(boundary, allow_empty=True)
    boundary_path = PurePosixPath(normalized) if normalized else PurePosixPath()
    protected: set[str] = set()
    for path in protected_paths:
        protected_path = cls._normalize_relative_path(path)
        try:
            PurePosixPath(protected_path).relative_to(boundary_path)
        except ValueError as exc:
            raise ValueError(
                "protected write path must remain inside the active boundary"
            ) from exc
        protected.add(protected_path)
    normalized_protected = frozenset(protected)
    outer_pending = _PENDING_ARTIFACT_WRITES.get()
    pending: list[Any] = outer_pending if outer_pending is not None else []
    owns_transaction = outer_pending is None
    outer_staged = _STAGED_ARTIFACT_WRITES.get()
    staged: dict[str, Any] = outer_staged if outer_staged is not None else {}
    boundary_token = _WRITE_BOUNDARY.set(normalized)
    protected_token = _WRITE_PROTECTED_PATHS.set(normalized_protected)
    pending_token = _PENDING_ARTIFACT_WRITES.set(pending)
    staged_token = _STAGED_ARTIFACT_WRITES.set(staged)
    try:
        yield
    except BaseException:
        if owns_transaction:
            for writer in reversed(pending):
                writer._abort()
        raise
    else:
        if owns_transaction:
            committed: list[Any] = []
            prepared_publications: dict[str, dict[str, Any]] = {}
            try:
                if publication_receipt is not None:
                    for writer in pending:
                        staged_key = getattr(writer, "_staged_key", None)
                        if staged_key is None:
                            continue
                        prepared_publications[staged_key] = cls._pending_writer_metadata(
                            writer, staged_key
                        )
                for writer in pending:
                    writer._commit()
                    committed.append(writer)
            except BaseException:
                cleanup_errors: list[BaseException] = []
                for writer in reversed(pending[len(committed) :]):
                    try:
                        writer._abort()
                    except BaseException as exc:
                        cleanup_errors.append(exc)
                        logger.warning(
                            "Could not abort a pending artifact publication",
                            exc_info=True,
                        )
                for writer in reversed(committed):
                    try:
                        writer._rollback_published()
                    except BaseException as exc:
                        cleanup_errors.append(exc)
                        logger.warning(
                            "Refused or failed to roll back a published artifact",
                            exc_info=True,
                        )
                if cleanup_errors:
                    raise RuntimeError(
                        "artifact transaction failed and could not be fully rolled back"
                    ) from cleanup_errors[0]
                raise
            else:
                if publication_receipt is not None:
                    publication_receipt.update(prepared_publications)
    finally:
        _STAGED_ARTIFACT_WRITES.reset(staged_token)
        _PENDING_ARTIFACT_WRITES.reset(pending_token)
        _WRITE_PROTECTED_PATHS.reset(protected_token)
        _WRITE_BOUNDARY.reset(boundary_token)

exists(rel) classmethod

Return whether a storage path (relative, local, or S3 URL) exists.

Source code in src/cs_copilot/storage/client.py

@classmethod
def exists(cls, rel: str) -> bool:
    """Return whether a storage path (relative, local, or S3 URL) exists."""

    candidate = str(rel).strip()
    if cls._is_explicit_local_path(candidate):
        local = (
            unquote(urlsplit(candidate).path) if candidate.startswith("file://") else candidate
        )
        return Path(local).exists()
    if not cls._is_s3_url(candidate) and not is_s3_enabled():
        return cls._local_session_path(candidate).exists()
    url = candidate if cls._is_s3_url(candidate) else cls.path(candidate)
    filesystem, filesystem_path = fsspec.core.url_to_fs(
        url, **get_s3_config().to_storage_options()
    )
    filesystem.invalidate_cache(filesystem_path.rsplit("/", 1)[0])
    return bool(filesystem.exists(filesystem_path))

first_free_path(rel) classmethod

Return rel, or its first free <stem>-v<N><suffix> sibling.

Workflow outputs are immutable once registered and confined writes are create-only, so re-running a step writes a new version of each output instead of overwriting the previous one. The same spelling (relative, local, or S3 URL) is returned.

Source code in src/cs_copilot/storage/client.py

@classmethod
def first_free_path(cls, rel: str) -> str:
    """Return ``rel``, or its first free ``<stem>-v<N><suffix>`` sibling.

    Workflow outputs are immutable once registered and confined writes are
    create-only, so re-running a step writes a new version of each output
    instead of overwriting the previous one. The same spelling (relative,
    local, or S3 URL) is returned.
    """

    if not cls.exists(rel):
        return rel
    version = 2
    while cls.exists(_versioned_name(rel, version)):
        version += 1
    return _versioned_name(rel, version)

latest_version_path(rel) classmethod

Return the newest existing version of rel written by :meth:first_free_path.

Readers that locate an output by its deterministic name use this to find the most recent re-run. rel itself is returned when it has no later version (or does not exist).

Source code in src/cs_copilot/storage/client.py

@classmethod
def latest_version_path(cls, rel: str) -> str:
    """Return the newest existing version of ``rel`` written by :meth:`first_free_path`.

    Readers that locate an output by its deterministic name use this to
    find the most recent re-run. ``rel`` itself is returned when it has no
    later version (or does not exist).
    """

    latest = rel
    if not cls.exists(rel):
        return rel
    version = 2
    while cls.exists(_versioned_name(rel, version)):
        latest = _versioned_name(rel, version)
        version += 1
    return latest

session_relative(path) classmethod

Return the session-relative key of a path in the active session, if any.

Source code in src/cs_copilot/storage/client.py

@classmethod
def session_relative(cls, path: str) -> str | None:
    """Return the session-relative key of a path in the active session, if any."""

    return cls._verified_read_key(path)

current_write_boundary() staticmethod

Return the active write boundary, or None outside guarded execution.

Source code in src/cs_copilot/storage/client.py

@staticmethod
def current_write_boundary() -> str | None:
    """Return the active write boundary, or ``None`` outside guarded execution."""

    return _WRITE_BOUNDARY.get()

current_write_protected_paths() staticmethod

Return immutable session-relative paths for the active tool call.

Source code in src/cs_copilot/storage/client.py

@staticmethod
def current_write_protected_paths() -> frozenset[str]:
    """Return immutable session-relative paths for the active tool call."""

    return _WRITE_PROTECTED_PATHS.get()

stage_publications(staging_id) classmethod

Redirect scoped writes to a worker-private publication namespace.

Source code in src/cs_copilot/storage/client.py

@classmethod
@contextmanager
def stage_publications(cls, staging_id: str) -> Iterator[None]:
    """Redirect scoped writes to a worker-private publication namespace."""

    normalized_id = cls._normalize_relative_path(staging_id)
    if len(PurePosixPath(normalized_id).parts) != 1:
        raise ValueError("staging_id must be one safe path component")
    publications: dict[str, str] = {}
    id_token = _EXTERNAL_STAGING_ID.set(normalized_id)
    publications_token = _EXTERNAL_STAGED_PUBLICATIONS.set(publications)
    try:
        yield
    finally:
        _EXTERNAL_STAGED_PUBLICATIONS.reset(publications_token)
        _EXTERNAL_STAGING_ID.reset(id_token)

current_staged_publications() staticmethod

Return final-to-staged keys created by the active worker invocation.

Source code in src/cs_copilot/storage/client.py

@staticmethod
def current_staged_publications() -> dict[str, str]:
    """Return final-to-staged keys created by the active worker invocation."""

    return dict(_EXTERNAL_STAGED_PUBLICATIONS.get() or {})

staged_publication_metadata() classmethod

Checksum committed worker staging objects through a no-follow read.

Source code in src/cs_copilot/storage/client.py

@classmethod
def staged_publication_metadata(cls) -> dict[str, dict[str, Any]]:
    """Checksum committed worker staging objects through a no-follow read."""

    metadata: dict[str, dict[str, Any]] = {}
    for final_key, staged_key in cls.current_staged_publications().items():
        digest = hashlib.sha256()
        size = 0
        if is_s3_enabled():
            config = get_s3_config()
            source = fsspec.open(
                cls.path(staged_key),
                mode="rb",
                **config.to_storage_options(),
            ).open()
        else:
            source = cls._open_local_verified_source(staged_key)
        with source:
            while True:
                chunk = source.read(1024 * 1024)
                if not chunk:
                    break
                if isinstance(chunk, str):
                    chunk = chunk.encode("utf-8")
                digest.update(chunk)
                size += len(chunk)
        metadata[final_key] = {
            "staged_path": staged_key,
            "sha256": digest.hexdigest(),
            "size_bytes": size,
        }
    return metadata

promote_staged_publications(publications) classmethod

Promote complete worker outputs to final create-only artifact keys.

Source code in src/cs_copilot/storage/client.py

@classmethod
def promote_staged_publications(
    cls,
    publications: Mapping[str, Mapping[str, Any]],
) -> None:
    """Promote complete worker outputs to final create-only artifact keys."""

    promoted: dict[str, Mapping[str, Any]] = {}
    try:
        for final_key, metadata in publications.items():
            final = cls._normalize_relative_path(final_key)
            staged = cls._normalize_relative_path(str(metadata["staged_path"]))
            source = cls._open_verified_snapshot(
                staged,
                "rb",
                expected_sha256=str(metadata["sha256"]),
                expected_size=int(metadata["size_bytes"]),
            )
            with source:
                with cls.open_atomic(final, "xb") as destination:
                    while True:
                        chunk = source.read(1024 * 1024)
                        if not chunk:
                            break
                        destination.write(chunk)
            promoted[final] = metadata
        for metadata in publications.values():
            cls._remove_storage_key(cls._normalize_relative_path(str(metadata["staged_path"])))
    except BaseException:
        cls.rollback_promoted_publications(promoted)
        raise

rollback_promoted_publications(publications) classmethod

Remove only finals that still match the worker's promoted bytes.

Source code in src/cs_copilot/storage/client.py

@classmethod
def rollback_promoted_publications(
    cls,
    publications: Mapping[str, Mapping[str, Any]],
) -> None:
    """Remove only finals that still match the worker's promoted bytes."""

    for final_key, metadata in reversed(tuple(publications.items())):
        final = cls._normalize_relative_path(final_key)
        try:
            cls._remove_storage_key_verified(
                final,
                expected_sha256=str(metadata["sha256"]),
                expected_size=int(metadata["size_bytes"]),
            )
        except FileNotFoundError:
            continue
        except Exception:
            logger.warning(
                "Refusing to roll back changed worker publication %s",
                final,
                exc_info=True,
            )
            continue

discard_staging_prefix(boundary, staging_id) classmethod

Remove one exact worker staging subtree after failure or cancellation.

Source code in src/cs_copilot/storage/client.py

@classmethod
def discard_staging_prefix(cls, boundary: str, staging_id: str) -> None:
    """Remove one exact worker staging subtree after failure or cancellation."""

    normalized_boundary = cls._normalize_relative_path(boundary)
    normalized_id = cls._normalize_relative_path(staging_id)
    if len(PurePosixPath(normalized_id).parts) != 1:
        raise ValueError("staging_id must be one safe path component")
    prefix = PurePosixPath(
        normalized_boundary,
        ".staging",
        normalized_id,
    ).as_posix()
    if is_s3_enabled():
        config = get_s3_config()
        fs, fs_path = fsspec.core.url_to_fs(
            cls.path(prefix),
            **config.to_storage_options(),
        )
        if fs.exists(fs_path):
            fs.rm(fs_path, recursive=True)
        return
    _secure_local_rmtree(cls._local_session_path(prefix).absolute())

confine_artifact_reads(artifacts) classmethod

Pin registered artifact reads to checksum-verified byte snapshots.

Source code in src/cs_copilot/storage/client.py

@classmethod
@contextmanager
def confine_artifact_reads(
    cls,
    artifacts: Mapping[str, tuple[str, int]],
) -> Iterator[None]:
    """Pin registered artifact reads to checksum-verified byte snapshots."""

    inherited = dict(_VERIFIED_ARTIFACT_READS.get() or {})
    for path, expected in artifacts.items():
        normalized = cls._normalize_relative_path(path)
        if (
            not isinstance(expected, tuple)
            or len(expected) != 2
            or not isinstance(expected[0], str)
            or not isinstance(expected[1], int)
            or expected[1] < 0
        ):
            raise ValueError("verified artifact metadata must be (sha256, size_bytes)")
        inherited[normalized] = (expected[0], expected[1])
    token = _VERIFIED_ARTIFACT_READS.set(inherited)
    try:
        yield
    finally:
        _VERIFIED_ARTIFACT_READS.reset(token)

current_verified_artifact_reads() staticmethod

Return a detached copy of invocation-pinned artifact read metadata.

Source code in src/cs_copilot/storage/client.py

@staticmethod
def current_verified_artifact_reads() -> dict[str, tuple[str, int]]:
    """Return a detached copy of invocation-pinned artifact read metadata."""

    return dict(_VERIFIED_ARTIFACT_READS.get() or {})

path(rel) classmethod

Convert a relative path to an S3 URL or local session path.

Explicit S3 and local paths are returned unchanged. Relative paths are resolved against the active backend.

Parameters:

Name	Type	Description	Default
rel	str	Relative path or absolute S3 URL	required

Returns:

Name	Type	Description
str	str	Full S3 URL or local path

Examples:

S3.path("data.csv") 's3://chatbot-assets/sessions/20250121-123456-abc123/data.csv'

S3.path("s3://mybucket/data.csv") 's3://mybucket/data.csv'

Source code in src/cs_copilot/storage/client.py

@classmethod
def path(cls, rel: str) -> str:
    """
    Convert a relative path to an S3 URL or local session path.

    Explicit S3 and local paths are returned unchanged. Relative paths are
    resolved against the active backend.

    Args:
        rel: Relative path or absolute S3 URL

    Returns:
        str: Full S3 URL or local path

    Examples:
    ---------
    >>> S3.path("data.csv")
    's3://chatbot-assets/sessions/20250121-123456-abc123/data.csv'

    >>> S3.path("s3://mybucket/data.csv")
    's3://mybucket/data.csv'
    """
    if cls._is_s3_url(rel) or cls._is_explicit_local_path(rel):
        return rel

    config = get_s3_config()
    if not is_s3_enabled():
        return os.fspath(cls._local_session_path(rel))

    key = PurePosixPath(cls.current_prefix().strip("/")) / str(rel).lstrip("/")
    return f"s3://{config.bucket_name}/{key.as_posix()}"

open(rel, mode='rb') classmethod

Open a file for reading or writing.

Supports three types of paths: 1. Absolute S3 URLs (s3://...) 2. Local absolute paths (/ or file://) 3. Relative paths (scoped to current session)

Parameters:

Name	Type	Description	Default
rel	str	File path (relative, absolute, or S3 URL)	required
mode	str	File mode ('r', 'w', 'rb', 'wb', etc.)	'rb'

Returns:

Type	Description
Any	File-like object opened with the specified mode

Examples:

Read from session-scoped S3 path

with S3.open("data.csv", "r") as f: ... df = pd.read_csv(f)

Write to session-scoped S3 path

with S3.open("output.csv", "w") as f: ... df.to_csv(f)

Read from absolute S3 URL

with S3.open("s3://bucket/data.csv", "r") as f: ... df = pd.read_csv(f)

Read from local file

with S3.open("/tmp/data.csv", "r") as f: ... df = pd.read_csv(f)

Source code in src/cs_copilot/storage/client.py

@classmethod
def open(cls, rel: str, mode: str = "rb") -> Any:
    """
    Open a file for reading or writing.

    Supports three types of paths:
    1. Absolute S3 URLs (s3://...)
    2. Local absolute paths (/ or file://)
    3. Relative paths (scoped to current session)

    Args:
        rel: File path (relative, absolute, or S3 URL)
        mode: File mode ('r', 'w', 'rb', 'wb', etc.)

    Returns:
        File-like object opened with the specified mode

    Examples:
    ---------
    >>> # Read from session-scoped S3 path
    >>> with S3.open("data.csv", "r") as f:
    ...     df = pd.read_csv(f)

    >>> # Write to session-scoped S3 path
    >>> with S3.open("output.csv", "w") as f:
    ...     df.to_csv(f)

    >>> # Read from absolute S3 URL
    >>> with S3.open("s3://bucket/data.csv", "r") as f:
    ...     df = pd.read_csv(f)

    >>> # Read from local file
    >>> with S3.open("/tmp/data.csv", "r") as f:
    ...     df = pd.read_csv(f)
    """
    config = get_s3_config()
    write_mode = cls._is_write_mode(mode)
    if not write_mode:
        read_key = cls._verified_read_key(rel)
        staged = (
            _STAGED_ARTIFACT_WRITES.get().get(read_key)
            if read_key is not None and _STAGED_ARTIFACT_WRITES.get() is not None
            else None
        )
        if staged is not None:
            return staged._open_staged_read(mode)
        expected_reads = _VERIFIED_ARTIFACT_READS.get() or {}
        expected = expected_reads.get(read_key) if read_key else None
        if read_key is not None and expected is not None:
            return cls._open_verified_snapshot(
                read_key,
                mode,
                expected_sha256=expected[0],
                expected_size=expected[1],
            )

    write_boundary = cls.current_write_boundary() if write_mode else None
    if write_mode and write_boundary is None:
        observation = _WRITE_OBSERVATION.get()
        if observation is not None:
            cls._observe_write(observation, rel, mode)
    scoped_write_key: str | None = None
    if write_boundary is not None:
        scoped_write_key = cls._validate_scoped_write(rel, boundary=write_boundary)
    storage_target = rel
    staging_id = _EXTERNAL_STAGING_ID.get()
    publications = _EXTERNAL_STAGED_PUBLICATIONS.get()
    if scoped_write_key is not None and staging_id is not None and publications is not None:
        relative_to_boundary = PurePosixPath(scoped_write_key).relative_to(
            PurePosixPath(write_boundary or "")
        )
        staged_key = PurePosixPath(
            write_boundary or "",
            ".staging",
            staging_id,
            relative_to_boundary,
        ).as_posix()
        publications[scoped_write_key] = staged_key
        storage_target = staged_key

    if cls._is_s3_url(storage_target):
        if scoped_write_key is None:
            return fsspec.open(
                storage_target,
                mode=mode,
                **config.to_storage_options(),
            )
        return cls._open_confined_s3_path(
            storage_target,
            mode,
            storage_options=config.to_storage_options(),
            public_write_key=scoped_write_key,
        )

    if cls._is_explicit_local_path(storage_target):
        if storage_target.startswith("file://"):
            return fsspec.open(storage_target, mode=mode)
        return cls._open_local_path(Path(storage_target), mode)

    if not is_s3_enabled():
        return cls._open_local_path(
            cls._local_session_path(storage_target),
            mode,
            scoped_write_key=scoped_write_key,
        )

    if scoped_write_key is None:
        return fsspec.open(
            cls.path(storage_target),
            mode=mode,
            **config.to_storage_options(),
        )
    return cls._open_confined_s3_path(
        cls.path(storage_target),
        mode,
        storage_options=config.to_storage_options(),
        public_write_key=scoped_write_key,
    )

open_atomic(rel, mode='xb') classmethod

Create one session-relative object that is invisible until complete.

Source code in src/cs_copilot/storage/client.py

@classmethod
def open_atomic(cls, rel: str, mode: str = "xb"):
    """Create one session-relative object that is invisible until complete."""

    if not cls._is_write_mode(mode):
        raise ValueError("open_atomic requires a write-capable mode")
    if cls._is_s3_url(rel) or cls._is_explicit_local_path(rel):
        raise ValueError("open_atomic requires a session-relative storage path")
    normalized = cls._normalize_relative_path(rel)
    config = get_s3_config()
    if not is_s3_enabled():
        return cls._open_confined_local_path(
            cls._local_session_path(normalized),
            mode,
            boundary="",
            scoped_write_key=normalized,
            defer_transaction=False,
        )
    return cls._open_confined_s3_path(
        cls.path(normalized),
        mode,
        storage_options=config.to_storage_options(),
        defer_transaction=False,
        public_write_key=None,
    )

ScopedWriteDenied

Bases: PermissionError

A write refused by :meth:S3.confine_writes, with a stable verdict.

Source code in src/cs_copilot/storage/client.py

class ScopedWriteDenied(PermissionError):
    """A write refused by :meth:`S3.confine_writes`, with a stable ``verdict``."""

    def __init__(self, message: str, *, verdict: str) -> None:
        super().__init__(message)
        self.verdict = verdict

WriteObservation dataclass

Writes attempted during one observed call, classified like confined writes.

Each record is {"path", "mode", "verdict", "created"}. verdict is how :meth:S3.confine_writes would have treated the write: create_new, overwrite_existing, append_existing, protected_artifact, runtime_metadata, outside_boundary, absolute_path, or unsafe_path. path is where the bytes actually land (session-relative when inside the session) and created tells whether the write created a new file inside the boundary. Observation never changes what the write does.

Source code in src/cs_copilot/storage/client.py

@dataclass
class WriteObservation:
    """Writes attempted during one observed call, classified like confined writes.

    Each record is ``{"path", "mode", "verdict", "created"}``. ``verdict`` is
    how :meth:`S3.confine_writes` would have treated the write:
    ``create_new``, ``overwrite_existing``, ``append_existing``,
    ``protected_artifact``, ``runtime_metadata``, ``outside_boundary``,
    ``absolute_path``, or ``unsafe_path``. ``path`` is where the bytes actually
    land (session-relative when inside the session) and ``created`` tells
    whether the write created a new file inside the boundary. Observation
    never changes what the write does.
    """

    boundary: str
    protected: frozenset[str] = frozenset()
    records: list[dict[str, Any]] = field(default_factory=list)
    _lock: threading.Lock = field(default_factory=threading.Lock, repr=False)

    def record(self, entry: dict[str, Any]) -> None:
        with self._lock:
            self.records.append(entry)

    def snapshot(self) -> list[dict[str, Any]]:
        with self._lock:
            return [dict(entry) for entry in self.records]

    def created_paths(self) -> list[str]:
        """Session-relative paths of new files written inside the boundary, in order."""

        return list(
            dict.fromkeys(entry["path"] for entry in self.snapshot() if entry.get("created"))
        )

created_paths()

Session-relative paths of new files written inside the boundary, in order.

Source code in src/cs_copilot/storage/client.py

def created_paths(self) -> list[str]:
    """Session-relative paths of new files written inside the boundary, in order."""

    return list(
        dict.fromkeys(entry["path"] for entry in self.snapshot() if entry.get("created"))
    )

S3Config dataclass

Configuration for S3/MinIO storage.

Source code in src/cs_copilot/storage/config.py

@dataclass
class S3Config:
    """Configuration for S3/MinIO storage."""

    endpoint_url: Optional[str]
    access_key_id: Optional[str]
    secret_access_key: Optional[str]
    bucket_name: Optional[str]
    region_name: str = "us-east-1"
    use_s3: bool = False

    @classmethod
    def from_env(cls) -> "S3Config":
        """
        Create S3Config from environment variables.

        Environment Variables:
        ----------------------
        Endpoint (one of):
            - MINIO_ENDPOINT
            - MINIO_ENDPOINT_URL
            - S3_ENDPOINT_URL

        Access Key (one of):
            - MINIO_ACCESS_KEY
            - AWS_ACCESS_KEY_ID

        Secret Key (one of):
            - MINIO_SECRET_KEY
            - AWS_SECRET_ACCESS_KEY

        Bucket Name (one of):
            - ASSETS_BUCKET
            - S3_BUCKET_NAME

        Other:
            - AWS_REGION (default: us-east-1)
            - USE_S3 (default: false)

        Returns:
            S3Config: Configuration instance
        """
        endpoint = _getenv_nonempty("MINIO_ENDPOINT", "MINIO_ENDPOINT_URL", "S3_ENDPOINT_URL")

        minio_access_key = _getenv_nonempty("MINIO_ACCESS_KEY")
        minio_secret_key = _getenv_nonempty("MINIO_SECRET_KEY")
        aws_access_key = _getenv_nonempty("AWS_ACCESS_KEY_ID")
        aws_secret_key = _getenv_nonempty("AWS_SECRET_ACCESS_KEY")

        if endpoint:
            access_key = minio_access_key or aws_access_key
            secret_key = minio_secret_key or aws_secret_key
        else:
            access_key = aws_access_key
            secret_key = aws_secret_key

        bucket_name = _getenv_nonempty("ASSETS_BUCKET", "S3_BUCKET_NAME")
        region_name = os.getenv("AWS_REGION", "us-east-1")
        use_s3 = os.getenv("USE_S3", "false").lower() == "true"

        return cls(
            endpoint_url=endpoint,
            access_key_id=access_key,
            secret_access_key=secret_key,
            bucket_name=bucket_name,
            region_name=region_name,
            use_s3=use_s3,
        )

    def storage_backend(self) -> str:
        """
        Resolve the active storage backend.

        Returns:
            str: "local", "aws", or "s3-compatible"

        Raises:
            StorageConfigError: If S3 is explicitly enabled but missing required settings.
        """
        if not self.use_s3:
            return "local"

        missing_bucket = not self.bucket_name
        missing_key = not self.access_key_id
        missing_secret = not self.secret_access_key

        if self.endpoint_url:
            missing = []
            if missing_bucket:
                missing.append("ASSETS_BUCKET or S3_BUCKET_NAME")
            if missing_key:
                missing.append("MINIO_ACCESS_KEY or AWS_ACCESS_KEY_ID")
            if missing_secret:
                missing.append("MINIO_SECRET_KEY or AWS_SECRET_ACCESS_KEY")
            if missing:
                raise StorageConfigError(
                    "USE_S3=true with an S3-compatible endpoint requires "
                    + ", ".join(missing)
                    + "."
                )
            return "s3-compatible"

        missing = []
        if missing_bucket:
            missing.append("ASSETS_BUCKET or S3_BUCKET_NAME")
        if missing_key:
            missing.append("AWS_ACCESS_KEY_ID")
        if missing_secret:
            missing.append("AWS_SECRET_ACCESS_KEY")
        if missing:
            raise StorageConfigError(
                "USE_S3=true without an endpoint requires " + ", ".join(missing) + " for AWS S3."
            )
        return "aws"

    def to_storage_options(self) -> dict:
        """
        Convert config to fsspec/s3fs storage options.

        Returns:
            dict: Storage options for fsspec.open()
        """
        opts = {"config_kwargs": {"s3": {"addressing_style": "path"}}}

        if self.access_key_id:
            opts["key"] = self.access_key_id

        if self.secret_access_key:
            opts["secret"] = self.secret_access_key

        if self.endpoint_url:
            opts["client_kwargs"] = {"endpoint_url": self.endpoint_url}

        return opts

from_env() classmethod

Create S3Config from environment variables.

Environment Variables:

Endpoint (one of): - MINIO_ENDPOINT - MINIO_ENDPOINT_URL - S3_ENDPOINT_URL

Access Key (one of): - MINIO_ACCESS_KEY - AWS_ACCESS_KEY_ID

Secret Key (one of): - MINIO_SECRET_KEY - AWS_SECRET_ACCESS_KEY

Bucket Name (one of): - ASSETS_BUCKET - S3_BUCKET_NAME

Other

• AWS_REGION (default: us-east-1)
• USE_S3 (default: false)

Returns:

Name	Type	Description
S3Config	S3Config	Configuration instance

Source code in src/cs_copilot/storage/config.py

@classmethod
def from_env(cls) -> "S3Config":
    """
    Create S3Config from environment variables.

    Environment Variables:
    ----------------------
    Endpoint (one of):
        - MINIO_ENDPOINT
        - MINIO_ENDPOINT_URL
        - S3_ENDPOINT_URL

    Access Key (one of):
        - MINIO_ACCESS_KEY
        - AWS_ACCESS_KEY_ID

    Secret Key (one of):
        - MINIO_SECRET_KEY
        - AWS_SECRET_ACCESS_KEY

    Bucket Name (one of):
        - ASSETS_BUCKET
        - S3_BUCKET_NAME

    Other:
        - AWS_REGION (default: us-east-1)
        - USE_S3 (default: false)

    Returns:
        S3Config: Configuration instance
    """
    endpoint = _getenv_nonempty("MINIO_ENDPOINT", "MINIO_ENDPOINT_URL", "S3_ENDPOINT_URL")

    minio_access_key = _getenv_nonempty("MINIO_ACCESS_KEY")
    minio_secret_key = _getenv_nonempty("MINIO_SECRET_KEY")
    aws_access_key = _getenv_nonempty("AWS_ACCESS_KEY_ID")
    aws_secret_key = _getenv_nonempty("AWS_SECRET_ACCESS_KEY")

    if endpoint:
        access_key = minio_access_key or aws_access_key
        secret_key = minio_secret_key or aws_secret_key
    else:
        access_key = aws_access_key
        secret_key = aws_secret_key

    bucket_name = _getenv_nonempty("ASSETS_BUCKET", "S3_BUCKET_NAME")
    region_name = os.getenv("AWS_REGION", "us-east-1")
    use_s3 = os.getenv("USE_S3", "false").lower() == "true"

    return cls(
        endpoint_url=endpoint,
        access_key_id=access_key,
        secret_access_key=secret_key,
        bucket_name=bucket_name,
        region_name=region_name,
        use_s3=use_s3,
    )

storage_backend()

Resolve the active storage backend.

Returns:

Name	Type	Description
str	str	"local", "aws", or "s3-compatible"

Raises:

Type	Description
StorageConfigError	If S3 is explicitly enabled but missing required settings.

Source code in src/cs_copilot/storage/config.py

def storage_backend(self) -> str:
    """
    Resolve the active storage backend.

    Returns:
        str: "local", "aws", or "s3-compatible"

    Raises:
        StorageConfigError: If S3 is explicitly enabled but missing required settings.
    """
    if not self.use_s3:
        return "local"

    missing_bucket = not self.bucket_name
    missing_key = not self.access_key_id
    missing_secret = not self.secret_access_key

    if self.endpoint_url:
        missing = []
        if missing_bucket:
            missing.append("ASSETS_BUCKET or S3_BUCKET_NAME")
        if missing_key:
            missing.append("MINIO_ACCESS_KEY or AWS_ACCESS_KEY_ID")
        if missing_secret:
            missing.append("MINIO_SECRET_KEY or AWS_SECRET_ACCESS_KEY")
        if missing:
            raise StorageConfigError(
                "USE_S3=true with an S3-compatible endpoint requires "
                + ", ".join(missing)
                + "."
            )
        return "s3-compatible"

    missing = []
    if missing_bucket:
        missing.append("ASSETS_BUCKET or S3_BUCKET_NAME")
    if missing_key:
        missing.append("AWS_ACCESS_KEY_ID")
    if missing_secret:
        missing.append("AWS_SECRET_ACCESS_KEY")
    if missing:
        raise StorageConfigError(
            "USE_S3=true without an endpoint requires " + ", ".join(missing) + " for AWS S3."
        )
    return "aws"

to_storage_options()

Convert config to fsspec/s3fs storage options.

Returns:

Name	Type	Description
dict	dict	Storage options for fsspec.open()

Source code in src/cs_copilot/storage/config.py

def to_storage_options(self) -> dict:
    """
    Convert config to fsspec/s3fs storage options.

    Returns:
        dict: Storage options for fsspec.open()
    """
    opts = {"config_kwargs": {"s3": {"addressing_style": "path"}}}

    if self.access_key_id:
        opts["key"] = self.access_key_id

    if self.secret_access_key:
        opts["secret"] = self.secret_access_key

    if self.endpoint_url:
        opts["client_kwargs"] = {"endpoint_url": self.endpoint_url}

    return opts

StorageConfigError

Bases: ValueError

Raised when S3 is explicitly enabled but the runtime config is incomplete.

Source code in src/cs_copilot/storage/config.py

class StorageConfigError(ValueError):
    """Raised when S3 is explicitly enabled but the runtime config is incomplete."""

OutputLayout dataclass

Structured output path builder for one workflow run.

Source code in src/cs_copilot/storage/layout.py

@dataclass(frozen=True)
class OutputLayout:
    """Structured output path builder for one workflow run."""

    session_id: str
    run_id: str
    workflow_slug: str

    def __post_init__(self) -> None:
        object.__setattr__(
            self,
            "session_id",
            validate_identifier(self.session_id, field="session_id"),
        )
        object.__setattr__(
            self,
            "run_id",
            validate_identifier(self.run_id, field="run_id"),
        )
        object.__setattr__(
            self,
            "workflow_slug",
            sanitize_workflow_slug(self.workflow_slug),
        )

    @property
    def run_root(self) -> str:
        return PurePosixPath(WORKFLOWS_DIR, self.run_id).as_posix()

    def rel_path(self, operation: OutputOperation, *parts: str) -> str:
        cleaned_parts = [sanitize_path_part(part) for part in parts if str(part).strip()]
        return PurePosixPath(self.run_root, operation.value, *cleaned_parts).as_posix()

    @property
    def manifest_rel_path(self) -> str:
        return PurePosixPath(self.run_root, "manifest.json").as_posix()

    @property
    def artifact_index_rel_path(self) -> str:
        return PurePosixPath(self.run_root, "artifacts", "index.json").as_posix()

    @property
    def events_rel_path(self) -> str:
        return PurePosixPath(self.run_root, "events").as_posix()

    def event_rel_path(self, event_id: str) -> str:
        return PurePosixPath(
            self.events_rel_path,
            f"{validate_identifier(event_id, field='event_id')}.jsonl",
        ).as_posix()

    def artifact_rel_path(self, path: str) -> str:
        """Resolve and validate a path relative to this run's root."""

        relative = normalize_run_relative_path(self.run_id, path)
        return PurePosixPath(self.run_root, relative).as_posix()

artifact_rel_path(path)

Resolve and validate a path relative to this run's root.

Source code in src/cs_copilot/storage/layout.py

def artifact_rel_path(self, path: str) -> str:
    """Resolve and validate a path relative to this run's root."""

    relative = normalize_run_relative_path(self.run_id, path)
    return PurePosixPath(self.run_root, relative).as_posix()

OutputOperation

Bases: str, Enum

Top-level operation folders inside a workflow run.

Source code in src/cs_copilot/storage/layout.py

class OutputOperation(str, Enum):
    """Top-level operation folders inside a workflow run."""

    CHEMICAL_SPACE = "01_chemical_space"
    ANALOG_GENERATION = "02_analog_generation"
    RETROSYNTHESIS = "03_retrosynthesis"
    REPORTS = "reports"

get_s3_config()

Get S3 configuration from environment variables.

This is evaluated at call time (not import time) so that notebooks can call load_dotenv() after importing modules and still have up-to-date credentials and endpoint settings.

Returns:

Name	Type	Description
S3Config	S3Config	Current S3 configuration

Source code in src/cs_copilot/storage/config.py

def get_s3_config() -> S3Config:
    """
    Get S3 configuration from environment variables.

    This is evaluated at call time (not import time) so that notebooks can
    call load_dotenv() after importing modules and still have up-to-date
    credentials and endpoint settings.

    Returns:
        S3Config: Current S3 configuration
    """
    return S3Config.from_env()

is_s3_enabled()

Check if S3 is enabled based on configuration.

Returns:

Name	Type	Description
bool	bool	True if S3 should be used

Source code in src/cs_copilot/storage/config.py

def is_s3_enabled() -> bool:
    """
    Check if S3 is enabled based on configuration.

    Returns:
        bool: True if S3 should be used
    """
    return get_s3_config().storage_backend() != "local"

active_output_context(context)

Pin one run context for the enclosing scope, restoring the previous one.

Source code in src/cs_copilot/storage/layout.py

@contextmanager
def active_output_context(context: dict[str, Any]) -> Iterator[dict[str, Any]]:
    """Pin one run context for the enclosing scope, restoring the previous one."""

    token = _RUN_CONTEXT.set(None)
    try:
        yield set_output_context(context)
    finally:
        _RUN_CONTEXT.reset(token)

ensure_output_context(session_state=None, *, workflow_slug=None, run_id=None, session_id=None)

Ensure and return explicit session/run/workflow output context.

Source code in src/cs_copilot/storage/layout.py

def ensure_output_context(
    session_state: Optional[dict[str, Any]] = None,
    *,
    workflow_slug: Optional[str] = None,
    run_id: Optional[str] = None,
    session_id: Optional[str] = None,
) -> dict[str, Any]:
    """Ensure and return explicit session/run/workflow output context."""

    resolved_session_id = validate_identifier(session_id or _session_id(), field="session_id")
    requested_slug = sanitize_workflow_slug(workflow_slug)

    candidates: list[dict[str, Any]] = []
    if isinstance(session_state, dict):
        existing = session_state.get(OUTPUT_CONTEXT_KEY)
        if isinstance(existing, dict):
            candidates.append(existing)
    contextual = _RUN_CONTEXT.get()
    if isinstance(contextual, dict):
        candidates.append(contextual)

    for existing in candidates:
        if not _valid_context(existing, session_id=resolved_session_id):
            continue
        if run_id is not None and existing["run_id"] != run_id:
            continue
        # Legacy toolkit helpers pass operation labels such as
        # ``chemical_space`` and ``reports`` here. Once a run is active those
        # labels must not fork the experiment; only an explicit run_id selects
        # a different run.
        context = dict(existing)
        _RUN_CONTEXT.set(context)
        if isinstance(session_state, dict):
            session_state[OUTPUT_CONTEXT_KEY] = context
        return context

    if run_id is None and S3.current_write_boundary() is not None:
        # Minting a run here would place every artifact under a brand-new run
        # root, which can never sit inside the boundary the caller opened.
        raise ScopedWriteDenied(
            "no workflow run is reachable for this scoped write; "
            "the caller must bind the run that owns the active write boundary",
            verdict="outside_boundary",
        )
    resolved_run_id = (
        validate_identifier(run_id, field="run_id")
        if run_id is not None
        else f"{requested_slug}-{uuid.uuid4().hex[:12]}"
    )
    context = {
        "layout_version": LAYOUT_VERSION,
        "session_id": resolved_session_id,
        "run_id": resolved_run_id,
        "workflow_slug": requested_slug,
    }
    _RUN_CONTEXT.set(context)
    if isinstance(session_state, dict):
        session_state[OUTPUT_CONTEXT_KEY] = context
    return context

normalize_run_relative_path(run_id, path)

Return a safe path relative to workflows/<run_id>.

Absolute paths, storage URLs, traversal components, paths belonging to another run, and runtime-control files are rejected. Artifact registration uses this function as its workflow-root containment boundary.

Source code in src/cs_copilot/storage/layout.py

def normalize_run_relative_path(run_id: str, path: str) -> str:
    """Return a safe path relative to ``workflows/<run_id>``.

    Absolute paths, storage URLs, traversal components, paths belonging to
    another run, and runtime-control files are rejected. Artifact registration
    uses this function as its workflow-root containment boundary.
    """

    run_id = validate_identifier(run_id, field="run_id")
    if not isinstance(path, str) or not path.strip():
        raise ValueError("artifact path cannot be empty")
    value = path.strip()
    if is_explicit_storage_path(value) or "\\" in value:
        raise ValueError("artifact path must be relative to the workflow run")

    candidate = PurePosixPath(value)
    if candidate.is_absolute() or any(part in {"", ".", ".."} for part in candidate.parts):
        raise ValueError("artifact path must not be absolute or contain traversal components")

    run_prefix = PurePosixPath(WORKFLOWS_DIR, run_id)
    if candidate.parts and candidate.parts[0] == WORKFLOWS_DIR:
        try:
            candidate = candidate.relative_to(run_prefix)
        except ValueError as exc:
            raise ValueError(f"artifact path is outside workflow run {run_id!r}") from exc

    if not candidate.parts:
        raise ValueError("artifact path must identify a file")
    if candidate == PurePosixPath("manifest.json"):
        raise ValueError("manifest.json is reserved for the workflow runtime")
    if candidate.parts[0] in {".staging", "events"}:
        raise ValueError(f"{candidate.parts[0]}/ is reserved for the workflow runtime")
    if candidate == PurePosixPath("artifacts", "index.json"):
        raise ValueError("artifacts/index.json is reserved for the workflow runtime")
    return candidate.as_posix()

open_local_run_artifact(layout, relative_path)

Open a local artifact for reading without following symlinks.

The workflow root and every descendant component are opened by descriptor. Each child lookup is relative to its already-open parent, so replacing a directory between validation and the final open cannot redirect the read. This helper intentionally fails closed on platforms without openat and O_NOFOLLOW support.

Source code in src/cs_copilot/storage/layout.py

def open_local_run_artifact(layout: OutputLayout, relative_path: str) -> BinaryIO:
    """Open a local artifact for reading without following symlinks.

    The workflow root and every descendant component are opened by descriptor.
    Each child lookup is relative to its already-open parent, so replacing a
    directory between validation and the final open cannot redirect the read.
    This helper intentionally fails closed on platforms without ``openat`` and
    ``O_NOFOLLOW`` support.
    """

    relative = normalize_run_relative_path(layout.run_id, relative_path)
    if not _OPEN_SUPPORTS_DIR_FD or not _O_NOFOLLOW:
        raise RuntimeError(
            "secure local artifact reads require directory-relative no-follow support"
        )

    common_flags = os.O_RDONLY | getattr(os, "O_CLOEXEC", 0) | _O_NOFOLLOW
    directory_flags = common_flags | getattr(os, "O_DIRECTORY", 0)
    current_fd: int | None = None
    file_fd: int | None = None
    try:
        current_fd = _open_existing_directory_tree(Path(S3.path(layout.run_root)).absolute())
        parts = PurePosixPath(relative).parts
        for component in parts[:-1]:
            next_fd = os.open(component, directory_flags, dir_fd=current_fd)
            os.close(current_fd)
            current_fd = next_fd
        file_fd = os.open(parts[-1], common_flags, dir_fd=current_fd)
        metadata = os.fstat(file_fd)
        if not stat.S_ISREG(metadata.st_mode) or metadata.st_nlink != 1:
            raise ValueError("artifact path must identify a regular, non-linked file")
        handle = os.fdopen(file_fd, "rb")
        file_fd = None
        return handle
    except OSError as exc:
        if isinstance(exc, PermissionError) or exc.errno in {errno.ELOOP, errno.ENOTDIR}:
            raise ValueError(
                f"artifact path resolves outside workflow run {layout.run_id!r} "
                "or contains a symlink"
            ) from exc
        raise
    finally:
        if file_fd is not None:
            os.close(file_fd)
        if current_fd is not None:
            os.close(current_fd)

scoped_artifact_path(path, operation, *folders, session_state=None, workflow_slug=None, run_id=None)

Scope an ordinary relative artifact path into the workflow-run layout.

Source code in src/cs_copilot/storage/layout.py

def scoped_artifact_path(
    path: str,
    operation: OutputOperation,
    *folders: str,
    session_state: Optional[dict[str, Any]] = None,
    workflow_slug: Optional[str] = None,
    run_id: Optional[str] = None,
) -> str:
    """Scope an ordinary relative artifact path into the workflow-run layout."""

    if is_explicit_storage_path(path) or is_workflow_scoped_path(path):
        return path
    filename = sanitize_path_part(path)
    return operation_rel_path(
        operation,
        *folders,
        filename,
        session_state=session_state,
        workflow_slug=workflow_slug,
        run_id=run_id,
    )

set_output_context(context)

Validate and activate a serialized output context.

Source code in src/cs_copilot/storage/layout.py

def set_output_context(context: dict[str, Any]) -> dict[str, Any]:
    """Validate and activate a serialized output context."""

    required = {"layout_version", "session_id", "run_id", "workflow_slug"}
    missing = sorted(required - set(context))
    if missing:
        raise ValueError(f"output context is missing fields: {', '.join(missing)}")
    if context["layout_version"] != LAYOUT_VERSION:
        raise ValueError(
            f"unsupported output layout version {context['layout_version']!r}; "
            f"expected {LAYOUT_VERSION}"
        )
    normalized = {
        "layout_version": LAYOUT_VERSION,
        "session_id": validate_identifier(context["session_id"], field="session_id"),
        "run_id": validate_identifier(context["run_id"], field="run_id"),
        "workflow_slug": sanitize_workflow_slug(context["workflow_slug"]),
    }
    _RUN_CONTEXT.set(normalized)
    return normalized

validate_identifier(value, *, field='identifier')

Validate an externally supplied identifier without silently rewriting it.

Source code in src/cs_copilot/storage/layout.py

def validate_identifier(value: Any, *, field: str = "identifier") -> str:
    """Validate an externally supplied identifier without silently rewriting it."""

    text = str(value or "").strip()
    if not _IDENTIFIER_RE.fullmatch(text):
        raise ValueError(
            f"{field} must start with an alphanumeric character and contain only "
            "letters, digits, '.', '_' or '-' (maximum 128 characters)"
        )
    return text

client

S3 storage client for unified file operations.

Provides transparent access to files stored in S3/MinIO or locally, with automatic session-based path management.

ScopedWriteDenied

Bases: PermissionError

A write refused by :meth:S3.confine_writes, with a stable verdict.

Source code in src/cs_copilot/storage/client.py

class ScopedWriteDenied(PermissionError):
    """A write refused by :meth:`S3.confine_writes`, with a stable ``verdict``."""

    def __init__(self, message: str, *, verdict: str) -> None:
        super().__init__(message)
        self.verdict = verdict

WriteObservation dataclass

Writes attempted during one observed call, classified like confined writes.

Each record is {"path", "mode", "verdict", "created"}. verdict is how :meth:S3.confine_writes would have treated the write: create_new, overwrite_existing, append_existing, protected_artifact, runtime_metadata, outside_boundary, absolute_path, or unsafe_path. path is where the bytes actually land (session-relative when inside the session) and created tells whether the write created a new file inside the boundary. Observation never changes what the write does.

Source code in src/cs_copilot/storage/client.py

@dataclass
class WriteObservation:
    """Writes attempted during one observed call, classified like confined writes.

    Each record is ``{"path", "mode", "verdict", "created"}``. ``verdict`` is
    how :meth:`S3.confine_writes` would have treated the write:
    ``create_new``, ``overwrite_existing``, ``append_existing``,
    ``protected_artifact``, ``runtime_metadata``, ``outside_boundary``,
    ``absolute_path``, or ``unsafe_path``. ``path`` is where the bytes actually
    land (session-relative when inside the session) and ``created`` tells
    whether the write created a new file inside the boundary. Observation
    never changes what the write does.
    """

    boundary: str
    protected: frozenset[str] = frozenset()
    records: list[dict[str, Any]] = field(default_factory=list)
    _lock: threading.Lock = field(default_factory=threading.Lock, repr=False)

    def record(self, entry: dict[str, Any]) -> None:
        with self._lock:
            self.records.append(entry)

    def snapshot(self) -> list[dict[str, Any]]:
        with self._lock:
            return [dict(entry) for entry in self.records]

    def created_paths(self) -> list[str]:
        """Session-relative paths of new files written inside the boundary, in order."""

        return list(
            dict.fromkeys(entry["path"] for entry in self.snapshot() if entry.get("created"))
        )

created_paths()

Session-relative paths of new files written inside the boundary, in order.

Source code in src/cs_copilot/storage/client.py

def created_paths(self) -> list[str]:
    """Session-relative paths of new files written inside the boundary, in order."""

    return list(
        dict.fromkeys(entry["path"] for entry in self.snapshot() if entry.get("created"))
    )

S3

Unified storage client for S3/MinIO and local file operations.

All file operations are scoped to the current session by default, unless an absolute S3 URL or local path is provided.

Class Attributes:

prefix : str Session-scoped prefix for all relative paths

Methods:

path(rel: str) -> str Convert a relative path to an S3 URL

open(rel: str, mode: str = "rb") Open a file for reading or writing

Examples:

Write to session-scoped S3 path

with S3.open("results.csv", "w") as f: ... f.write("data")

Read from absolute S3 URL

with S3.open("s3://bucket/key.csv", "r") as f: ... data = f.read()

Read from local file

with S3.open("/tmp/local.csv", "r") as f: ... data = f.read()

Source code in src/cs_copilot/storage/client.py

class S3(metaclass=_S3Meta):
    """
    Unified storage client for S3/MinIO and local file operations.

    All file operations are scoped to the current session by default,
    unless an absolute S3 URL or local path is provided.

    Class Attributes:
    -----------------
    prefix : str
        Session-scoped prefix for all relative paths

    Methods:
    --------
    path(rel: str) -> str
        Convert a relative path to an S3 URL

    open(rel: str, mode: str = "rb")
        Open a file for reading or writing

    Examples:
    ---------
    >>> # Write to session-scoped S3 path
    >>> with S3.open("results.csv", "w") as f:
    ...     f.write("data")

    >>> # Read from absolute S3 URL
    >>> with S3.open("s3://bucket/key.csv", "r") as f:
    ...     data = f.read()

    >>> # Read from local file
    >>> with S3.open("/tmp/local.csv", "r") as f:
    ...     data = f.read()
    """

    _fallback_prefix = _DEFAULT_PREFIX
    logger.info(f"Initialized S3 client with SESSION_ID: {SESSION_ID}")

    @classmethod
    def set_session_prefix(cls, prefix: str) -> None:
        """Set the storage prefix for the current execution context.

        Chainlit serves multiple chat sessions in one Python process.  A plain
        class-level prefix is shared across concurrent chats, so relative
        storage paths can leak into another user's session.  The context-local
        prefix isolates async tasks while keeping ``S3.prefix`` as a fallback
        for CLI/tests/backwards compatibility.
        """
        normalized = _normalize_storage_prefix(prefix)
        cls._fallback_prefix = normalized
        _SESSION_PREFIX.set(normalized)

    @classmethod
    def current_prefix(cls) -> str:
        """Return the active context-local prefix, falling back to ``S3.prefix``."""
        return _SESSION_PREFIX.get() or cls._fallback_prefix

    @classmethod
    @contextmanager
    def scoped_session_prefix(cls, prefix: str) -> Iterator[None]:
        """Bind a session prefix for this execution context only.

        Unlike :meth:`set_session_prefix`, the process-wide fallback is left
        untouched, so concurrent chats in one process cannot leak into each
        other through it.
        """

        token = _SESSION_PREFIX.set(_normalize_storage_prefix(prefix))
        try:
            yield
        finally:
            _SESSION_PREFIX.reset(token)

    @classmethod
    @contextmanager
    def observe_writes(
        cls,
        boundary: str,
        *,
        protected_paths: Iterable[str] = (),
    ) -> Iterator[WriteObservation]:
        """Record, without enforcing, how confined writes would treat each write.

        While active and no :meth:`confine_writes` boundary is set, every write
        through :meth:`open` is classified against ``boundary`` and the
        protected paths, then performed exactly as it would be without the
        observer. Classification errors are recorded, never raised.
        """

        observation = WriteObservation(
            boundary=cls._normalize_relative_path(boundary, allow_empty=True),
            protected=frozenset(str(path) for path in protected_paths),
        )
        token = _WRITE_OBSERVATION.set(observation)
        try:
            yield observation
        finally:
            _WRITE_OBSERVATION.reset(token)

    @classmethod
    def _observe_write(cls, observation: WriteObservation, rel: str, mode: str) -> None:
        try:
            entry = cls._classify_observed_write(observation, rel, mode)
        except Exception:  # noqa: BLE001 - observation must never change a write
            logger.debug("Could not classify an observed write to %r", rel, exc_info=True)
            entry = {"path": str(rel), "verdict": "unclassified", "created": False}
        observation.record({**entry, "mode": mode})

    @classmethod
    def _classify_observed_write(
        cls,
        observation: WriteObservation,
        rel: str,
        mode: str,
    ) -> dict[str, Any]:
        candidate = str(rel).strip()
        actual = cls._verified_read_key(candidate)
        boundary = PurePosixPath(observation.boundary) if observation.boundary else PurePosixPath()
        inside = False
        if actual is not None:
            try:
                PurePosixPath(actual).relative_to(boundary)
                inside = True
            except ValueError:
                inside = False
        existed = cls._storage_key_exists(actual) if inside and actual is not None else False
        token = _WRITE_PROTECTED_PATHS.set(observation.protected)
        try:
            cls._validate_scoped_write(candidate, boundary=observation.boundary)
        except ScopedWriteDenied as denied:
            verdict = denied.verdict
        except (PermissionError, ValueError):
            verdict = "unsafe_path"
        else:
            if "a" in mode or "+" in mode:
                verdict = "append_existing" if existed else "create_new"
            else:
                verdict = "overwrite_existing" if existed else "create_new"
        finally:
            _WRITE_PROTECTED_PATHS.reset(token)
        return {
            "path": actual or candidate,
            "verdict": verdict,
            # Runtime metadata and registered artifacts are never new evidence.
            "created": bool(
                inside and not existed and verdict not in {"runtime_metadata", "protected_artifact"}
            ),
        }

    @classmethod
    def _storage_key_exists(cls, key: str) -> bool:
        if not is_s3_enabled():
            return cls._local_session_path(key).exists()
        filesystem, filesystem_path = fsspec.core.url_to_fs(
            cls.path(key),
            **get_s3_config().to_storage_options(),
        )
        return bool(filesystem.exists(filesystem_path))

    @classmethod
    @contextmanager
    def confine_writes(
        cls,
        boundary: str,
        *,
        protected_paths: Iterable[str] = (),
        publication_receipt: dict[str, dict[str, Any]] | None = None,
        commit_policy: str = "on_exit",
    ) -> Iterator[None]:
        """Confine relative writes to one session subtree for this execution context.

        The MCP adapter activates this guard while a session-scoped tool runs.
        It is context-local, propagates through ``asyncio.to_thread``, and is
        enforced again by the worker process. Existing registered artifacts
        can be supplied as protected paths so a tool cannot mutate immutable
        scientific evidence. Reads remain unaffected.

        ``commit_policy="on_exit"`` (default) keeps every write private until
        the scope exits cleanly. ``"on_close"`` publishes each file as soon as
        it is closed, so code that reads its own outputs through other APIs
        keeps working; the files are still recorded in ``publication_receipt``
        and rolled back if the scope exits with an exception.
        """

        if commit_policy == "on_close":
            with cls._confine_writes_on_close(
                boundary,
                protected_paths=protected_paths,
                publication_receipt=publication_receipt,
            ):
                yield
            return
        if commit_policy != "on_exit":
            raise ValueError(f"unknown commit policy: {commit_policy!r}")

        normalized = cls._normalize_relative_path(boundary, allow_empty=True)
        boundary_path = PurePosixPath(normalized) if normalized else PurePosixPath()
        protected: set[str] = set()
        for path in protected_paths:
            protected_path = cls._normalize_relative_path(path)
            try:
                PurePosixPath(protected_path).relative_to(boundary_path)
            except ValueError as exc:
                raise ValueError(
                    "protected write path must remain inside the active boundary"
                ) from exc
            protected.add(protected_path)
        normalized_protected = frozenset(protected)
        outer_pending = _PENDING_ARTIFACT_WRITES.get()
        pending: list[Any] = outer_pending if outer_pending is not None else []
        owns_transaction = outer_pending is None
        outer_staged = _STAGED_ARTIFACT_WRITES.get()
        staged: dict[str, Any] = outer_staged if outer_staged is not None else {}
        boundary_token = _WRITE_BOUNDARY.set(normalized)
        protected_token = _WRITE_PROTECTED_PATHS.set(normalized_protected)
        pending_token = _PENDING_ARTIFACT_WRITES.set(pending)
        staged_token = _STAGED_ARTIFACT_WRITES.set(staged)
        try:
            yield
        except BaseException:
            if owns_transaction:
                for writer in reversed(pending):
                    writer._abort()
            raise
        else:
            if owns_transaction:
                committed: list[Any] = []
                prepared_publications: dict[str, dict[str, Any]] = {}
                try:
                    if publication_receipt is not None:
                        for writer in pending:
                            staged_key = getattr(writer, "_staged_key", None)
                            if staged_key is None:
                                continue
                            prepared_publications[staged_key] = cls._pending_writer_metadata(
                                writer, staged_key
                            )
                    for writer in pending:
                        writer._commit()
                        committed.append(writer)
                except BaseException:
                    cleanup_errors: list[BaseException] = []
                    for writer in reversed(pending[len(committed) :]):
                        try:
                            writer._abort()
                        except BaseException as exc:
                            cleanup_errors.append(exc)
                            logger.warning(
                                "Could not abort a pending artifact publication",
                                exc_info=True,
                            )
                    for writer in reversed(committed):
                        try:
                            writer._rollback_published()
                        except BaseException as exc:
                            cleanup_errors.append(exc)
                            logger.warning(
                                "Refused or failed to roll back a published artifact",
                                exc_info=True,
                            )
                    if cleanup_errors:
                        raise RuntimeError(
                            "artifact transaction failed and could not be fully rolled back"
                        ) from cleanup_errors[0]
                    raise
                else:
                    if publication_receipt is not None:
                        publication_receipt.update(prepared_publications)
        finally:
            _STAGED_ARTIFACT_WRITES.reset(staged_token)
            _PENDING_ARTIFACT_WRITES.reset(pending_token)
            _WRITE_PROTECTED_PATHS.reset(protected_token)
            _WRITE_BOUNDARY.reset(boundary_token)

    @classmethod
    @contextmanager
    def _confine_writes_on_close(
        cls,
        boundary: str,
        *,
        protected_paths: Iterable[str],
        publication_receipt: dict[str, dict[str, Any]] | None,
    ) -> Iterator[None]:
        if _PENDING_ARTIFACT_WRITES.get() is not None:
            raise RuntimeError("on-close confinement cannot nest inside a deferred transaction")
        normalized = cls._normalize_relative_path(boundary, allow_empty=True)
        boundary_path = PurePosixPath(normalized) if normalized else PurePosixPath()
        protected: set[str] = set()
        for path in protected_paths:
            protected_path = cls._normalize_relative_path(path)
            try:
                PurePosixPath(protected_path).relative_to(boundary_path)
            except ValueError as exc:
                raise ValueError(
                    "protected write path must remain inside the active boundary"
                ) from exc
            protected.add(protected_path)
        committed: list[Any] = []
        boundary_token = _WRITE_BOUNDARY.set(normalized)
        protected_token = _WRITE_PROTECTED_PATHS.set(frozenset(protected))
        committed_token = _COMMITTED_ON_CLOSE.set(committed)
        try:
            yield
        except BaseException:
            cleanup_errors: list[BaseException] = []
            for writer in reversed(committed):
                try:
                    writer._rollback_published()
                except BaseException as exc:
                    cleanup_errors.append(exc)
                    logger.warning(
                        "Refused or failed to roll back a published artifact",
                        exc_info=True,
                    )
            if cleanup_errors:
                raise RuntimeError(
                    "artifact transaction failed and could not be fully rolled back"
                ) from cleanup_errors[0]
            raise
        else:
            if publication_receipt is not None:
                for writer in committed:
                    key = getattr(writer, "_staged_key", None)
                    if key is None or writer._expected_sha256 is None:
                        continue
                    publication_receipt[key] = {
                        "staged_path": key,
                        "sha256": writer._expected_sha256,
                        "size_bytes": writer._expected_size,
                    }
        finally:
            _COMMITTED_ON_CLOSE.reset(committed_token)
            _WRITE_PROTECTED_PATHS.reset(protected_token)
            _WRITE_BOUNDARY.reset(boundary_token)

    @classmethod
    def exists(cls, rel: str) -> bool:
        """Return whether a storage path (relative, local, or S3 URL) exists."""

        candidate = str(rel).strip()
        if cls._is_explicit_local_path(candidate):
            local = (
                unquote(urlsplit(candidate).path) if candidate.startswith("file://") else candidate
            )
            return Path(local).exists()
        if not cls._is_s3_url(candidate) and not is_s3_enabled():
            return cls._local_session_path(candidate).exists()
        url = candidate if cls._is_s3_url(candidate) else cls.path(candidate)
        filesystem, filesystem_path = fsspec.core.url_to_fs(
            url, **get_s3_config().to_storage_options()
        )
        filesystem.invalidate_cache(filesystem_path.rsplit("/", 1)[0])
        return bool(filesystem.exists(filesystem_path))

    @classmethod
    def first_free_path(cls, rel: str) -> str:
        """Return ``rel``, or its first free ``<stem>-v<N><suffix>`` sibling.

        Workflow outputs are immutable once registered and confined writes are
        create-only, so re-running a step writes a new version of each output
        instead of overwriting the previous one. The same spelling (relative,
        local, or S3 URL) is returned.
        """

        if not cls.exists(rel):
            return rel
        version = 2
        while cls.exists(_versioned_name(rel, version)):
            version += 1
        return _versioned_name(rel, version)

    @classmethod
    def latest_version_path(cls, rel: str) -> str:
        """Return the newest existing version of ``rel`` written by :meth:`first_free_path`.

        Readers that locate an output by its deterministic name use this to
        find the most recent re-run. ``rel`` itself is returned when it has no
        later version (or does not exist).
        """

        latest = rel
        if not cls.exists(rel):
            return rel
        version = 2
        while cls.exists(_versioned_name(rel, version)):
            latest = _versioned_name(rel, version)
            version += 1
        return latest

    @classmethod
    def session_relative(cls, path: str) -> str | None:
        """Return the session-relative key of a path in the active session, if any."""

        return cls._verified_read_key(path)

    @staticmethod
    def _pending_writer_metadata(writer: Any, final_key: str) -> dict[str, Any]:
        """Checksum a private transaction object before it is published."""

        digest = hashlib.sha256()
        size = 0
        with writer._open_staged_read("rb") as source:
            while True:
                chunk = source.read(1024 * 1024)
                if not chunk:
                    break
                if isinstance(chunk, str):
                    chunk = chunk.encode("utf-8")
                digest.update(chunk)
                size += len(chunk)
        return {
            "staged_path": final_key,
            "sha256": digest.hexdigest(),
            "size_bytes": size,
        }

    @staticmethod
    def current_write_boundary() -> str | None:
        """Return the active write boundary, or ``None`` outside guarded execution."""

        return _WRITE_BOUNDARY.get()

    @staticmethod
    def current_write_protected_paths() -> frozenset[str]:
        """Return immutable session-relative paths for the active tool call."""

        return _WRITE_PROTECTED_PATHS.get()

    @classmethod
    @contextmanager
    def stage_publications(cls, staging_id: str) -> Iterator[None]:
        """Redirect scoped writes to a worker-private publication namespace."""

        normalized_id = cls._normalize_relative_path(staging_id)
        if len(PurePosixPath(normalized_id).parts) != 1:
            raise ValueError("staging_id must be one safe path component")
        publications: dict[str, str] = {}
        id_token = _EXTERNAL_STAGING_ID.set(normalized_id)
        publications_token = _EXTERNAL_STAGED_PUBLICATIONS.set(publications)
        try:
            yield
        finally:
            _EXTERNAL_STAGED_PUBLICATIONS.reset(publications_token)
            _EXTERNAL_STAGING_ID.reset(id_token)

    @staticmethod
    def current_staged_publications() -> dict[str, str]:
        """Return final-to-staged keys created by the active worker invocation."""

        return dict(_EXTERNAL_STAGED_PUBLICATIONS.get() or {})

    @classmethod
    def staged_publication_metadata(cls) -> dict[str, dict[str, Any]]:
        """Checksum committed worker staging objects through a no-follow read."""

        metadata: dict[str, dict[str, Any]] = {}
        for final_key, staged_key in cls.current_staged_publications().items():
            digest = hashlib.sha256()
            size = 0
            if is_s3_enabled():
                config = get_s3_config()
                source = fsspec.open(
                    cls.path(staged_key),
                    mode="rb",
                    **config.to_storage_options(),
                ).open()
            else:
                source = cls._open_local_verified_source(staged_key)
            with source:
                while True:
                    chunk = source.read(1024 * 1024)
                    if not chunk:
                        break
                    if isinstance(chunk, str):
                        chunk = chunk.encode("utf-8")
                    digest.update(chunk)
                    size += len(chunk)
            metadata[final_key] = {
                "staged_path": staged_key,
                "sha256": digest.hexdigest(),
                "size_bytes": size,
            }
        return metadata

    @classmethod
    def promote_staged_publications(
        cls,
        publications: Mapping[str, Mapping[str, Any]],
    ) -> None:
        """Promote complete worker outputs to final create-only artifact keys."""

        promoted: dict[str, Mapping[str, Any]] = {}
        try:
            for final_key, metadata in publications.items():
                final = cls._normalize_relative_path(final_key)
                staged = cls._normalize_relative_path(str(metadata["staged_path"]))
                source = cls._open_verified_snapshot(
                    staged,
                    "rb",
                    expected_sha256=str(metadata["sha256"]),
                    expected_size=int(metadata["size_bytes"]),
                )
                with source:
                    with cls.open_atomic(final, "xb") as destination:
                        while True:
                            chunk = source.read(1024 * 1024)
                            if not chunk:
                                break
                            destination.write(chunk)
                promoted[final] = metadata
            for metadata in publications.values():
                cls._remove_storage_key(cls._normalize_relative_path(str(metadata["staged_path"])))
        except BaseException:
            cls.rollback_promoted_publications(promoted)
            raise

    @classmethod
    def rollback_promoted_publications(
        cls,
        publications: Mapping[str, Mapping[str, Any]],
    ) -> None:
        """Remove only finals that still match the worker's promoted bytes."""

        for final_key, metadata in reversed(tuple(publications.items())):
            final = cls._normalize_relative_path(final_key)
            try:
                cls._remove_storage_key_verified(
                    final,
                    expected_sha256=str(metadata["sha256"]),
                    expected_size=int(metadata["size_bytes"]),
                )
            except FileNotFoundError:
                continue
            except Exception:
                logger.warning(
                    "Refusing to roll back changed worker publication %s",
                    final,
                    exc_info=True,
                )
                continue

    @classmethod
    def discard_staging_prefix(cls, boundary: str, staging_id: str) -> None:
        """Remove one exact worker staging subtree after failure or cancellation."""

        normalized_boundary = cls._normalize_relative_path(boundary)
        normalized_id = cls._normalize_relative_path(staging_id)
        if len(PurePosixPath(normalized_id).parts) != 1:
            raise ValueError("staging_id must be one safe path component")
        prefix = PurePosixPath(
            normalized_boundary,
            ".staging",
            normalized_id,
        ).as_posix()
        if is_s3_enabled():
            config = get_s3_config()
            fs, fs_path = fsspec.core.url_to_fs(
                cls.path(prefix),
                **config.to_storage_options(),
            )
            if fs.exists(fs_path):
                fs.rm(fs_path, recursive=True)
            return
        _secure_local_rmtree(cls._local_session_path(prefix).absolute())

    @classmethod
    def _remove_storage_key(cls, key: str) -> None:
        if is_s3_enabled():
            config = get_s3_config()
            fs, fs_path = fsspec.core.url_to_fs(
                cls.path(key),
                **config.to_storage_options(),
            )
            if fs.exists(fs_path):
                fs.rm(fs_path)
            return
        _secure_local_unlink(cls._local_session_path(key).absolute())

    @classmethod
    def _remove_storage_key_verified(
        cls,
        key: str,
        *,
        expected_sha256: str,
        expected_size: int,
    ) -> bool:
        """Remove one exact publication only while its registered bytes match."""

        if is_s3_enabled():
            config = get_s3_config()
            fs, fs_path = fsspec.core.url_to_fs(
                cls.path(key),
                **config.to_storage_options(),
            )
            return _remove_fsspec_path_verified(
                fs,
                fs_path,
                expected_sha256=expected_sha256,
                expected_size=expected_size,
            )
        return _secure_local_unlink_verified(
            cls._local_session_path(key).absolute(),
            expected_sha256=expected_sha256,
            expected_size=expected_size,
        )

    @classmethod
    @contextmanager
    def confine_artifact_reads(
        cls,
        artifacts: Mapping[str, tuple[str, int]],
    ) -> Iterator[None]:
        """Pin registered artifact reads to checksum-verified byte snapshots."""

        inherited = dict(_VERIFIED_ARTIFACT_READS.get() or {})
        for path, expected in artifacts.items():
            normalized = cls._normalize_relative_path(path)
            if (
                not isinstance(expected, tuple)
                or len(expected) != 2
                or not isinstance(expected[0], str)
                or not isinstance(expected[1], int)
                or expected[1] < 0
            ):
                raise ValueError("verified artifact metadata must be (sha256, size_bytes)")
            inherited[normalized] = (expected[0], expected[1])
        token = _VERIFIED_ARTIFACT_READS.set(inherited)
        try:
            yield
        finally:
            _VERIFIED_ARTIFACT_READS.reset(token)

    @staticmethod
    def current_verified_artifact_reads() -> dict[str, tuple[str, int]]:
        """Return a detached copy of invocation-pinned artifact read metadata."""

        return dict(_VERIFIED_ARTIFACT_READS.get() or {})

    @classmethod
    def path(cls, rel: str) -> str:
        """
        Convert a relative path to an S3 URL or local session path.

        Explicit S3 and local paths are returned unchanged. Relative paths are
        resolved against the active backend.

        Args:
            rel: Relative path or absolute S3 URL

        Returns:
            str: Full S3 URL or local path

        Examples:
        ---------
        >>> S3.path("data.csv")
        's3://chatbot-assets/sessions/20250121-123456-abc123/data.csv'

        >>> S3.path("s3://mybucket/data.csv")
        's3://mybucket/data.csv'
        """
        if cls._is_s3_url(rel) or cls._is_explicit_local_path(rel):
            return rel

        config = get_s3_config()
        if not is_s3_enabled():
            return os.fspath(cls._local_session_path(rel))

        key = PurePosixPath(cls.current_prefix().strip("/")) / str(rel).lstrip("/")
        return f"s3://{config.bucket_name}/{key.as_posix()}"

    @classmethod
    def open(cls, rel: str, mode: str = "rb") -> Any:
        """
        Open a file for reading or writing.

        Supports three types of paths:
        1. Absolute S3 URLs (s3://...)
        2. Local absolute paths (/ or file://)
        3. Relative paths (scoped to current session)

        Args:
            rel: File path (relative, absolute, or S3 URL)
            mode: File mode ('r', 'w', 'rb', 'wb', etc.)

        Returns:
            File-like object opened with the specified mode

        Examples:
        ---------
        >>> # Read from session-scoped S3 path
        >>> with S3.open("data.csv", "r") as f:
        ...     df = pd.read_csv(f)

        >>> # Write to session-scoped S3 path
        >>> with S3.open("output.csv", "w") as f:
        ...     df.to_csv(f)

        >>> # Read from absolute S3 URL
        >>> with S3.open("s3://bucket/data.csv", "r") as f:
        ...     df = pd.read_csv(f)

        >>> # Read from local file
        >>> with S3.open("/tmp/data.csv", "r") as f:
        ...     df = pd.read_csv(f)
        """
        config = get_s3_config()
        write_mode = cls._is_write_mode(mode)
        if not write_mode:
            read_key = cls._verified_read_key(rel)
            staged = (
                _STAGED_ARTIFACT_WRITES.get().get(read_key)
                if read_key is not None and _STAGED_ARTIFACT_WRITES.get() is not None
                else None
            )
            if staged is not None:
                return staged._open_staged_read(mode)
            expected_reads = _VERIFIED_ARTIFACT_READS.get() or {}
            expected = expected_reads.get(read_key) if read_key else None
            if read_key is not None and expected is not None:
                return cls._open_verified_snapshot(
                    read_key,
                    mode,
                    expected_sha256=expected[0],
                    expected_size=expected[1],
                )

        write_boundary = cls.current_write_boundary() if write_mode else None
        if write_mode and write_boundary is None:
            observation = _WRITE_OBSERVATION.get()
            if observation is not None:
                cls._observe_write(observation, rel, mode)
        scoped_write_key: str | None = None
        if write_boundary is not None:
            scoped_write_key = cls._validate_scoped_write(rel, boundary=write_boundary)
        storage_target = rel
        staging_id = _EXTERNAL_STAGING_ID.get()
        publications = _EXTERNAL_STAGED_PUBLICATIONS.get()
        if scoped_write_key is not None and staging_id is not None and publications is not None:
            relative_to_boundary = PurePosixPath(scoped_write_key).relative_to(
                PurePosixPath(write_boundary or "")
            )
            staged_key = PurePosixPath(
                write_boundary or "",
                ".staging",
                staging_id,
                relative_to_boundary,
            ).as_posix()
            publications[scoped_write_key] = staged_key
            storage_target = staged_key

        if cls._is_s3_url(storage_target):
            if scoped_write_key is None:
                return fsspec.open(
                    storage_target,
                    mode=mode,
                    **config.to_storage_options(),
                )
            return cls._open_confined_s3_path(
                storage_target,
                mode,
                storage_options=config.to_storage_options(),
                public_write_key=scoped_write_key,
            )

        if cls._is_explicit_local_path(storage_target):
            if storage_target.startswith("file://"):
                return fsspec.open(storage_target, mode=mode)
            return cls._open_local_path(Path(storage_target), mode)

        if not is_s3_enabled():
            return cls._open_local_path(
                cls._local_session_path(storage_target),
                mode,
                scoped_write_key=scoped_write_key,
            )

        if scoped_write_key is None:
            return fsspec.open(
                cls.path(storage_target),
                mode=mode,
                **config.to_storage_options(),
            )
        return cls._open_confined_s3_path(
            cls.path(storage_target),
            mode,
            storage_options=config.to_storage_options(),
            public_write_key=scoped_write_key,
        )

    @classmethod
    def _verified_read_key(cls, path: str) -> str | None:
        """Resolve a public storage spelling to a session-relative artifact key."""

        if not isinstance(path, str) or not path.strip():
            return None
        candidate = path.strip()
        try:
            if cls._is_s3_url(candidate):
                parsed = urlsplit(candidate)
                config = get_s3_config()
                prefix = cls.current_prefix().strip("/")
                object_path = unquote(parsed.path).lstrip("/")
                expected_prefix = f"{prefix}/"
                if (
                    parsed.query
                    or parsed.fragment
                    or parsed.netloc != config.bucket_name
                    or not object_path.startswith(expected_prefix)
                ):
                    return None
                return cls._normalize_relative_path(object_path[len(expected_prefix) :])
            if cls._is_explicit_local_path(candidate):
                local_value = (
                    unquote(urlsplit(candidate).path)
                    if candidate.startswith("file://")
                    else candidate
                )
                local_path = Path(local_value).absolute()
                session_root = cls._local_session_path("").absolute()
                return cls._normalize_relative_path(local_path.relative_to(session_root).as_posix())
            expanded_prefix = PurePosixPath(
                LOCAL_STORAGE_ROOT.as_posix(),
                cls.current_prefix(),
            )
            candidate_path = PurePosixPath(candidate)
            try:
                expanded_relative = candidate_path.relative_to(expanded_prefix)
            except ValueError:
                pass
            else:
                return cls._normalize_relative_path(expanded_relative.as_posix())
            return cls._normalize_relative_path(candidate)
        except (ValueError, OSError):
            return None

    @classmethod
    def _open_verified_snapshot(
        cls,
        key: str,
        mode: str,
        *,
        expected_sha256: str,
        expected_size: int,
    ):
        """Copy one artifact once, verify it, and return the pinned byte snapshot."""

        if mode not in {"r", "rt", "rb"}:
            raise PermissionError("verified artifacts are available only in read mode")
        if is_s3_enabled():
            config = get_s3_config()
            source = fsspec.open(
                cls.path(key),
                mode="rb",
                **config.to_storage_options(),
            ).open()
        else:
            source = cls._open_local_verified_source(key)

        snapshot = tempfile.SpooledTemporaryFile(max_size=16 * 1024 * 1024, mode="w+b")
        digest = hashlib.sha256()
        size = 0
        try:
            with source:
                while True:
                    chunk = source.read(1024 * 1024)
                    if not chunk:
                        break
                    if isinstance(chunk, str):
                        chunk = chunk.encode("utf-8")
                    size += len(chunk)
                    if size > expected_size:
                        raise PermissionError(
                            f"registered artifact {key!r} exceeded its expected size"
                        )
                    digest.update(chunk)
                    snapshot.write(chunk)
            if size != expected_size or digest.hexdigest() != expected_sha256:
                raise PermissionError(f"registered artifact {key!r} failed checksum verification")
            snapshot.seek(0)
            if "b" in mode:
                return snapshot
            return io.TextIOWrapper(snapshot, encoding="utf-8")
        except BaseException:
            snapshot.close()
            raise

    @classmethod
    def _open_local_verified_source(cls, key: str):
        """Descriptor-open a registered local artifact without following links."""

        if not _OPEN_SUPPORTS_DIR_FD or not _O_NOFOLLOW:
            raise RuntimeError("secure local reads require directory-relative no-follow support")
        relative = PurePosixPath(cls._normalize_relative_path(key))
        common_flags = os.O_RDONLY | getattr(os, "O_CLOEXEC", 0) | _O_NOFOLLOW
        directory_flags = common_flags | getattr(os, "O_DIRECTORY", 0)
        current_fd: int | None = None
        file_fd: int | None = None
        try:
            current_fd = _open_existing_directory_tree(cls._local_session_path("").absolute())
            for component in relative.parts[:-1]:
                next_fd = os.open(component, directory_flags, dir_fd=current_fd)
                os.close(current_fd)
                current_fd = next_fd
            file_fd = os.open(relative.parts[-1], common_flags, dir_fd=current_fd)
            metadata = os.fstat(file_fd)
            if not stat.S_ISREG(metadata.st_mode) or metadata.st_nlink != 1:
                raise PermissionError("registered artifact must be a regular, non-linked file")
            handle = os.fdopen(file_fd, "rb")
            file_fd = None
            return handle
        except OSError as exc:
            if exc.errno in {errno.ELOOP, errno.ENOTDIR}:
                raise PermissionError(
                    "registered artifact contains a symlink or non-directory component"
                ) from exc
            raise
        finally:
            if file_fd is not None:
                os.close(file_fd)
            if current_fd is not None:
                os.close(current_fd)

    @classmethod
    def open_atomic(cls, rel: str, mode: str = "xb"):
        """Create one session-relative object that is invisible until complete."""

        if not cls._is_write_mode(mode):
            raise ValueError("open_atomic requires a write-capable mode")
        if cls._is_s3_url(rel) or cls._is_explicit_local_path(rel):
            raise ValueError("open_atomic requires a session-relative storage path")
        normalized = cls._normalize_relative_path(rel)
        config = get_s3_config()
        if not is_s3_enabled():
            return cls._open_confined_local_path(
                cls._local_session_path(normalized),
                mode,
                boundary="",
                scoped_write_key=normalized,
                defer_transaction=False,
            )
        return cls._open_confined_s3_path(
            cls.path(normalized),
            mode,
            storage_options=config.to_storage_options(),
            defer_transaction=False,
            public_write_key=None,
        )

    @staticmethod
    def _is_s3_url(path: str) -> bool:
        return isinstance(path, str) and path.startswith("s3://")

    @staticmethod
    def _is_explicit_local_path(path: str) -> bool:
        return isinstance(path, str) and (path.startswith("/") or path.startswith("file://"))

    @classmethod
    def _local_session_path(cls, rel: str) -> Path:
        session_root = LOCAL_STORAGE_ROOT / Path(cls.current_prefix().strip("/"))
        candidate = Path(rel)
        try:
            candidate.relative_to(session_root)
        except ValueError:
            return session_root / candidate
        return candidate

    @staticmethod
    def _is_write_mode(mode: str) -> bool:
        return any(flag in mode for flag in ("w", "a", "x", "+"))

    @classmethod
    def _open_local_path(
        cls,
        path: Path,
        mode: str,
        *,
        scoped_write_key: str | None = None,
    ):
        if cls._is_write_mode(mode):
            boundary = cls.current_write_boundary()
            if boundary is not None:
                return cls._open_confined_local_path(
                    path,
                    mode,
                    boundary=boundary,
                    scoped_write_key=scoped_write_key,
                )
            path.parent.mkdir(parents=True, exist_ok=True)
        return builtins.open(path, mode)

    @classmethod
    def _validate_scoped_write(cls, rel: str, *, boundary: str) -> str:
        """Validate a write destination against the active session/run prefix."""

        if not isinstance(rel, str) or not rel.strip():
            raise PermissionError("write destination must be a non-empty storage path")
        candidate = rel.strip()
        if "\x00" in candidate or "\\" in candidate:
            raise PermissionError("write destination contains an unsafe path component")

        if cls._is_s3_url(candidate):
            parsed = urlsplit(candidate)
            expected = urlsplit(cls.path(boundary))
            decoded_path = unquote(parsed.path)
            expected_path = unquote(expected.path).rstrip("/")
            if (
                parsed.query
                or parsed.fragment
                or parsed.netloc != expected.netloc
                or parsed.scheme.lower() != "s3"
                or expected.scheme.lower() != "s3"
                or not decoded_path.startswith(f"{expected_path}/")
            ):
                raise ScopedWriteDenied(
                    "S3 write destination is outside the active boundary",
                    verdict="outside_boundary",
                )
            relative = decoded_path[len(expected_path) + 1 :]
            cls._normalize_relative_path(relative)
            cls._reject_reserved_run_path(relative)
            session_relative = (
                PurePosixPath(boundary, relative).as_posix() if boundary else relative
            )
            cls._reject_protected_write(session_relative)
            return session_relative

        if cls._is_explicit_local_path(candidate):
            raise ScopedWriteDenied(
                "absolute and file:// write destinations are forbidden during scoped execution",
                verdict="absolute_path",
            )

        normalized = cls._normalize_relative_path(candidate)
        normalized_path = PurePosixPath(normalized)
        boundary_path = PurePosixPath(boundary) if boundary else PurePosixPath()
        try:
            relative = normalized_path.relative_to(boundary_path)
        except ValueError as exc:
            raise ScopedWriteDenied(
                "write destination is outside the active boundary",
                verdict="outside_boundary",
            ) from exc
        cls._reject_reserved_run_path(relative.as_posix())
        cls._reject_protected_write(normalized)
        return normalized

    @staticmethod
    def _normalize_relative_path(value: str, *, allow_empty: bool = False) -> str:
        raw = str(value).strip()
        if allow_empty and not raw:
            return ""
        if raw.startswith("/"):
            raise ValueError("storage boundary must be a safe relative path")
        text = raw.rstrip("/") if allow_empty else raw
        candidate = PurePosixPath(text)
        if (
            not text
            or not candidate.parts
            or candidate.is_absolute()
            or "\\" in text
            or any(part in {"", ".", ".."} for part in candidate.parts)
        ):
            raise ValueError("storage boundary must be a safe relative path")
        return candidate.as_posix()

    @staticmethod
    def _reject_reserved_run_path(relative: str) -> None:
        path = PurePosixPath(relative)
        parts = path.parts
        if len(parts) >= 3 and parts[0] == "workflows":
            path = PurePosixPath(*parts[2:])
            parts = path.parts
        normalized = path.as_posix()
        if normalized in _RESERVED_RUN_FILES or (parts and parts[0] in {".staging", "events"}):
            raise ScopedWriteDenied(
                "write destination is reserved for workflow runtime metadata",
                verdict="runtime_metadata",
            )

    @staticmethod
    def _reject_protected_write(session_relative: str) -> None:
        if session_relative in _WRITE_PROTECTED_PATHS.get():
            raise ScopedWriteDenied(
                "write destination is an immutable registered workflow artifact",
                verdict="protected_artifact",
            )

    @classmethod
    def _open_confined_s3_path(
        cls,
        path: str,
        mode: str,
        *,
        storage_options: dict,
        defer_transaction: bool = True,
        public_write_key: str | None = None,
    ):
        """Open one scoped S3 artifact with create-once semantics."""

        selected_mode = cls._exclusive_creation_mode(mode)
        deferred = defer_transaction and _PENDING_ARTIFACT_WRITES.get() is not None
        try:
            filesystem, filesystem_path = fsspec.core.url_to_fs(
                path,
                **storage_options,
            )
            handle = filesystem.open(
                filesystem_path,
                mode=selected_mode,
                autocommit=not deferred,
            )
        except (FileExistsError, IsADirectoryError) as exc:
            raise PermissionError(
                "scoped artifact writes are create-only; destination already exists"
            ) from exc
        except OSError as exc:
            if exc.errno == errno.EEXIST:
                raise PermissionError(
                    "scoped artifact writes are create-only; destination already exists"
                ) from exc
            raise
        return _AtomicS3CreateFile(
            handle,
            deferred=deferred,
            mode=selected_mode,
            staged_key=public_write_key or cls._verified_read_key(path),
        )

    @staticmethod
    def _exclusive_creation_mode(mode: str) -> str:
        """Translate the first scoped S3 write into an exclusive create."""

        if not isinstance(mode, str):
            raise ValueError("file mode must be a string")
        bases = [character for character in "rwax" if character in mode]
        if len(bases) != 1:
            raise ValueError(f"invalid file mode: {mode!r}")
        base = bases[0]
        if base == "x":
            return mode
        if base in {"w", "a"}:
            return mode.replace(base, "x", 1)
        raise PermissionError("scoped update modes cannot modify a pre-existing artifact")

    @classmethod
    def _open_confined_local_path(
        cls,
        path: Path,
        mode: str,
        *,
        boundary: str,
        scoped_write_key: str | None,
        defer_transaction: bool = True,
    ):
        """Open a local write by descriptor without following any symlink."""

        if not _OPEN_SUPPORTS_DIR_FD or not _O_NOFOLLOW:
            raise RuntimeError("secure local writes require directory-relative no-follow support")

        session_root = cls._local_session_path("").absolute()
        boundary_root = session_root.joinpath(*PurePosixPath(boundary).parts).absolute()
        destination = path.absolute()
        try:
            relative = destination.relative_to(boundary_root)
        except ValueError as exc:
            raise ScopedWriteDenied(
                "write destination is outside the active boundary",
                verdict="outside_boundary",
            ) from exc
        if not relative.parts:
            raise PermissionError("write destination must identify a file")

        directory_flags = (
            os.O_RDONLY | getattr(os, "O_CLOEXEC", 0) | getattr(os, "O_DIRECTORY", 0) | _O_NOFOLLOW
        )
        current_fd: int | None = None
        file_fd: int | None = None
        try:
            current_fd = cls._open_or_create_directory_tree(
                boundary_root,
                flags=directory_flags,
            )
            for component in relative.parts[:-1]:
                try:
                    os.mkdir(component, mode=0o700, dir_fd=current_fd)
                except FileExistsError:
                    pass
                next_fd = os.open(component, directory_flags, dir_fd=current_fd)
                os.close(current_fd)
                current_fd = next_fd

            if scoped_write_key is not None:
                destination_name = relative.parts[-1]
                temporary_name = f".{destination_name}.{uuid.uuid4().hex}.pending"
                flags, truncate = cls._write_open_flags(mode, exclusive_create=True)
                file_fd = os.open(temporary_name, flags, 0o600, dir_fd=current_fd)
            else:
                destination_name = relative.parts[-1]
                temporary_name = ""
                flags, truncate = cls._write_open_flags(mode)
                file_fd = os.open(destination_name, flags, 0o600, dir_fd=current_fd)
            metadata = os.fstat(file_fd)
            if not stat.S_ISREG(metadata.st_mode) or metadata.st_nlink != 1:
                raise PermissionError(
                    "write destination must be a regular, non-linked artifact file"
                )
            if truncate:
                os.ftruncate(file_fd, 0)
            fdopen_mode = mode.replace("x", "w", 1) if "x" in mode else mode
            handle = os.fdopen(file_fd, fdopen_mode)
            file_fd = None
            if scoped_write_key is not None:
                owned_directory_fd = current_fd
                current_fd = None
                return _AtomicLocalCreateFile(
                    handle,
                    directory_fd=owned_directory_fd,
                    temporary_name=temporary_name,
                    destination_name=destination_name,
                    destination_path=destination,
                    staged_key=scoped_write_key,
                    deferred=(defer_transaction and _PENDING_ARTIFACT_WRITES.get() is not None),
                )
            return handle
        except OSError as exc:
            if exc.errno == errno.EEXIST:
                raise PermissionError(
                    "scoped artifact writes are create-only; destination already exists"
                ) from exc
            if exc.errno in {errno.ELOOP, errno.ENOTDIR}:
                raise PermissionError(
                    "write destination contains a symlink or non-directory component"
                ) from exc
            raise
        finally:
            if file_fd is not None:
                os.close(file_fd)
            if current_fd is not None:
                os.close(current_fd)

    @staticmethod
    def _open_or_create_directory_tree(path: Path, *, flags: int) -> int:
        """Return a descriptor for an absolute directory tree, creating it safely."""

        if not path.is_absolute():
            raise ValueError("secure directory traversal requires an absolute path")
        current_fd = os.open(path.anchor, flags)
        try:
            for component in path.parts[1:]:
                try:
                    os.mkdir(component, mode=0o700, dir_fd=current_fd)
                except FileExistsError:
                    pass
                next_fd = os.open(component, flags, dir_fd=current_fd)
                os.close(current_fd)
                current_fd = next_fd
            result = current_fd
            current_fd = -1
            return result
        finally:
            if current_fd >= 0:
                os.close(current_fd)

    @staticmethod
    def _write_open_flags(
        mode: str,
        *,
        exclusive_create: bool = False,
    ) -> tuple[int, bool]:
        """Translate a Python write mode into no-follow ``os.open`` flags."""

        if not isinstance(mode, str):
            raise ValueError("file mode must be a string")
        if any(character not in "rwaxbt+" for character in mode):
            raise ValueError(f"unsupported file mode: {mode!r}")
        bases = [character for character in "rwax" if character in mode]
        if len(bases) != 1 or ("b" in mode and "t" in mode):
            raise ValueError(f"invalid file mode: {mode!r}")

        base = bases[0]
        if base == "r" and "+" not in mode:
            raise ValueError("confined writer requires a write-capable file mode")
        access = os.O_RDWR if "+" in mode else os.O_WRONLY
        flags = access | getattr(os, "O_CLOEXEC", 0) | getattr(os, "O_NONBLOCK", 0) | _O_NOFOLLOW
        truncate = False
        if base == "w":
            flags |= os.O_CREAT
            if exclusive_create:
                flags |= os.O_EXCL
            truncate = True
        elif base == "a":
            flags |= os.O_CREAT | os.O_APPEND
            if exclusive_create:
                flags |= os.O_EXCL
        elif base == "x":
            flags |= os.O_CREAT | os.O_EXCL
        elif base == "r" and exclusive_create:
            raise PermissionError("scoped update modes cannot modify a pre-existing artifact")
        return flags, truncate

set_session_prefix(prefix) classmethod

Set the storage prefix for the current execution context.

Chainlit serves multiple chat sessions in one Python process. A plain class-level prefix is shared across concurrent chats, so relative storage paths can leak into another user's session. The context-local prefix isolates async tasks while keeping S3.prefix as a fallback for CLI/tests/backwards compatibility.

Source code in src/cs_copilot/storage/client.py

@classmethod
def set_session_prefix(cls, prefix: str) -> None:
    """Set the storage prefix for the current execution context.

    Chainlit serves multiple chat sessions in one Python process.  A plain
    class-level prefix is shared across concurrent chats, so relative
    storage paths can leak into another user's session.  The context-local
    prefix isolates async tasks while keeping ``S3.prefix`` as a fallback
    for CLI/tests/backwards compatibility.
    """
    normalized = _normalize_storage_prefix(prefix)
    cls._fallback_prefix = normalized
    _SESSION_PREFIX.set(normalized)

current_prefix() classmethod

Return the active context-local prefix, falling back to S3.prefix.

Source code in src/cs_copilot/storage/client.py

@classmethod
def current_prefix(cls) -> str:
    """Return the active context-local prefix, falling back to ``S3.prefix``."""
    return _SESSION_PREFIX.get() or cls._fallback_prefix

scoped_session_prefix(prefix) classmethod

Bind a session prefix for this execution context only.

Unlike :meth:set_session_prefix, the process-wide fallback is left untouched, so concurrent chats in one process cannot leak into each other through it.

Source code in src/cs_copilot/storage/client.py

@classmethod
@contextmanager
def scoped_session_prefix(cls, prefix: str) -> Iterator[None]:
    """Bind a session prefix for this execution context only.

    Unlike :meth:`set_session_prefix`, the process-wide fallback is left
    untouched, so concurrent chats in one process cannot leak into each
    other through it.
    """

    token = _SESSION_PREFIX.set(_normalize_storage_prefix(prefix))
    try:
        yield
    finally:
        _SESSION_PREFIX.reset(token)

observe_writes(boundary, *, protected_paths=()) classmethod

Record, without enforcing, how confined writes would treat each write.

While active and no :meth:confine_writes boundary is set, every write through :meth:open is classified against boundary and the protected paths, then performed exactly as it would be without the observer. Classification errors are recorded, never raised.

Source code in src/cs_copilot/storage/client.py

@classmethod
@contextmanager
def observe_writes(
    cls,
    boundary: str,
    *,
    protected_paths: Iterable[str] = (),
) -> Iterator[WriteObservation]:
    """Record, without enforcing, how confined writes would treat each write.

    While active and no :meth:`confine_writes` boundary is set, every write
    through :meth:`open` is classified against ``boundary`` and the
    protected paths, then performed exactly as it would be without the
    observer. Classification errors are recorded, never raised.
    """

    observation = WriteObservation(
        boundary=cls._normalize_relative_path(boundary, allow_empty=True),
        protected=frozenset(str(path) for path in protected_paths),
    )
    token = _WRITE_OBSERVATION.set(observation)
    try:
        yield observation
    finally:
        _WRITE_OBSERVATION.reset(token)

confine_writes(boundary, *, protected_paths=(), publication_receipt=None, commit_policy='on_exit') classmethod

Confine relative writes to one session subtree for this execution context.

The MCP adapter activates this guard while a session-scoped tool runs. It is context-local, propagates through asyncio.to_thread, and is enforced again by the worker process. Existing registered artifacts can be supplied as protected paths so a tool cannot mutate immutable scientific evidence. Reads remain unaffected.

commit_policy="on_exit" (default) keeps every write private until the scope exits cleanly. "on_close" publishes each file as soon as it is closed, so code that reads its own outputs through other APIs keeps working; the files are still recorded in publication_receipt and rolled back if the scope exits with an exception.

Source code in src/cs_copilot/storage/client.py

@classmethod
@contextmanager
def confine_writes(
    cls,
    boundary: str,
    *,
    protected_paths: Iterable[str] = (),
    publication_receipt: dict[str, dict[str, Any]] | None = None,
    commit_policy: str = "on_exit",
) -> Iterator[None]:
    """Confine relative writes to one session subtree for this execution context.

    The MCP adapter activates this guard while a session-scoped tool runs.
    It is context-local, propagates through ``asyncio.to_thread``, and is
    enforced again by the worker process. Existing registered artifacts
    can be supplied as protected paths so a tool cannot mutate immutable
    scientific evidence. Reads remain unaffected.

    ``commit_policy="on_exit"`` (default) keeps every write private until
    the scope exits cleanly. ``"on_close"`` publishes each file as soon as
    it is closed, so code that reads its own outputs through other APIs
    keeps working; the files are still recorded in ``publication_receipt``
    and rolled back if the scope exits with an exception.
    """

    if commit_policy == "on_close":
        with cls._confine_writes_on_close(
            boundary,
            protected_paths=protected_paths,
            publication_receipt=publication_receipt,
        ):
            yield
        return
    if commit_policy != "on_exit":
        raise ValueError(f"unknown commit policy: {commit_policy!r}")

    normalized = cls._normalize_relative_path(boundary, allow_empty=True)
    boundary_path = PurePosixPath(normalized) if normalized else PurePosixPath()
    protected: set[str] = set()
    for path in protected_paths:
        protected_path = cls._normalize_relative_path(path)
        try:
            PurePosixPath(protected_path).relative_to(boundary_path)
        except ValueError as exc:
            raise ValueError(
                "protected write path must remain inside the active boundary"
            ) from exc
        protected.add(protected_path)
    normalized_protected = frozenset(protected)
    outer_pending = _PENDING_ARTIFACT_WRITES.get()
    pending: list[Any] = outer_pending if outer_pending is not None else []
    owns_transaction = outer_pending is None
    outer_staged = _STAGED_ARTIFACT_WRITES.get()
    staged: dict[str, Any] = outer_staged if outer_staged is not None else {}
    boundary_token = _WRITE_BOUNDARY.set(normalized)
    protected_token = _WRITE_PROTECTED_PATHS.set(normalized_protected)
    pending_token = _PENDING_ARTIFACT_WRITES.set(pending)
    staged_token = _STAGED_ARTIFACT_WRITES.set(staged)
    try:
        yield
    except BaseException:
        if owns_transaction:
            for writer in reversed(pending):
                writer._abort()
        raise
    else:
        if owns_transaction:
            committed: list[Any] = []
            prepared_publications: dict[str, dict[str, Any]] = {}
            try:
                if publication_receipt is not None:
                    for writer in pending:
                        staged_key = getattr(writer, "_staged_key", None)
                        if staged_key is None:
                            continue
                        prepared_publications[staged_key] = cls._pending_writer_metadata(
                            writer, staged_key
                        )
                for writer in pending:
                    writer._commit()
                    committed.append(writer)
            except BaseException:
                cleanup_errors: list[BaseException] = []
                for writer in reversed(pending[len(committed) :]):
                    try:
                        writer._abort()
                    except BaseException as exc:
                        cleanup_errors.append(exc)
                        logger.warning(
                            "Could not abort a pending artifact publication",
                            exc_info=True,
                        )
                for writer in reversed(committed):
                    try:
                        writer._rollback_published()
                    except BaseException as exc:
                        cleanup_errors.append(exc)
                        logger.warning(
                            "Refused or failed to roll back a published artifact",
                            exc_info=True,
                        )
                if cleanup_errors:
                    raise RuntimeError(
                        "artifact transaction failed and could not be fully rolled back"
                    ) from cleanup_errors[0]
                raise
            else:
                if publication_receipt is not None:
                    publication_receipt.update(prepared_publications)
    finally:
        _STAGED_ARTIFACT_WRITES.reset(staged_token)
        _PENDING_ARTIFACT_WRITES.reset(pending_token)
        _WRITE_PROTECTED_PATHS.reset(protected_token)
        _WRITE_BOUNDARY.reset(boundary_token)

exists(rel) classmethod

Return whether a storage path (relative, local, or S3 URL) exists.

Source code in src/cs_copilot/storage/client.py

@classmethod
def exists(cls, rel: str) -> bool:
    """Return whether a storage path (relative, local, or S3 URL) exists."""

    candidate = str(rel).strip()
    if cls._is_explicit_local_path(candidate):
        local = (
            unquote(urlsplit(candidate).path) if candidate.startswith("file://") else candidate
        )
        return Path(local).exists()
    if not cls._is_s3_url(candidate) and not is_s3_enabled():
        return cls._local_session_path(candidate).exists()
    url = candidate if cls._is_s3_url(candidate) else cls.path(candidate)
    filesystem, filesystem_path = fsspec.core.url_to_fs(
        url, **get_s3_config().to_storage_options()
    )
    filesystem.invalidate_cache(filesystem_path.rsplit("/", 1)[0])
    return bool(filesystem.exists(filesystem_path))

first_free_path(rel) classmethod

Return rel, or its first free <stem>-v<N><suffix> sibling.

Workflow outputs are immutable once registered and confined writes are create-only, so re-running a step writes a new version of each output instead of overwriting the previous one. The same spelling (relative, local, or S3 URL) is returned.

Source code in src/cs_copilot/storage/client.py

@classmethod
def first_free_path(cls, rel: str) -> str:
    """Return ``rel``, or its first free ``<stem>-v<N><suffix>`` sibling.

    Workflow outputs are immutable once registered and confined writes are
    create-only, so re-running a step writes a new version of each output
    instead of overwriting the previous one. The same spelling (relative,
    local, or S3 URL) is returned.
    """

    if not cls.exists(rel):
        return rel
    version = 2
    while cls.exists(_versioned_name(rel, version)):
        version += 1
    return _versioned_name(rel, version)

latest_version_path(rel) classmethod

Return the newest existing version of rel written by :meth:first_free_path.

Readers that locate an output by its deterministic name use this to find the most recent re-run. rel itself is returned when it has no later version (or does not exist).

Source code in src/cs_copilot/storage/client.py

@classmethod
def latest_version_path(cls, rel: str) -> str:
    """Return the newest existing version of ``rel`` written by :meth:`first_free_path`.

    Readers that locate an output by its deterministic name use this to
    find the most recent re-run. ``rel`` itself is returned when it has no
    later version (or does not exist).
    """

    latest = rel
    if not cls.exists(rel):
        return rel
    version = 2
    while cls.exists(_versioned_name(rel, version)):
        latest = _versioned_name(rel, version)
        version += 1
    return latest

session_relative(path) classmethod

Return the session-relative key of a path in the active session, if any.

Source code in src/cs_copilot/storage/client.py

@classmethod
def session_relative(cls, path: str) -> str | None:
    """Return the session-relative key of a path in the active session, if any."""

    return cls._verified_read_key(path)

current_write_boundary() staticmethod

Return the active write boundary, or None outside guarded execution.

Source code in src/cs_copilot/storage/client.py

@staticmethod
def current_write_boundary() -> str | None:
    """Return the active write boundary, or ``None`` outside guarded execution."""

    return _WRITE_BOUNDARY.get()

current_write_protected_paths() staticmethod

Return immutable session-relative paths for the active tool call.

Source code in src/cs_copilot/storage/client.py

@staticmethod
def current_write_protected_paths() -> frozenset[str]:
    """Return immutable session-relative paths for the active tool call."""

    return _WRITE_PROTECTED_PATHS.get()

stage_publications(staging_id) classmethod

Redirect scoped writes to a worker-private publication namespace.

Source code in src/cs_copilot/storage/client.py

@classmethod
@contextmanager
def stage_publications(cls, staging_id: str) -> Iterator[None]:
    """Redirect scoped writes to a worker-private publication namespace."""

    normalized_id = cls._normalize_relative_path(staging_id)
    if len(PurePosixPath(normalized_id).parts) != 1:
        raise ValueError("staging_id must be one safe path component")
    publications: dict[str, str] = {}
    id_token = _EXTERNAL_STAGING_ID.set(normalized_id)
    publications_token = _EXTERNAL_STAGED_PUBLICATIONS.set(publications)
    try:
        yield
    finally:
        _EXTERNAL_STAGED_PUBLICATIONS.reset(publications_token)
        _EXTERNAL_STAGING_ID.reset(id_token)

current_staged_publications() staticmethod

Return final-to-staged keys created by the active worker invocation.

Source code in src/cs_copilot/storage/client.py

@staticmethod
def current_staged_publications() -> dict[str, str]:
    """Return final-to-staged keys created by the active worker invocation."""

    return dict(_EXTERNAL_STAGED_PUBLICATIONS.get() or {})

staged_publication_metadata() classmethod

Checksum committed worker staging objects through a no-follow read.

Source code in src/cs_copilot/storage/client.py

@classmethod
def staged_publication_metadata(cls) -> dict[str, dict[str, Any]]:
    """Checksum committed worker staging objects through a no-follow read."""

    metadata: dict[str, dict[str, Any]] = {}
    for final_key, staged_key in cls.current_staged_publications().items():
        digest = hashlib.sha256()
        size = 0
        if is_s3_enabled():
            config = get_s3_config()
            source = fsspec.open(
                cls.path(staged_key),
                mode="rb",
                **config.to_storage_options(),
            ).open()
        else:
            source = cls._open_local_verified_source(staged_key)
        with source:
            while True:
                chunk = source.read(1024 * 1024)
                if not chunk:
                    break
                if isinstance(chunk, str):
                    chunk = chunk.encode("utf-8")
                digest.update(chunk)
                size += len(chunk)
        metadata[final_key] = {
            "staged_path": staged_key,
            "sha256": digest.hexdigest(),
            "size_bytes": size,
        }
    return metadata

promote_staged_publications(publications) classmethod

Promote complete worker outputs to final create-only artifact keys.

Source code in src/cs_copilot/storage/client.py

@classmethod
def promote_staged_publications(
    cls,
    publications: Mapping[str, Mapping[str, Any]],
) -> None:
    """Promote complete worker outputs to final create-only artifact keys."""

    promoted: dict[str, Mapping[str, Any]] = {}
    try:
        for final_key, metadata in publications.items():
            final = cls._normalize_relative_path(final_key)
            staged = cls._normalize_relative_path(str(metadata["staged_path"]))
            source = cls._open_verified_snapshot(
                staged,
                "rb",
                expected_sha256=str(metadata["sha256"]),
                expected_size=int(metadata["size_bytes"]),
            )
            with source:
                with cls.open_atomic(final, "xb") as destination:
                    while True:
                        chunk = source.read(1024 * 1024)
                        if not chunk:
                            break
                        destination.write(chunk)
            promoted[final] = metadata
        for metadata in publications.values():
            cls._remove_storage_key(cls._normalize_relative_path(str(metadata["staged_path"])))
    except BaseException:
        cls.rollback_promoted_publications(promoted)
        raise

rollback_promoted_publications(publications) classmethod

Remove only finals that still match the worker's promoted bytes.

Source code in src/cs_copilot/storage/client.py

@classmethod
def rollback_promoted_publications(
    cls,
    publications: Mapping[str, Mapping[str, Any]],
) -> None:
    """Remove only finals that still match the worker's promoted bytes."""

    for final_key, metadata in reversed(tuple(publications.items())):
        final = cls._normalize_relative_path(final_key)
        try:
            cls._remove_storage_key_verified(
                final,
                expected_sha256=str(metadata["sha256"]),
                expected_size=int(metadata["size_bytes"]),
            )
        except FileNotFoundError:
            continue
        except Exception:
            logger.warning(
                "Refusing to roll back changed worker publication %s",
                final,
                exc_info=True,
            )
            continue

discard_staging_prefix(boundary, staging_id) classmethod

Remove one exact worker staging subtree after failure or cancellation.

Source code in src/cs_copilot/storage/client.py

@classmethod
def discard_staging_prefix(cls, boundary: str, staging_id: str) -> None:
    """Remove one exact worker staging subtree after failure or cancellation."""

    normalized_boundary = cls._normalize_relative_path(boundary)
    normalized_id = cls._normalize_relative_path(staging_id)
    if len(PurePosixPath(normalized_id).parts) != 1:
        raise ValueError("staging_id must be one safe path component")
    prefix = PurePosixPath(
        normalized_boundary,
        ".staging",
        normalized_id,
    ).as_posix()
    if is_s3_enabled():
        config = get_s3_config()
        fs, fs_path = fsspec.core.url_to_fs(
            cls.path(prefix),
            **config.to_storage_options(),
        )
        if fs.exists(fs_path):
            fs.rm(fs_path, recursive=True)
        return
    _secure_local_rmtree(cls._local_session_path(prefix).absolute())

confine_artifact_reads(artifacts) classmethod

Pin registered artifact reads to checksum-verified byte snapshots.

Source code in src/cs_copilot/storage/client.py

@classmethod
@contextmanager
def confine_artifact_reads(
    cls,
    artifacts: Mapping[str, tuple[str, int]],
) -> Iterator[None]:
    """Pin registered artifact reads to checksum-verified byte snapshots."""

    inherited = dict(_VERIFIED_ARTIFACT_READS.get() or {})
    for path, expected in artifacts.items():
        normalized = cls._normalize_relative_path(path)
        if (
            not isinstance(expected, tuple)
            or len(expected) != 2
            or not isinstance(expected[0], str)
            or not isinstance(expected[1], int)
            or expected[1] < 0
        ):
            raise ValueError("verified artifact metadata must be (sha256, size_bytes)")
        inherited[normalized] = (expected[0], expected[1])
    token = _VERIFIED_ARTIFACT_READS.set(inherited)
    try:
        yield
    finally:
        _VERIFIED_ARTIFACT_READS.reset(token)

current_verified_artifact_reads() staticmethod

Return a detached copy of invocation-pinned artifact read metadata.

Source code in src/cs_copilot/storage/client.py

@staticmethod
def current_verified_artifact_reads() -> dict[str, tuple[str, int]]:
    """Return a detached copy of invocation-pinned artifact read metadata."""

    return dict(_VERIFIED_ARTIFACT_READS.get() or {})

path(rel) classmethod

Convert a relative path to an S3 URL or local session path.

Explicit S3 and local paths are returned unchanged. Relative paths are resolved against the active backend.

Parameters:

Name	Type	Description	Default
rel	str	Relative path or absolute S3 URL	required

Returns:

Name	Type	Description
str	str	Full S3 URL or local path

Examples:

S3.path("data.csv") 's3://chatbot-assets/sessions/20250121-123456-abc123/data.csv'

S3.path("s3://mybucket/data.csv") 's3://mybucket/data.csv'

Source code in src/cs_copilot/storage/client.py

@classmethod
def path(cls, rel: str) -> str:
    """
    Convert a relative path to an S3 URL or local session path.

    Explicit S3 and local paths are returned unchanged. Relative paths are
    resolved against the active backend.

    Args:
        rel: Relative path or absolute S3 URL

    Returns:
        str: Full S3 URL or local path

    Examples:
    ---------
    >>> S3.path("data.csv")
    's3://chatbot-assets/sessions/20250121-123456-abc123/data.csv'

    >>> S3.path("s3://mybucket/data.csv")
    's3://mybucket/data.csv'
    """
    if cls._is_s3_url(rel) or cls._is_explicit_local_path(rel):
        return rel

    config = get_s3_config()
    if not is_s3_enabled():
        return os.fspath(cls._local_session_path(rel))

    key = PurePosixPath(cls.current_prefix().strip("/")) / str(rel).lstrip("/")
    return f"s3://{config.bucket_name}/{key.as_posix()}"

open(rel, mode='rb') classmethod

Open a file for reading or writing.

Supports three types of paths: 1. Absolute S3 URLs (s3://...) 2. Local absolute paths (/ or file://) 3. Relative paths (scoped to current session)

Parameters:

Name	Type	Description	Default
rel	str	File path (relative, absolute, or S3 URL)	required
mode	str	File mode ('r', 'w', 'rb', 'wb', etc.)	'rb'

Returns:

Type	Description
Any	File-like object opened with the specified mode

Examples:

Read from session-scoped S3 path

with S3.open("data.csv", "r") as f: ... df = pd.read_csv(f)

Write to session-scoped S3 path

with S3.open("output.csv", "w") as f: ... df.to_csv(f)

Read from absolute S3 URL

with S3.open("s3://bucket/data.csv", "r") as f: ... df = pd.read_csv(f)

Read from local file

with S3.open("/tmp/data.csv", "r") as f: ... df = pd.read_csv(f)

Source code in src/cs_copilot/storage/client.py

@classmethod
def open(cls, rel: str, mode: str = "rb") -> Any:
    """
    Open a file for reading or writing.

    Supports three types of paths:
    1. Absolute S3 URLs (s3://...)
    2. Local absolute paths (/ or file://)
    3. Relative paths (scoped to current session)

    Args:
        rel: File path (relative, absolute, or S3 URL)
        mode: File mode ('r', 'w', 'rb', 'wb', etc.)

    Returns:
        File-like object opened with the specified mode

    Examples:
    ---------
    >>> # Read from session-scoped S3 path
    >>> with S3.open("data.csv", "r") as f:
    ...     df = pd.read_csv(f)

    >>> # Write to session-scoped S3 path
    >>> with S3.open("output.csv", "w") as f:
    ...     df.to_csv(f)

    >>> # Read from absolute S3 URL
    >>> with S3.open("s3://bucket/data.csv", "r") as f:
    ...     df = pd.read_csv(f)

    >>> # Read from local file
    >>> with S3.open("/tmp/data.csv", "r") as f:
    ...     df = pd.read_csv(f)
    """
    config = get_s3_config()
    write_mode = cls._is_write_mode(mode)
    if not write_mode:
        read_key = cls._verified_read_key(rel)
        staged = (
            _STAGED_ARTIFACT_WRITES.get().get(read_key)
            if read_key is not None and _STAGED_ARTIFACT_WRITES.get() is not None
            else None
        )
        if staged is not None:
            return staged._open_staged_read(mode)
        expected_reads = _VERIFIED_ARTIFACT_READS.get() or {}
        expected = expected_reads.get(read_key) if read_key else None
        if read_key is not None and expected is not None:
            return cls._open_verified_snapshot(
                read_key,
                mode,
                expected_sha256=expected[0],
                expected_size=expected[1],
            )

    write_boundary = cls.current_write_boundary() if write_mode else None
    if write_mode and write_boundary is None:
        observation = _WRITE_OBSERVATION.get()
        if observation is not None:
            cls._observe_write(observation, rel, mode)
    scoped_write_key: str | None = None
    if write_boundary is not None:
        scoped_write_key = cls._validate_scoped_write(rel, boundary=write_boundary)
    storage_target = rel
    staging_id = _EXTERNAL_STAGING_ID.get()
    publications = _EXTERNAL_STAGED_PUBLICATIONS.get()
    if scoped_write_key is not None and staging_id is not None and publications is not None:
        relative_to_boundary = PurePosixPath(scoped_write_key).relative_to(
            PurePosixPath(write_boundary or "")
        )
        staged_key = PurePosixPath(
            write_boundary or "",
            ".staging",
            staging_id,
            relative_to_boundary,
        ).as_posix()
        publications[scoped_write_key] = staged_key
        storage_target = staged_key

    if cls._is_s3_url(storage_target):
        if scoped_write_key is None:
            return fsspec.open(
                storage_target,
                mode=mode,
                **config.to_storage_options(),
            )
        return cls._open_confined_s3_path(
            storage_target,
            mode,
            storage_options=config.to_storage_options(),
            public_write_key=scoped_write_key,
        )

    if cls._is_explicit_local_path(storage_target):
        if storage_target.startswith("file://"):
            return fsspec.open(storage_target, mode=mode)
        return cls._open_local_path(Path(storage_target), mode)

    if not is_s3_enabled():
        return cls._open_local_path(
            cls._local_session_path(storage_target),
            mode,
            scoped_write_key=scoped_write_key,
        )

    if scoped_write_key is None:
        return fsspec.open(
            cls.path(storage_target),
            mode=mode,
            **config.to_storage_options(),
        )
    return cls._open_confined_s3_path(
        cls.path(storage_target),
        mode,
        storage_options=config.to_storage_options(),
        public_write_key=scoped_write_key,
    )

open_atomic(rel, mode='xb') classmethod

Create one session-relative object that is invisible until complete.

Source code in src/cs_copilot/storage/client.py

@classmethod
def open_atomic(cls, rel: str, mode: str = "xb"):
    """Create one session-relative object that is invisible until complete."""

    if not cls._is_write_mode(mode):
        raise ValueError("open_atomic requires a write-capable mode")
    if cls._is_s3_url(rel) or cls._is_explicit_local_path(rel):
        raise ValueError("open_atomic requires a session-relative storage path")
    normalized = cls._normalize_relative_path(rel)
    config = get_s3_config()
    if not is_s3_enabled():
        return cls._open_confined_local_path(
            cls._local_session_path(normalized),
            mode,
            boundary="",
            scoped_write_key=normalized,
            defer_transaction=False,
        )
    return cls._open_confined_s3_path(
        cls.path(normalized),
        mode,
        storage_options=config.to_storage_options(),
        defer_transaction=False,
        public_write_key=None,
    )

config

Storage configuration module.

Manages S3/MinIO connection settings and environment variable fallbacks.

StorageConfigError

Bases: ValueError

Raised when S3 is explicitly enabled but the runtime config is incomplete.

Source code in src/cs_copilot/storage/config.py

class StorageConfigError(ValueError):
    """Raised when S3 is explicitly enabled but the runtime config is incomplete."""

S3Config dataclass

Configuration for S3/MinIO storage.

Source code in src/cs_copilot/storage/config.py

@dataclass
class S3Config:
    """Configuration for S3/MinIO storage."""

    endpoint_url: Optional[str]
    access_key_id: Optional[str]
    secret_access_key: Optional[str]
    bucket_name: Optional[str]
    region_name: str = "us-east-1"
    use_s3: bool = False

    @classmethod
    def from_env(cls) -> "S3Config":
        """
        Create S3Config from environment variables.

        Environment Variables:
        ----------------------
        Endpoint (one of):
            - MINIO_ENDPOINT
            - MINIO_ENDPOINT_URL
            - S3_ENDPOINT_URL

        Access Key (one of):
            - MINIO_ACCESS_KEY
            - AWS_ACCESS_KEY_ID

        Secret Key (one of):
            - MINIO_SECRET_KEY
            - AWS_SECRET_ACCESS_KEY

        Bucket Name (one of):
            - ASSETS_BUCKET
            - S3_BUCKET_NAME

        Other:
            - AWS_REGION (default: us-east-1)
            - USE_S3 (default: false)

        Returns:
            S3Config: Configuration instance
        """
        endpoint = _getenv_nonempty("MINIO_ENDPOINT", "MINIO_ENDPOINT_URL", "S3_ENDPOINT_URL")

        minio_access_key = _getenv_nonempty("MINIO_ACCESS_KEY")
        minio_secret_key = _getenv_nonempty("MINIO_SECRET_KEY")
        aws_access_key = _getenv_nonempty("AWS_ACCESS_KEY_ID")
        aws_secret_key = _getenv_nonempty("AWS_SECRET_ACCESS_KEY")

        if endpoint:
            access_key = minio_access_key or aws_access_key
            secret_key = minio_secret_key or aws_secret_key
        else:
            access_key = aws_access_key
            secret_key = aws_secret_key

        bucket_name = _getenv_nonempty("ASSETS_BUCKET", "S3_BUCKET_NAME")
        region_name = os.getenv("AWS_REGION", "us-east-1")
        use_s3 = os.getenv("USE_S3", "false").lower() == "true"

        return cls(
            endpoint_url=endpoint,
            access_key_id=access_key,
            secret_access_key=secret_key,
            bucket_name=bucket_name,
            region_name=region_name,
            use_s3=use_s3,
        )

    def storage_backend(self) -> str:
        """
        Resolve the active storage backend.

        Returns:
            str: "local", "aws", or "s3-compatible"

        Raises:
            StorageConfigError: If S3 is explicitly enabled but missing required settings.
        """
        if not self.use_s3:
            return "local"

        missing_bucket = not self.bucket_name
        missing_key = not self.access_key_id
        missing_secret = not self.secret_access_key

        if self.endpoint_url:
            missing = []
            if missing_bucket:
                missing.append("ASSETS_BUCKET or S3_BUCKET_NAME")
            if missing_key:
                missing.append("MINIO_ACCESS_KEY or AWS_ACCESS_KEY_ID")
            if missing_secret:
                missing.append("MINIO_SECRET_KEY or AWS_SECRET_ACCESS_KEY")
            if missing:
                raise StorageConfigError(
                    "USE_S3=true with an S3-compatible endpoint requires "
                    + ", ".join(missing)
                    + "."
                )
            return "s3-compatible"

        missing = []
        if missing_bucket:
            missing.append("ASSETS_BUCKET or S3_BUCKET_NAME")
        if missing_key:
            missing.append("AWS_ACCESS_KEY_ID")
        if missing_secret:
            missing.append("AWS_SECRET_ACCESS_KEY")
        if missing:
            raise StorageConfigError(
                "USE_S3=true without an endpoint requires " + ", ".join(missing) + " for AWS S3."
            )
        return "aws"

    def to_storage_options(self) -> dict:
        """
        Convert config to fsspec/s3fs storage options.

        Returns:
            dict: Storage options for fsspec.open()
        """
        opts = {"config_kwargs": {"s3": {"addressing_style": "path"}}}

        if self.access_key_id:
            opts["key"] = self.access_key_id

        if self.secret_access_key:
            opts["secret"] = self.secret_access_key

        if self.endpoint_url:
            opts["client_kwargs"] = {"endpoint_url": self.endpoint_url}

        return opts

from_env() classmethod

Create S3Config from environment variables.

Environment Variables:

Endpoint (one of): - MINIO_ENDPOINT - MINIO_ENDPOINT_URL - S3_ENDPOINT_URL

Access Key (one of): - MINIO_ACCESS_KEY - AWS_ACCESS_KEY_ID

Secret Key (one of): - MINIO_SECRET_KEY - AWS_SECRET_ACCESS_KEY

Bucket Name (one of): - ASSETS_BUCKET - S3_BUCKET_NAME

Other

• AWS_REGION (default: us-east-1)
• USE_S3 (default: false)

Returns:

Name	Type	Description
S3Config	S3Config	Configuration instance

Source code in src/cs_copilot/storage/config.py

@classmethod
def from_env(cls) -> "S3Config":
    """
    Create S3Config from environment variables.

    Environment Variables:
    ----------------------
    Endpoint (one of):
        - MINIO_ENDPOINT
        - MINIO_ENDPOINT_URL
        - S3_ENDPOINT_URL

    Access Key (one of):
        - MINIO_ACCESS_KEY
        - AWS_ACCESS_KEY_ID

    Secret Key (one of):
        - MINIO_SECRET_KEY
        - AWS_SECRET_ACCESS_KEY

    Bucket Name (one of):
        - ASSETS_BUCKET
        - S3_BUCKET_NAME

    Other:
        - AWS_REGION (default: us-east-1)
        - USE_S3 (default: false)

    Returns:
        S3Config: Configuration instance
    """
    endpoint = _getenv_nonempty("MINIO_ENDPOINT", "MINIO_ENDPOINT_URL", "S3_ENDPOINT_URL")

    minio_access_key = _getenv_nonempty("MINIO_ACCESS_KEY")
    minio_secret_key = _getenv_nonempty("MINIO_SECRET_KEY")
    aws_access_key = _getenv_nonempty("AWS_ACCESS_KEY_ID")
    aws_secret_key = _getenv_nonempty("AWS_SECRET_ACCESS_KEY")

    if endpoint:
        access_key = minio_access_key or aws_access_key
        secret_key = minio_secret_key or aws_secret_key
    else:
        access_key = aws_access_key
        secret_key = aws_secret_key

    bucket_name = _getenv_nonempty("ASSETS_BUCKET", "S3_BUCKET_NAME")
    region_name = os.getenv("AWS_REGION", "us-east-1")
    use_s3 = os.getenv("USE_S3", "false").lower() == "true"

    return cls(
        endpoint_url=endpoint,
        access_key_id=access_key,
        secret_access_key=secret_key,
        bucket_name=bucket_name,
        region_name=region_name,
        use_s3=use_s3,
    )

storage_backend()

Resolve the active storage backend.

Returns:

Name	Type	Description
str	str	"local", "aws", or "s3-compatible"

Raises:

Type	Description
StorageConfigError	If S3 is explicitly enabled but missing required settings.

Source code in src/cs_copilot/storage/config.py

def storage_backend(self) -> str:
    """
    Resolve the active storage backend.

    Returns:
        str: "local", "aws", or "s3-compatible"

    Raises:
        StorageConfigError: If S3 is explicitly enabled but missing required settings.
    """
    if not self.use_s3:
        return "local"

    missing_bucket = not self.bucket_name
    missing_key = not self.access_key_id
    missing_secret = not self.secret_access_key

    if self.endpoint_url:
        missing = []
        if missing_bucket:
            missing.append("ASSETS_BUCKET or S3_BUCKET_NAME")
        if missing_key:
            missing.append("MINIO_ACCESS_KEY or AWS_ACCESS_KEY_ID")
        if missing_secret:
            missing.append("MINIO_SECRET_KEY or AWS_SECRET_ACCESS_KEY")
        if missing:
            raise StorageConfigError(
                "USE_S3=true with an S3-compatible endpoint requires "
                + ", ".join(missing)
                + "."
            )
        return "s3-compatible"

    missing = []
    if missing_bucket:
        missing.append("ASSETS_BUCKET or S3_BUCKET_NAME")
    if missing_key:
        missing.append("AWS_ACCESS_KEY_ID")
    if missing_secret:
        missing.append("AWS_SECRET_ACCESS_KEY")
    if missing:
        raise StorageConfigError(
            "USE_S3=true without an endpoint requires " + ", ".join(missing) + " for AWS S3."
        )
    return "aws"

to_storage_options()

Convert config to fsspec/s3fs storage options.

Returns:

Name	Type	Description
dict	dict	Storage options for fsspec.open()

Source code in src/cs_copilot/storage/config.py

def to_storage_options(self) -> dict:
    """
    Convert config to fsspec/s3fs storage options.

    Returns:
        dict: Storage options for fsspec.open()
    """
    opts = {"config_kwargs": {"s3": {"addressing_style": "path"}}}

    if self.access_key_id:
        opts["key"] = self.access_key_id

    if self.secret_access_key:
        opts["secret"] = self.secret_access_key

    if self.endpoint_url:
        opts["client_kwargs"] = {"endpoint_url": self.endpoint_url}

    return opts

get_s3_config()

Get S3 configuration from environment variables.

This is evaluated at call time (not import time) so that notebooks can call load_dotenv() after importing modules and still have up-to-date credentials and endpoint settings.

Returns:

Name	Type	Description
S3Config	S3Config	Current S3 configuration

Source code in src/cs_copilot/storage/config.py

def get_s3_config() -> S3Config:
    """
    Get S3 configuration from environment variables.

    This is evaluated at call time (not import time) so that notebooks can
    call load_dotenv() after importing modules and still have up-to-date
    credentials and endpoint settings.

    Returns:
        S3Config: Current S3 configuration
    """
    return S3Config.from_env()

is_s3_enabled()

Check if S3 is enabled based on configuration.

Returns:

Name	Type	Description
bool	bool	True if S3 should be used

Source code in src/cs_copilot/storage/config.py

def is_s3_enabled() -> bool:
    """
    Check if S3 is enabled based on configuration.

    Returns:
        bool: True if S3 should be used
    """
    return get_s3_config().storage_backend() != "local"

layout

Workflow-run output layout helpers.

The storage prefix identifies a chat/session. A session can contain multiple workflow runs, so run identity is represented independently and persisted in session_state[OUTPUT_CONTEXT_KEY] (and a context variable for direct tool calls that do not receive session state).

OutputOperation

Bases: str, Enum

Top-level operation folders inside a workflow run.

Source code in src/cs_copilot/storage/layout.py

class OutputOperation(str, Enum):
    """Top-level operation folders inside a workflow run."""

    CHEMICAL_SPACE = "01_chemical_space"
    ANALOG_GENERATION = "02_analog_generation"
    RETROSYNTHESIS = "03_retrosynthesis"
    REPORTS = "reports"

OutputLayout dataclass

Structured output path builder for one workflow run.

Source code in src/cs_copilot/storage/layout.py

@dataclass(frozen=True)
class OutputLayout:
    """Structured output path builder for one workflow run."""

    session_id: str
    run_id: str
    workflow_slug: str

    def __post_init__(self) -> None:
        object.__setattr__(
            self,
            "session_id",
            validate_identifier(self.session_id, field="session_id"),
        )
        object.__setattr__(
            self,
            "run_id",
            validate_identifier(self.run_id, field="run_id"),
        )
        object.__setattr__(
            self,
            "workflow_slug",
            sanitize_workflow_slug(self.workflow_slug),
        )

    @property
    def run_root(self) -> str:
        return PurePosixPath(WORKFLOWS_DIR, self.run_id).as_posix()

    def rel_path(self, operation: OutputOperation, *parts: str) -> str:
        cleaned_parts = [sanitize_path_part(part) for part in parts if str(part).strip()]
        return PurePosixPath(self.run_root, operation.value, *cleaned_parts).as_posix()

    @property
    def manifest_rel_path(self) -> str:
        return PurePosixPath(self.run_root, "manifest.json").as_posix()

    @property
    def artifact_index_rel_path(self) -> str:
        return PurePosixPath(self.run_root, "artifacts", "index.json").as_posix()

    @property
    def events_rel_path(self) -> str:
        return PurePosixPath(self.run_root, "events").as_posix()

    def event_rel_path(self, event_id: str) -> str:
        return PurePosixPath(
            self.events_rel_path,
            f"{validate_identifier(event_id, field='event_id')}.jsonl",
        ).as_posix()

    def artifact_rel_path(self, path: str) -> str:
        """Resolve and validate a path relative to this run's root."""

        relative = normalize_run_relative_path(self.run_id, path)
        return PurePosixPath(self.run_root, relative).as_posix()

artifact_rel_path(path)

Resolve and validate a path relative to this run's root.

Source code in src/cs_copilot/storage/layout.py

def artifact_rel_path(self, path: str) -> str:
    """Resolve and validate a path relative to this run's root."""

    relative = normalize_run_relative_path(self.run_id, path)
    return PurePosixPath(self.run_root, relative).as_posix()

open_local_run_artifact(layout, relative_path)

Open a local artifact for reading without following symlinks.

The workflow root and every descendant component are opened by descriptor. Each child lookup is relative to its already-open parent, so replacing a directory between validation and the final open cannot redirect the read. This helper intentionally fails closed on platforms without openat and O_NOFOLLOW support.

Source code in src/cs_copilot/storage/layout.py

def open_local_run_artifact(layout: OutputLayout, relative_path: str) -> BinaryIO:
    """Open a local artifact for reading without following symlinks.

    The workflow root and every descendant component are opened by descriptor.
    Each child lookup is relative to its already-open parent, so replacing a
    directory between validation and the final open cannot redirect the read.
    This helper intentionally fails closed on platforms without ``openat`` and
    ``O_NOFOLLOW`` support.
    """

    relative = normalize_run_relative_path(layout.run_id, relative_path)
    if not _OPEN_SUPPORTS_DIR_FD or not _O_NOFOLLOW:
        raise RuntimeError(
            "secure local artifact reads require directory-relative no-follow support"
        )

    common_flags = os.O_RDONLY | getattr(os, "O_CLOEXEC", 0) | _O_NOFOLLOW
    directory_flags = common_flags | getattr(os, "O_DIRECTORY", 0)
    current_fd: int | None = None
    file_fd: int | None = None
    try:
        current_fd = _open_existing_directory_tree(Path(S3.path(layout.run_root)).absolute())
        parts = PurePosixPath(relative).parts
        for component in parts[:-1]:
            next_fd = os.open(component, directory_flags, dir_fd=current_fd)
            os.close(current_fd)
            current_fd = next_fd
        file_fd = os.open(parts[-1], common_flags, dir_fd=current_fd)
        metadata = os.fstat(file_fd)
        if not stat.S_ISREG(metadata.st_mode) or metadata.st_nlink != 1:
            raise ValueError("artifact path must identify a regular, non-linked file")
        handle = os.fdopen(file_fd, "rb")
        file_fd = None
        return handle
    except OSError as exc:
        if isinstance(exc, PermissionError) or exc.errno in {errno.ELOOP, errno.ENOTDIR}:
            raise ValueError(
                f"artifact path resolves outside workflow run {layout.run_id!r} "
                "or contains a symlink"
            ) from exc
        raise
    finally:
        if file_fd is not None:
            os.close(file_fd)
        if current_fd is not None:
            os.close(current_fd)

sanitize_path_part(value, *, default='artifact')

Return a safe single path component while preserving useful suffix dots.

Source code in src/cs_copilot/storage/layout.py

def sanitize_path_part(value: Any, *, default: str = "artifact") -> str:
    """Return a safe single path component while preserving useful suffix dots."""

    text = str(value or "").replace("\\", "/").strip().strip("/")
    text = PurePosixPath(text).name if "/" in text else text
    text = _SAFE_PART_RE.sub("_", text).strip("._-")
    return text or default

validate_identifier(value, *, field='identifier')

Validate an externally supplied identifier without silently rewriting it.

Source code in src/cs_copilot/storage/layout.py

def validate_identifier(value: Any, *, field: str = "identifier") -> str:
    """Validate an externally supplied identifier without silently rewriting it."""

    text = str(value or "").strip()
    if not _IDENTIFIER_RE.fullmatch(text):
        raise ValueError(
            f"{field} must start with an alphanumeric character and contain only "
            "letters, digits, '.', '_' or '-' (maximum 128 characters)"
        )
    return text

normalize_run_relative_path(run_id, path)

Return a safe path relative to workflows/<run_id>.

Absolute paths, storage URLs, traversal components, paths belonging to another run, and runtime-control files are rejected. Artifact registration uses this function as its workflow-root containment boundary.

Source code in src/cs_copilot/storage/layout.py

def normalize_run_relative_path(run_id: str, path: str) -> str:
    """Return a safe path relative to ``workflows/<run_id>``.

    Absolute paths, storage URLs, traversal components, paths belonging to
    another run, and runtime-control files are rejected. Artifact registration
    uses this function as its workflow-root containment boundary.
    """

    run_id = validate_identifier(run_id, field="run_id")
    if not isinstance(path, str) or not path.strip():
        raise ValueError("artifact path cannot be empty")
    value = path.strip()
    if is_explicit_storage_path(value) or "\\" in value:
        raise ValueError("artifact path must be relative to the workflow run")

    candidate = PurePosixPath(value)
    if candidate.is_absolute() or any(part in {"", ".", ".."} for part in candidate.parts):
        raise ValueError("artifact path must not be absolute or contain traversal components")

    run_prefix = PurePosixPath(WORKFLOWS_DIR, run_id)
    if candidate.parts and candidate.parts[0] == WORKFLOWS_DIR:
        try:
            candidate = candidate.relative_to(run_prefix)
        except ValueError as exc:
            raise ValueError(f"artifact path is outside workflow run {run_id!r}") from exc

    if not candidate.parts:
        raise ValueError("artifact path must identify a file")
    if candidate == PurePosixPath("manifest.json"):
        raise ValueError("manifest.json is reserved for the workflow runtime")
    if candidate.parts[0] in {".staging", "events"}:
        raise ValueError(f"{candidate.parts[0]}/ is reserved for the workflow runtime")
    if candidate == PurePosixPath("artifacts", "index.json"):
        raise ValueError("artifacts/index.json is reserved for the workflow runtime")
    return candidate.as_posix()

ensure_output_context(session_state=None, *, workflow_slug=None, run_id=None, session_id=None)

Ensure and return explicit session/run/workflow output context.

Source code in src/cs_copilot/storage/layout.py

def ensure_output_context(
    session_state: Optional[dict[str, Any]] = None,
    *,
    workflow_slug: Optional[str] = None,
    run_id: Optional[str] = None,
    session_id: Optional[str] = None,
) -> dict[str, Any]:
    """Ensure and return explicit session/run/workflow output context."""

    resolved_session_id = validate_identifier(session_id or _session_id(), field="session_id")
    requested_slug = sanitize_workflow_slug(workflow_slug)

    candidates: list[dict[str, Any]] = []
    if isinstance(session_state, dict):
        existing = session_state.get(OUTPUT_CONTEXT_KEY)
        if isinstance(existing, dict):
            candidates.append(existing)
    contextual = _RUN_CONTEXT.get()
    if isinstance(contextual, dict):
        candidates.append(contextual)

    for existing in candidates:
        if not _valid_context(existing, session_id=resolved_session_id):
            continue
        if run_id is not None and existing["run_id"] != run_id:
            continue
        # Legacy toolkit helpers pass operation labels such as
        # ``chemical_space`` and ``reports`` here. Once a run is active those
        # labels must not fork the experiment; only an explicit run_id selects
        # a different run.
        context = dict(existing)
        _RUN_CONTEXT.set(context)
        if isinstance(session_state, dict):
            session_state[OUTPUT_CONTEXT_KEY] = context
        return context

    if run_id is None and S3.current_write_boundary() is not None:
        # Minting a run here would place every artifact under a brand-new run
        # root, which can never sit inside the boundary the caller opened.
        raise ScopedWriteDenied(
            "no workflow run is reachable for this scoped write; "
            "the caller must bind the run that owns the active write boundary",
            verdict="outside_boundary",
        )
    resolved_run_id = (
        validate_identifier(run_id, field="run_id")
        if run_id is not None
        else f"{requested_slug}-{uuid.uuid4().hex[:12]}"
    )
    context = {
        "layout_version": LAYOUT_VERSION,
        "session_id": resolved_session_id,
        "run_id": resolved_run_id,
        "workflow_slug": requested_slug,
    }
    _RUN_CONTEXT.set(context)
    if isinstance(session_state, dict):
        session_state[OUTPUT_CONTEXT_KEY] = context
    return context

set_output_context(context)

Validate and activate a serialized output context.

Source code in src/cs_copilot/storage/layout.py

def set_output_context(context: dict[str, Any]) -> dict[str, Any]:
    """Validate and activate a serialized output context."""

    required = {"layout_version", "session_id", "run_id", "workflow_slug"}
    missing = sorted(required - set(context))
    if missing:
        raise ValueError(f"output context is missing fields: {', '.join(missing)}")
    if context["layout_version"] != LAYOUT_VERSION:
        raise ValueError(
            f"unsupported output layout version {context['layout_version']!r}; "
            f"expected {LAYOUT_VERSION}"
        )
    normalized = {
        "layout_version": LAYOUT_VERSION,
        "session_id": validate_identifier(context["session_id"], field="session_id"),
        "run_id": validate_identifier(context["run_id"], field="run_id"),
        "workflow_slug": sanitize_workflow_slug(context["workflow_slug"]),
    }
    _RUN_CONTEXT.set(normalized)
    return normalized

active_output_context(context)

Pin one run context for the enclosing scope, restoring the previous one.

Source code in src/cs_copilot/storage/layout.py

@contextmanager
def active_output_context(context: dict[str, Any]) -> Iterator[dict[str, Any]]:
    """Pin one run context for the enclosing scope, restoring the previous one."""

    token = _RUN_CONTEXT.set(None)
    try:
        yield set_output_context(context)
    finally:
        _RUN_CONTEXT.reset(token)

scoped_artifact_path(path, operation, *folders, session_state=None, workflow_slug=None, run_id=None)

Scope an ordinary relative artifact path into the workflow-run layout.

Source code in src/cs_copilot/storage/layout.py

def scoped_artifact_path(
    path: str,
    operation: OutputOperation,
    *folders: str,
    session_state: Optional[dict[str, Any]] = None,
    workflow_slug: Optional[str] = None,
    run_id: Optional[str] = None,
) -> str:
    """Scope an ordinary relative artifact path into the workflow-run layout."""

    if is_explicit_storage_path(path) or is_workflow_scoped_path(path):
        return path
    filename = sanitize_path_part(path)
    return operation_rel_path(
        operation,
        *folders,
        filename,
        session_state=session_state,
        workflow_slug=workflow_slug,
        run_id=run_id,
    )