Storage seam

The fsspec-backed IO layer that makes the framework cloud-agnostic (ADR-0009) — file://, s3://, gcs://, az:// are one line of config apart.

storage

fsspec-backed Storage seam (ADR-0009).

All framework IO — input documents, inter-Stage records, checkpoints, exports, the Quality Report — flows through this one abstraction. Local disk, S3, GCS, Azure Blob, and in-memory are addressed by URI scheme with no framework code change; credentials resolve through fsspec's own default chains, so the framework never stores secrets.

SchemeNotAllowedError

Bases: ConfigError

Raised when a Storage is built for a URI scheme not on the allowlist.

A :class:~latence_core.errors.ConfigError (a disallowed scheme is an operator config mistake, #73) — and, transitively, still a ValueError, so the pre-taxonomy except ValueError / except SchemeNotAllowedError guards keep catching it.

Storage

Storage(
    protocol: str = "file",
    *,
    allowed_schemes: Iterable[str] | None = None,
    **storage_options: Any
)

Thin, URI-scheme-agnostic IO surface over fsspec.

A :class:Storage is bound to a single filesystem (derived from the scheme of the URIs it is asked to operate on). Mixing schemes on one instance is a programming error and raises ValueError.

Source code in packages/latence-core/src/latence_core/storage.py

def __init__(
    self,
    protocol: str = "file",
    *,
    allowed_schemes: Iterable[str] | None = None,
    **storage_options: Any,
) -> None:
    allowed = (
        frozenset(allowed_schemes) if allowed_schemes is not None else DEFAULT_ALLOWED_SCHEMES
    )
    if protocol not in allowed:
        permitted = ", ".join(sorted(allowed))
        msg = (
            f"Storage scheme {protocol!r} is not permitted. "
            f"Allowed schemes: {permitted}. "
            f"Pass allowed_schemes=[...] to widen this at your own risk."
        )
        raise SchemeNotAllowedError(msg)
    self._protocol = protocol
    self._allowed_schemes = allowed
    self._fs: fsspec.AbstractFileSystem = fsspec.filesystem(protocol, **storage_options)

for_uri classmethod

for_uri(
    uri: str,
    *,
    allowed_schemes: Iterable[str] | None = None,
    **storage_options: Any
) -> Storage

Build a Storage bound to the filesystem implied by uri's scheme.

Source code in packages/latence-core/src/latence_core/storage.py

@classmethod
def for_uri(
    cls,
    uri: str,
    *,
    allowed_schemes: Iterable[str] | None = None,
    **storage_options: Any,
) -> Storage:
    """Build a Storage bound to the filesystem implied by ``uri``'s scheme."""
    protocol = cls._protocol_of(uri)
    return cls(protocol=protocol, allowed_schemes=allowed_schemes, **storage_options)

write_atomic

write_atomic(uri: str, data: bytes) -> None

Write data to uri so a reader never sees a partial file.

The buffered face of :meth:atomic_writer — same publish-or-nothing guarantee, same temp-file sweep on failure, for a payload that already fits in one bytes buffer. This is the WAL discipline the Runner relies on for durable, resumable checkpoints (ADR-0010): the destination either does not exist or is the complete previous/next version, never a torn write, and a failure leaves no .tmp sibling behind.

See :meth:atomic_writer for what "atomic" does and does not mean per backend.

Source code in packages/latence-core/src/latence_core/storage.py

def write_atomic(self, uri: str, data: bytes) -> None:
    """Write ``data`` to ``uri`` so a reader never sees a partial file.

    The buffered face of :meth:`atomic_writer` — same publish-or-nothing guarantee, same
    temp-file sweep on failure, for a payload that already fits in one ``bytes`` buffer.
    This is the WAL discipline the Runner relies on for durable, resumable checkpoints
    (ADR-0010): the destination either does not exist or is the complete previous/next
    version, never a torn write, and a failure leaves no ``.tmp`` sibling behind.

    See :meth:`atomic_writer` for what "atomic" does and does not mean per backend.
    """
    with self.atomic_writer(uri) as fh:
        fh.write(data)

atomic_writer

atomic_writer(
    uri: str,
    *,
    destination: Callable[[], str | None] | None = None
) -> Iterator[IO[bytes]]

Stream bytes into uri: publish on clean exit, sweep the partial write on failure.

The one-call streaming write-atomic primitive (ADR-0010/0033). It opens a unique temp sibling of uri, creates uri's parent directory, yields the open binary handle, and on clean exit closes it and renames it onto uri. A reader never sees the temp path under uri and never sees a torn uri — the same guarantee as :meth:write_atomic, without materialising the payload in memory first::

with storage.atomic_writer(uri) as fh:
    for batch in batches:
        fh.write(batch)

The cleanup obligation lives here, not in the caller. The two-call :meth:temp_uri / :meth:finalize_atomic protocol this replaces pushed onto every caller a duty its signature never stated: sweep the temp path when the write dies between the two calls. One of three callers actually did it, and the cost of the other two was silent: each temp name carries a fresh random token, so on a flaky store every retry stranded another full-size copy — unbounded growth of unreferenced, billable objects that no lifecycle policy and no expire_snapshots knows about, still holding the run's (possibly sensitive) bytes. On any exception the temp file is removed best-effort and the original error is re-raised untouched (ADR-0034: the failure is never masked by its own cleanup).

The sweep is reported, not silent: the propagating exception carries PEP-678 notes (exc.__notes__) — one stating what the publication published (for the single-file face, always nothing), and one per temp file saying either that it was removed or — when the store refused the delete — naming the exact path that still holds the bytes so an operator can delete it by hand. A caller that wraps the failure in its own typed error can fold those notes into its message.

destination is for a content-addressed artifact whose final name is only knowable once its bytes have been written (the Iceberg data file, named after the sha256 of its own content). It is called after the handle is closed and its return value replaces uri as the publish target; returning None publishes nothing and sweeps the temp file (the "an identical file is already there, leave it alone" case). uri still names the temp sibling, so the in-flight write lands next to where it is going.

What "atomic" means per backend — this is interface, not trivia:

• Local / POSIX: fsspec's rename maps to os.replace, a genuinely atomic same-filesystem operation.
• Object stores (S3, GCS, Azure): there is no rename. fsspec's mv is a server-side copy followed by a delete — two operations, not one atomic one. What still holds is the guarantee callers actually depend on: a single-object write is atomic and read-after-write consistent on all three stores, so a reader of uri sees either the old object or the whole new one, never a torn one. What does not hold is the rest of the rename contract: a crash between the copy and the delete leaves the temp object behind (this method's sweep is best-effort and cannot run if the process is killed), and two publishers racing onto one uri both succeed with a last-writer-wins result rather than one losing. Sizes matter too: S3 caps a single CopyObject at 5 GB, so a larger artifact is copied as a multipart upload whose failure can leave an incomplete multipart upload the bucket's lifecycle policy — not this method — has to reap.

Source code in packages/latence-core/src/latence_core/storage.py

@contextmanager
def atomic_writer(
    self,
    uri: str,
    *,
    destination: Callable[[], str | None] | None = None,
) -> Iterator[IO[bytes]]:
    """Stream bytes into ``uri``: publish on clean exit, sweep the partial write on failure.

    The one-call streaming write-atomic primitive (ADR-0010/0033). It opens a unique temp
    sibling of ``uri``, creates ``uri``'s parent directory, yields the open binary handle,
    and on clean exit closes it and renames it onto ``uri``. A reader never sees the temp
    path under ``uri`` and never sees a torn ``uri`` — the same guarantee as
    :meth:`write_atomic`, without materialising the payload in memory first::

        with storage.atomic_writer(uri) as fh:
            for batch in batches:
                fh.write(batch)

    **The cleanup obligation lives here, not in the caller.** The two-call
    :meth:`temp_uri` / :meth:`finalize_atomic` protocol this replaces pushed onto every
    caller a duty its signature never stated: sweep the temp path when the write dies
    between the two calls. One of three callers actually did it, and the cost of the other
    two was silent: each temp name carries a fresh random token, so on a flaky store *every
    retry stranded another full-size copy* — unbounded growth of unreferenced, billable
    objects that no lifecycle policy and no ``expire_snapshots`` knows about, still holding
    the run's (possibly sensitive) bytes. On any exception the temp file is removed
    best-effort and the **original error is re-raised untouched** (ADR-0034: the failure is
    never masked by its own cleanup).

    The sweep is *reported, not silent*: the propagating exception carries PEP-678 notes
    (``exc.__notes__``) — one stating what the publication published (for the single-file face,
    always nothing), and one per temp file saying either that it was removed or — when the
    store refused the delete — naming the exact path that still holds the bytes so an operator
    can delete it by hand. A caller that wraps the failure in its own typed error can fold
    those notes into its message.

    ``destination`` is for a **content-addressed** artifact whose final name is only knowable
    once its bytes have been written (the Iceberg data file, named after the sha256 of its own
    content). It is called after the handle is closed and its return value replaces ``uri`` as
    the publish target; returning ``None`` publishes nothing and sweeps the temp file (the
    "an identical file is already there, leave it alone" case). ``uri`` still names the temp
    sibling, so the in-flight write lands next to where it is going.

    **What "atomic" means per backend** — this is interface, not trivia:

    * *Local / POSIX*: fsspec's rename maps to ``os.replace``, a genuinely atomic
      same-filesystem operation.
    * *Object stores (S3, GCS, Azure)*: there is no rename. fsspec's ``mv`` is a server-side
      **copy followed by a delete — two operations, not one atomic one.** What still holds is
      the guarantee callers actually depend on: a single-object write is atomic and
      read-after-write consistent on all three stores, so a reader of ``uri`` sees either the
      old object or the whole new one, never a torn one. What does *not* hold is the rest of
      the rename contract: a crash between the copy and the delete leaves the temp object
      behind (this method's sweep is best-effort and cannot run if the process is killed), and
      two publishers racing onto one ``uri`` both succeed with a last-writer-wins result rather
      than one losing. Sizes matter too: S3 caps a single ``CopyObject`` at 5 GB, so a larger
      artifact is copied as a multipart upload whose failure can leave an incomplete multipart
      upload the bucket's lifecycle policy — not this method — has to reap.
    """
    with self._atomic_publication([uri], [destination]) as handles:
        yield handles[0]

atomic_writers

atomic_writers(
    uris: Sequence[str],
) -> AbstractContextManager[tuple[IO[bytes], ...]]

Stream a set of sibling files, publishing all of them only once all are complete.

The multi-file face of :meth:atomic_writer, for a deliverable that is a group of files which must not appear one-by-one — the Export's records.jsonl + records.parquet (+ the optional multi-vector and BM25 sidecars), which a downstream reader treats as one corpus (ADR-0033/0050). Every handle is written to its own temp sibling; not one byte is published until every handle has closed cleanly, so a write error part-way through the corpus can no longer leave a published records.parquet next to a missing records.jsonl. On any exception every temp file is swept, exactly as :meth:atomic_writer sweeps its one.

Handles are yielded in the order of uris and must all be used inside the block::

with storage.atomic_writers([jsonl_uri, parquet_uri]) as (jsonl, parquet):
    ...

Honest limit: the publish phase is still N renames, not one transaction. No store the framework targets offers a cross-object commit, so a failure during the publish loop leaves the already-renamed siblings in place and sweeps the rest; the exception's notes name what landed. This is strictly stronger than publishing as you go and is the most any of file/S3/GCS/Azure can actually give — a retried run rewrites the whole set.

Source code in packages/latence-core/src/latence_core/storage.py

def atomic_writers(self, uris: Sequence[str]) -> AbstractContextManager[tuple[IO[bytes], ...]]:
    """Stream a **set** of sibling files, publishing all of them only once all are complete.

    The multi-file face of :meth:`atomic_writer`, for a deliverable that is a *group* of files
    which must not appear one-by-one — the Export's ``records.jsonl`` + ``records.parquet`` (+
    the optional multi-vector and BM25 sidecars), which a downstream reader treats as one
    corpus (ADR-0033/0050). Every handle is written to its own temp sibling; not one byte is
    published until *every* handle has closed cleanly, so a write error part-way through the
    corpus can no longer leave a published ``records.parquet`` next to a missing
    ``records.jsonl``. On any exception every temp file is swept, exactly as
    :meth:`atomic_writer` sweeps its one.

    Handles are yielded in the order of ``uris`` and must all be used inside the block::

        with storage.atomic_writers([jsonl_uri, parquet_uri]) as (jsonl, parquet):
            ...

    Honest limit: the publish phase is still N renames, not one transaction. No store the
    framework targets offers a cross-object commit, so a failure *during* the publish loop
    leaves the already-renamed siblings in place and sweeps the rest; the exception's notes
    name what landed. This is strictly stronger than publishing as you go and is the most any
    of file/S3/GCS/Azure can actually give — a retried run rewrites the whole set.
    """
    return self._atomic_publication(list(uris), [None] * len(uris))

temp_uri

temp_uri(uri: str) -> str

A unique sibling temp path for uri (the write side of :meth:finalize_atomic).

.. deprecated:: superseded by :meth:atomic_writer / :meth:atomic_writers.

This is half of a two-call protocol that leaks a cleanup obligation the interface never states — the caller, not this module, is left holding the duty to sweep the temp path when the write between the two calls fails. :meth:atomic_writer owns that duty. Retained only until the last remaining caller (the Runner's checkpoint writer) moves across; it is not part of the interface a new caller should learn.

Source code in packages/latence-core/src/latence_core/storage.py

def temp_uri(self, uri: str) -> str:
    """A unique sibling temp path for ``uri`` (the write side of :meth:`finalize_atomic`).

    .. deprecated:: superseded by :meth:`atomic_writer` / :meth:`atomic_writers`.

       This is half of a two-call protocol that leaks a cleanup obligation the interface
       never states — the caller, not this module, is left holding the duty to sweep the temp
       path when the write between the two calls fails. :meth:`atomic_writer` owns that duty.
       Retained only until the last remaining caller (the Runner's checkpoint writer) moves
       across; it is not part of the interface a new caller should learn.
    """
    return f"{uri}.{_token()}.tmp"

finalize_atomic

finalize_atomic(tmp_uri: str, uri: str) -> None

Atomically publish a fully-written temp file (from :meth:temp_uri) as uri.

The rename half of the two-call streamed write. See :meth:atomic_writer for what "atomic" actually means on each backend (on an object store this is a copy and a delete — two operations — not a rename).

.. deprecated:: superseded by :meth:atomic_writer / :meth:atomic_writers, which own the cleanup obligation this pairing pushes onto its callers.

Source code in packages/latence-core/src/latence_core/storage.py

def finalize_atomic(self, tmp_uri: str, uri: str) -> None:
    """Atomically publish a fully-written temp file (from :meth:`temp_uri`) as ``uri``.

    The rename half of the two-call streamed write. See :meth:`atomic_writer` for what
    "atomic" actually means on each backend (on an object store this is a copy **and** a
    delete — two operations — not a rename).

    .. deprecated:: superseded by :meth:`atomic_writer` / :meth:`atomic_writers`, which own
       the cleanup obligation this pairing pushes onto its callers.
    """
    self._fs.mv(self._check(tmp_uri), self._check(uri))

remove

remove(uri: str) -> None

Physically delete a single file (the Purge / GDPR-erasure primitive, ADR-0018).

Idempotent: removing a non-existent path is a no-op, so a purge that has already erased a file (or a file the caller only believes exists) does not raise. Unlike :meth:write_atomic this is a destructive, non-recoverable delete — used only on the Purge path where the S11 acceptance criterion requires that nothing derived survives (the source file, and the derived records expunged from prior Versions and checkpoints).

Source code in packages/latence-core/src/latence_core/storage.py

def remove(self, uri: str) -> None:
    """Physically delete a single file (the Purge / GDPR-erasure primitive, ADR-0018).

    Idempotent: removing a non-existent path is a no-op, so a purge that has already erased a
    file (or a file the caller only *believes* exists) does not raise. Unlike
    :meth:`write_atomic` this is a destructive, non-recoverable delete — used only on the
    Purge path where the S11 acceptance criterion requires that *nothing derived survives*
    (the source file, and the derived records expunged from prior Versions and checkpoints).
    """
    checked = self._check(uri)
    if self._fs.exists(checked):
        self._fs.rm_file(checked) if hasattr(self._fs, "rm_file") else self._fs.rm(checked)

remove_tree

remove_tree(uri: str) -> None

Recursively delete a directory and everything under it (the Purge erasure primitive).

Idempotent: a missing path is a no-op. Destructive and non-recoverable — used on the Purge path to wipe re-derivable run artifact trees (checkpoints, export outputs, run manifests) that would otherwise retain a purged document's raw text (the S11 "nothing derived survives" acceptance criterion).

Source code in packages/latence-core/src/latence_core/storage.py

def remove_tree(self, uri: str) -> None:
    """Recursively delete a directory and everything under it (the Purge erasure primitive).

    Idempotent: a missing path is a no-op. Destructive and non-recoverable — used on the Purge
    path to wipe re-derivable run artifact trees (checkpoints, export outputs, run manifests)
    that would otherwise retain a purged document's raw text (the S11 "nothing derived
    survives" acceptance criterion).
    """
    checked = self._check(uri)
    if self._fs.exists(checked):
        self._fs.rm(checked, recursive=True)

modified

modified(uri: str) -> datetime | None

The UTC last-modified time of uri, or None when the store does not expose one.

A purely read-only metadata probe (#186): the run console reads when a run's manifest and Quality Report landed straight off Storage, so a run list carries real timestamps without the framework persisting any new state (ADR-0010 — run state is the files that are already there). fsspec backends spell the field differently (mtime on local, LastModified on S3, updated on GCS, last_modified on Azure) and an in-memory store exposes none at all, so this normalises the known spellings and returns None rather than fabricating a time when the backend has nothing to give. A missing path is None too — the caller renders "unknown", it is never an error.

Source code in packages/latence-core/src/latence_core/storage.py

def modified(self, uri: str) -> datetime | None:
    """The UTC last-modified time of ``uri``, or ``None`` when the store does not expose one.

    A purely read-only metadata probe (#186): the run console reads *when* a run's manifest
    and Quality Report landed straight off Storage, so a run list carries real timestamps
    without the framework persisting any new state (ADR-0010 — run state is the files that
    are already there). fsspec backends spell the field differently (``mtime`` on local,
    ``LastModified`` on S3, ``updated`` on GCS, ``last_modified`` on Azure) and an in-memory
    store exposes none at all, so this normalises the known spellings and returns ``None``
    rather than fabricating a time when the backend has nothing to give. A missing path is
    ``None`` too — the caller renders "unknown", it is never an error.
    """
    try:
        info = self._fs.info(self._check(uri))
    except (FileNotFoundError, KeyError, OSError, ValueError):
        return None
    if not isinstance(info, dict):
        return None
    for key in ("mtime", "LastModified", "last_modified", "updated", "created"):
        value = info.get(key)
        stamp = _as_utc(value)
        if stamp is not None:
            return stamp
    return None

walk_files

walk_files(
    uri: str, follow_symlinks: bool = False
) -> Iterator[str]

Yield every file (not directory) under uri, recursively, sorted.

Every yielded path is a scheme-bearing URI the same Storage can read back: fsspec's walk strips the protocol and yields bare bucket/key (object stores) or bare filesystem paths (local), so each result is re-decorated with this Storage's protocol via unstrip_protocol before being yielded. Without this, feeding a walk_files result straight back into :meth:read_bytes on an object store raised ValueError (a bare bucket/key has no ://, so :meth:_protocol_of mis-reads it as file and :meth:_check rejects it against the s3/memory binding) — the H-E1 s3:// e2e blocker.

With follow_symlinks=False (the secure default) a file whose real, symlink-resolved path escapes uri's own real path is dropped, so a symlink planted inside the folder cannot be used to read files outside it. fsspec's own walk follows symlinks (its ls does), so the escape check is applied here rather than relying on a walk flag. Only the local filesystem has symlinks; on object/in-memory stores the filter is a no-op because os.path.realpath leaves non-local paths unchanged.

Memory characteristic — this is interface, not an implementation detail. Despite returning an Iterator, this is NOT a streaming listing: the whole recursive listing is accumulated and sorted before the first path is yielded, so peak memory is O(number of files under uri) and the first result costs a full prefix walk. That is deliberate and load-bearing rather than an oversight: the framework's determinism invariant requires a stable document order, and fsspec's walk inherits its per-directory order from ls (os.listdir order locally, the store's paging order remotely), which is not stable across hosts or runs. A globally sorted order cannot be produced without seeing every path, and a per-directory sort would yield a different (if deterministic) order than the global one, silently changing every downstream artifact. Callers pointing this at a bucket prefix with millions of objects should scope the prefix; a paged listing would be a different, order-weakening method, not a change to this one.

Source code in packages/latence-core/src/latence_core/storage.py

def walk_files(self, uri: str, follow_symlinks: bool = False) -> Iterator[str]:
    """Yield every file (not directory) under ``uri``, recursively, sorted.

    Every yielded path is a **scheme-bearing URI** the same Storage can read
    back: ``fsspec``'s ``walk`` strips the protocol and yields bare
    ``bucket/key`` (object stores) or bare filesystem paths (local), so each
    result is re-decorated with this Storage's protocol via
    ``unstrip_protocol`` before being yielded. Without this, feeding a
    ``walk_files`` result straight back into :meth:`read_bytes` on an object
    store raised ``ValueError`` (a bare ``bucket/key`` has no ``://``, so
    :meth:`_protocol_of` mis-reads it as ``file`` and :meth:`_check` rejects
    it against the ``s3``/``memory`` binding) — the H-E1 s3:// e2e blocker.

    With ``follow_symlinks=False`` (the secure default) a file whose real,
    symlink-resolved path escapes ``uri``'s own real path is dropped, so a
    symlink planted inside the folder cannot be used to read files outside
    it. fsspec's own ``walk`` follows symlinks (its ``ls`` does), so the
    escape check is applied here rather than relying on a walk flag. Only the
    local filesystem has symlinks; on object/in-memory stores the filter is a
    no-op because ``os.path.realpath`` leaves non-local paths unchanged.

    **Memory characteristic — this is interface, not an implementation detail.** Despite
    returning an ``Iterator``, this is NOT a streaming listing: the whole recursive listing is
    accumulated and sorted before the first path is yielded, so peak memory is O(number of
    files under ``uri``) and the first result costs a full prefix walk. That is deliberate and
    load-bearing rather than an oversight: the framework's determinism invariant requires a
    stable document order, and ``fsspec``'s ``walk`` inherits its per-directory order from
    ``ls`` (``os.listdir`` order locally, the store's paging order remotely), which is not
    stable across hosts or runs. A globally sorted order cannot be produced without seeing
    every path, and a per-directory sort would yield a *different* (if deterministic) order
    than the global one, silently changing every downstream artifact. Callers pointing this at
    a bucket prefix with millions of objects should scope the prefix; a paged listing would be
    a different, order-weakening method, not a change to this one.
    """
    checked = self._check(uri)
    # fsspec's local walk strips the URI scheme and yields bare filesystem
    # paths; resolve the root the same way so the escape check compares like
    # with like (a raw ``os.path.realpath('file://...')`` would be garbage).
    is_local = self._protocol == "file"
    root_bare = checked.split("://", 1)[1] if is_local and "://" in checked else checked
    root_real = os.path.realpath(root_bare) if is_local else checked
    found: list[str] = []
    for root, _dirs, files in self._fs.walk(checked):
        for name in files:
            path = f"{root.rstrip('/')}/{name}"
            # The symlink-escape check runs on the bare local path (that is
            # what ``os.path.realpath`` understands); re-decorate afterwards.
            if not follow_symlinks and is_local and self._escapes(path, root_real):
                continue
            # Yield a scheme-bearing URI so the caller can round-trip it back
            # through read_bytes/open under this Storage's binding (s3://,
            # memory://, file://) — not a bare, binding-mismatched path.
            found.append(self._fs.unstrip_protocol(path))
    yield from sorted(found)