Add laborious

laborious/utils/bundle/__init__.py

@@ -0,0 +1,27 @@
+"""Reading `.sientia` bundles: the wire format, the seven gates and the step vocabulary.
+
+There is no writer here. The producer is the Streamlit app in `sientia-projects-templates`
+(`app/src/operations/model_export/`), and a second production writer for a format with one producer
+would be a fork waiting to happen. What stands in for it in the tests is
+`tests/helpers/bundle_factory.py` plus the golden fixtures under `tests/fixtures/bundle/`.
+"""
+
+from laborious.utils.bundle.format import BundleFormatError, BundleHeader
+from laborious.utils.bundle.reader import (
+    BundleGateError,
+    BundleLimits,
+    BundleReader,
+    OpenedBundle,
+)
+from laborious.utils.bundle.steps import GATE_TO_STEP, ImportStep
+
+__all__ = [
+    'GATE_TO_STEP',
+    'BundleFormatError',
+    'BundleGateError',
+    'BundleHeader',
+    'BundleLimits',
+    'BundleReader',
+    'ImportStep',
+    'OpenedBundle',
+]

laborious/utils/bundle/format.py

@@ -0,0 +1,216 @@
+"""The `.sientia` wire format, as the producer writes it.
+
+Every constant here is copied from the producer that emits the bundles this reader opens — the
+Streamlit app in `sientia-projects-templates`, modules
+`app/src/operations/model_export/crypto.py` and `app/src/operations/model_export/bundle.py`. The
+producer file each value came from is named next to it. The golden fixtures under
+`tests/fixtures/bundle/` are the anti-drift device: if the producer's layout moves, the tests here
+fail against the committed bytes instead of failing in production.
+
+This module deliberately imports nothing from Laborious, Temporal, MLflow or libsodium: it is pure
+`struct` and `hashlib` over bytes, so it stays liftable into a shared package if the format ever
+gets one (the cancelled QTZPOC-15). `reader.py` is where the crypto lives.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import struct
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Final
+
+# --- Producer: app/src/operations/model_export/crypto.py ---
+MAGIC: Final[bytes] = b'SIENTIA1'
+FORMAT_VERSION: Final[int] = 1
+AEAD_ID_XCHACHA20POLY1305_SECRETSTREAM: Final[int] = 1
+KDF_ID_ARGON2ID: Final[int] = 1
+SALT_SIZE: Final[int] = 16
+DIGEST_SIZE: Final[int] = 32
+RESERVED_SIGNATURE_SIZE: Final[int] = 64
+# `KDF_MEMLIMIT_BYTES` / `KDF_OPSLIMIT` on the producer: what it writes is also the ceiling the
+# reader refuses to exceed, so header bytes cannot buy an arbitrary allocation on the worker.
+MAX_KDF_MEMLIMIT: Final[int] = 256 * 1024 * 1024  # 268435456
+MAX_KDF_OPSLIMIT: Final[int] = 3
+# `KDF_PARALLELISM_RESERVED`: libsodium's `crypto_pwhash` has no lane parameter, so the field is
+# reserved and must be exactly 1.
+PARALLELISM: Final[int] = 1
+CHUNK_SIZE: Final[int] = 64 * 1024  # 65536, fixed by the format
+KEY_SIZE: Final[int] = 32  # crypto_secretstream_xchacha20poly1305_KEYBYTES
+
+# magic(8s) version(B) aead_id(B) kdf_id(B) salt(16s) memlimit(Q) opslimit(Q)
+# parallelism(B) digest(32s) reserved_signature(64s) — producer's `_HEADER_STRUCT`.
+HEADER_STRUCT: Final[struct.Struct] = struct.Struct('>8sBBB16sQQB32s64s')
+HEADER_SIZE: Final[int] = HEADER_STRUCT.size  # 140
+
+# --- libsodium constants the framing depends on ---
+# Not producer constants, but constants of the primitive the producer uses. Kept as literals so
+# this module stays import-free; `tests/laborious/utils/bundle/test_format.py` asserts they equal
+# the values `nacl.bindings` reports, which is what makes the literals safe.
+STREAM_HEADER_SIZE: Final[int] = 24  # crypto_secretstream_xchacha20poly1305_HEADERBYTES
+ABYTES: Final[int] = 17  # crypto_secretstream_xchacha20poly1305_ABYTES
+
+# The pre-upload check prefix of `08-security-and-encryption.md` § 5.3: enough bytes to hold the
+# header, the stream header and one full chunk. Derived from constants, never read from the file.
+PREFIX_SIZE: Final[int] = HEADER_SIZE + STREAM_HEADER_SIZE + CHUNK_SIZE + ABYTES  # 65717
+
+# The digest covers the chunk stream only. The producer computes `sha256(ciphertext)` over the
+# joined chunks and *then* writes `header + stream_header + ciphertext`, so on disk the covered
+# range starts after both headers. Verified against a bundle produced by the producer's own code:
+# `sha256(file[164:])` matches the header field and `sha256(file[140:])` does not.
+DIGEST_OFFSET: Final[int] = HEADER_SIZE + STREAM_HEADER_SIZE  # 164
+
+MIN_FILE_SIZE: Final[int] = HEADER_SIZE + STREAM_HEADER_SIZE
+
+SUPPORTED_FORMAT_VERSIONS: Final[frozenset[int]] = frozenset({FORMAT_VERSION})
+SUPPORTED_AEAD_IDS: Final[frozenset[int]] = frozenset({AEAD_ID_XCHACHA20POLY1305_SECRETSTREAM})
+SUPPORTED_KDF_IDS: Final[frozenset[int]] = frozenset({KDF_ID_ARGON2ID})
+
+_HASH_BLOCK_SIZE: Final[int] = 1024 * 1024
+
+
+class BundleFormatError(Exception):
+    """The bytes are not a `.sientia` bundle this reader implements.
+
+    Raised for every header-level rejection: wrong magic, unknown version, unknown primitive ids,
+    cost parameters above the ceiling, a reserved field carrying a value, a file too short to hold
+    a header, or a chunk stream whose digest does not match the header's. `reader.py` turns this
+    into a gate-2 rejection; nothing here knows about gates or import steps.
+    """
+
+
+@dataclass(frozen=True)
+class BundleHeader:
+    """The 140-byte plaintext header, parsed and validated."""
+
+    format_version: int
+    aead_id: int
+    kdf_id: int
+    salt: bytes
+    kdf_memlimit_bytes: int
+    kdf_opslimit: int
+    kdf_parallelism: int
+    ciphertext_digest: bytes
+    reserved_signature: bytes
+
+
+def parse_header(raw: bytes) -> BundleHeader:
+    """Parse and validate the plaintext header.
+
+    Everything the reader trusts about the file's shape is decided here, before any offset past
+    the header is used and before a key is derived.
+
+    Args:
+        raw: at least `HEADER_SIZE` bytes read from the start of the object.
+
+    Returns:
+        BundleHeader: the validated header.
+
+    Raises:
+        BundleFormatError: on any invalid or unsupported field.
+    """
+    if len(raw) < HEADER_SIZE:
+        raise BundleFormatError(
+            f'file is shorter than the {HEADER_SIZE}-byte header ({len(raw)} bytes available)'
+        )
+
+    (
+        magic,
+        format_version,
+        aead_id,
+        kdf_id,
+        salt,
+        kdf_memlimit_bytes,
+        kdf_opslimit,
+        kdf_parallelism,
+        ciphertext_digest,
+        reserved_signature,
+    ) = HEADER_STRUCT.unpack(raw[:HEADER_SIZE])
+
+    if magic != MAGIC:
+        raise BundleFormatError('file does not start with the expected bundle marker')
+    if format_version not in SUPPORTED_FORMAT_VERSIONS:
+        raise BundleFormatError(f'unsupported bundle format version: {format_version}')
+    if aead_id not in SUPPORTED_AEAD_IDS:
+        raise BundleFormatError(f'unsupported encryption identifier: {aead_id}')
+    if kdf_id not in SUPPORTED_KDF_IDS:
+        raise BundleFormatError(f'unsupported key derivation identifier: {kdf_id}')
+    if kdf_memlimit_bytes > MAX_KDF_MEMLIMIT:
+        raise BundleFormatError(
+            f'declared key derivation memory {kdf_memlimit_bytes} exceeds the '
+            f'{MAX_KDF_MEMLIMIT} ceiling'
+        )
+    if kdf_opslimit > MAX_KDF_OPSLIMIT:
+        raise BundleFormatError(
+            f'declared key derivation passes {kdf_opslimit} exceeds the {MAX_KDF_OPSLIMIT} ceiling'
+        )
+    if kdf_memlimit_bytes <= 0 or kdf_opslimit <= 0:
+        raise BundleFormatError('key derivation parameters must be positive')
+    if kdf_parallelism != PARALLELISM:
+        raise BundleFormatError(
+            f'reserved parallelism field must be {PARALLELISM}, found {kdf_parallelism}'
+        )
+    if reserved_signature != bytes(RESERVED_SIGNATURE_SIZE):
+        raise BundleFormatError('reserved signature field must be empty in this format version')
+
+    return BundleHeader(
+        format_version=format_version,
+        aead_id=aead_id,
+        kdf_id=kdf_id,
+        salt=salt,
+        kdf_memlimit_bytes=kdf_memlimit_bytes,
+        kdf_opslimit=kdf_opslimit,
+        kdf_parallelism=kdf_parallelism,
+        ciphertext_digest=ciphertext_digest,
+        reserved_signature=reserved_signature,
+    )
+
+
+def read_header(path: Path | str) -> BundleHeader:
+    """Read and validate the header of the file at `path`.
+
+    Raises:
+        BundleFormatError: when the file cannot hold a header plus a stream header, or when the
+            header itself is invalid.
+    """
+    file_path = Path(path)
+    size = file_path.stat().st_size
+    if size < MIN_FILE_SIZE:
+        raise BundleFormatError(
+            f'file is {size} bytes, smaller than the {MIN_FILE_SIZE} bytes a bundle needs for its '
+            'header and encrypted stream header'
+        )
+    with file_path.open('rb') as handle:
+        return parse_header(handle.read(HEADER_SIZE))
+
+
+def compute_ciphertext_digest(path: Path | str) -> bytes:
+    """SHA-256 of the chunk stream — the bytes from `DIGEST_OFFSET` to the end of the file."""
+    digest = hashlib.sha256()
+    with Path(path).open('rb') as handle:
+        handle.seek(DIGEST_OFFSET)
+        while block := handle.read(_HASH_BLOCK_SIZE):
+            digest.update(block)
+    return digest.digest()
+
+
+def verify_ciphertext_digest(path: Path | str, header: BundleHeader) -> None:
+    """Compare the file's chunk-stream digest against the header's.
+
+    Raises:
+        BundleFormatError: when they differ — a truncated, overwritten or corrupted transfer. This
+            runs before any key derivation, so a damaged file never costs an Argon2id pass.
+    """
+    if compute_ciphertext_digest(path) != header.ciphertext_digest:
+        raise BundleFormatError(
+            'the encrypted content does not match the digest recorded in the bundle header'
+        )
+
+
+def file_digest(path: Path | str) -> str:
+    """Hex SHA-256 of the whole file, for the `expected_digest` comparison of gate 1."""
+    digest = hashlib.sha256()
+    with Path(path).open('rb') as handle:
+        while block := handle.read(_HASH_BLOCK_SIZE):
+            digest.update(block)
+    return digest.hexdigest()

laborious/utils/bundle/reader.py

@@ -0,0 +1,473 @@
+"""Reading a `.sientia` bundle: the seven gates, in order, in one function.
+
+`BundleReader.open()` is the only entry point. Gate ordering is a property of that function rather
+than of the caller, so it cannot be reordered by accident and can be unit-tested without Temporal:
+
+1. object-level limits (`check_object_limits`, called by the download activity before the body is
+   fetched — the only gate that does not need the file)
+2. header validation, including the chunk-stream digest, before any key derivation
+3. AEAD decryption — the integrity check; a wrong password and a tampered file fail identically
+4. archive inspection over the central directory, writing nothing to disk
+5. extraction into a fresh, isolated directory
+6. structure and schema of the extracted tree
+7. content policy over the file set
+
+Nothing here logs, notifies or emits a metric: rejections are raised as `BundleGateError` and the
+activity layer owns the one failure channel.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import shutil
+import stat
+import time
+import unicodedata
+import zipfile
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Any
+
+import nacl.bindings as sodium
+import nacl.exceptions
+
+from laborious.utils.bundle import format as fmt
+from laborious.utils.bundle.steps import GATE_TO_STEP, ImportStep
+
+# Gate 6 — the layout `bundle.build_zip` writes and `model_repository.py` reads back.
+METADATA_FILE_NAME = 'metadata.json'
+ARTIFACTS_DIR_NAME = 'artifacts'
+REQUIRED_TOP_LEVEL_KEYS = frozenset({'parameters', 'metadata'})
+REQUIRED_METADATA_FIELDS = (
+    'model_name',
+    'experiment_name',
+    'run_id',
+    'model_version',
+    'model_project',
+    'export_timestamp',
+)
+REQUIRED_PARAMETERS = ('target_variable',)
+REQUIRED_MODEL_DIRS = ('prediction_model', 'data_model')
+REQUIRED_MODEL_FILES = ('MLmodel', 'model.pkl')
+
+# The import record's `experiment_name` column is `VARCHAR(50)` with a length-3 floor, and the name
+# is written onto the record before provisioning starts — so the column's limits are enforced here,
+# not three activities later as a constraint violation.
+EXPERIMENT_NAME_MIN_LENGTH = 3
+EXPERIMENT_NAME_MAX_LENGTH = 50
+
+# Gate 7 — an MLflow 2.x artifact tree for this platform: the model directories' known files, the
+# model card, the CSVs the training template logs and the bundle's own metadata.
+ALLOWED_FILE_NAMES = frozenset(
+    {
+        'MLmodel',
+        'metadata.json',
+        'model.pkl',
+        'conda.yaml',
+        'python_env.yaml',
+        'requirements.txt',
+        'model_card.json',
+        'model_card.svg',
+    }
+)
+ALLOWED_FILE_SUFFIXES = frozenset({'.csv', '.json', '.yaml', '.yml', '.txt', '.pkl', '.svg', '.md'})
+
+_DECRYPT_FAILURE_MESSAGE = (
+    'could not open this bundle: wrong password, or the file is damaged or was altered'
+)
+
+
+class BundleGateError(Exception):
+    """A gate rejected the bundle.
+
+    Carries the gate number and the import step that gate maps onto, so the workflow never has to
+    guess which step to record.
+    """
+
+    def __init__(self, message: str, *, gate: int, step: ImportStep | None = None) -> None:
+        super().__init__(message)
+        self.gate = gate
+        self.step = step if step is not None else GATE_TO_STEP[gate]
+
+
+@dataclass(frozen=True)
+class BundleLimits:
+    """Ceilings enforced by gates 1 and 4. Values come from `build_import_config()`."""
+
+    max_object_bytes: int = 1024 * 1024 * 1024
+    max_entries: int = 5000
+    max_uncompressed_bytes: int = 4 * 1024 * 1024 * 1024
+    max_compression_ratio: float = 200.0
+    object_prefix: str = 'imported_models/'
+    object_suffix: str = '.sientia'
+
+
+@dataclass(frozen=True)
+class OpenedBundle:
+    """What a bundle yields once every gate has passed. Nothing secret is carried."""
+
+    extracted_dir: Path
+    metadata: dict[str, Any] = field(default_factory=dict)
+    parameters: dict[str, Any] = field(default_factory=dict)
+    digest: str = ''
+    header: fmt.BundleHeader | None = None
+    # How long Argon2id took, so the worker's most expensive step is measurable without the
+    # activity layer having to reach inside the reader.
+    kdf_seconds: float = 0.0
+
+
+class BundleReader:
+    """Opens `.sientia` bundles produced by the exporter in `sientia-projects-templates`."""
+
+    def __init__(self, limits: BundleLimits | None = None) -> None:
+        self.limits = limits or BundleLimits()
+
+    # ------------------------------------------------------------------ gate 1
+
+    def check_object_limits(self, *, size: int, key: str) -> None:
+        """Gate 1, object level: reject before the object's body is fetched.
+
+        Args:
+            size: the object's size as reported by `stat_object`.
+            key: the object key.
+
+        Raises:
+            BundleGateError: oversized object, key outside the import prefix, or wrong suffix.
+        """
+        if size > self.limits.max_object_bytes:
+            raise BundleGateError(
+                f'uploaded object is {size} bytes, above the '
+                f'{self.limits.max_object_bytes} byte limit',
+                gate=1,
+            )
+        if size <= 0:
+            raise BundleGateError('uploaded object is empty', gate=1)
+        if not key.startswith(self.limits.object_prefix):
+            raise BundleGateError('uploaded object is not under the import prefix', gate=1)
+        if not key.endswith(self.limits.object_suffix):
+            raise BundleGateError(
+                f'uploaded object does not end in {self.limits.object_suffix}', gate=1
+            )
+
+    # --------------------------------------------------------------------- open
+
+    def open(
+        self,
+        encrypted_path: Path | str,
+        password: str,
+        extract_root: Path | str,
+        work_dir: Path | str | None = None,
+    ) -> OpenedBundle:
+        """Run gates 2 to 7 in order and return the opened bundle.
+
+        Args:
+            encrypted_path: the downloaded `.sientia` file.
+            password: the plaintext bundle password. Never stored, never logged, never returned.
+            extract_root: the directory to extract into. It MUST NOT exist yet (gate 5).
+            work_dir: where the decrypted zip is written. Defaults to `extract_root`'s parent.
+
+        Returns:
+            OpenedBundle: extracted tree, `metadata`, `parameters`, digest and header.
+
+        Raises:
+            BundleGateError: with the gate that rejected the bundle and its import step.
+        """
+        encrypted = Path(encrypted_path)
+        root = Path(extract_root)
+        staging = Path(work_dir) if work_dir is not None else root.parent
+
+        header = self._gate_2_header(encrypted)
+        zip_path, kdf_seconds = self._gate_3_decrypt(encrypted, password, header, staging)
+        try:
+            self._gate_4_inspect(zip_path)
+            self._gate_5_extract(zip_path, root)
+            metadata, parameters = self._gate_6_structure(root)
+            self._gate_7_content_policy(root)
+        except BaseException:
+            # A rejection after decryption leaves nothing extracted behind (gate 5's contract) and
+            # never leaves a decrypted zip on the worker for the next import to trip over.
+            shutil.rmtree(root, ignore_errors=True)
+            zip_path.unlink(missing_ok=True)
+            raise
+        zip_path.unlink(missing_ok=True)
+
+        return OpenedBundle(
+            extracted_dir=root,
+            metadata=metadata,
+            parameters=parameters,
+            digest=header.ciphertext_digest.hex(),
+            header=header,
+            kdf_seconds=kdf_seconds,
+        )
+
+    # ------------------------------------------------------------------ gate 2
+
+    def _gate_2_header(self, encrypted: Path) -> fmt.BundleHeader:
+        """Parse the header and check the chunk-stream digest, before deriving anything."""
+        try:
+            header = fmt.read_header(encrypted)
+            fmt.verify_ciphertext_digest(encrypted, header)
+        except fmt.BundleFormatError as error:
+            raise BundleGateError(str(error), gate=2) from error
+        except OSError as error:
+            raise BundleGateError('downloaded bundle could not be read', gate=2) from error
+        return header
+
+    # ------------------------------------------------------------------ gate 3
+
+    def _gate_3_decrypt(
+        self, encrypted: Path, password: str, header: fmt.BundleHeader, staging: Path
+    ) -> tuple[Path, float]:
+        """Derive the key with the header's parameters and decrypt the stream to a file.
+
+        The AEAD *is* the integrity check: a wrong password and a tampered byte fail here with the
+        same message, which never claims to distinguish them.
+        """
+        staging.mkdir(parents=True, exist_ok=True)
+        zip_path = staging / 'bundle.zip'
+        try:
+            started = time.monotonic()
+            key = self._derive_key(password, header)
+            kdf_seconds = time.monotonic() - started
+            self._pull_stream(encrypted, key, zip_path)
+        except BundleGateError:
+            zip_path.unlink(missing_ok=True)
+            raise
+        except (nacl.exceptions.CryptoError, ValueError, RuntimeError) as error:
+            zip_path.unlink(missing_ok=True)
+            raise BundleGateError(_DECRYPT_FAILURE_MESSAGE, gate=3) from error
+        except OSError as error:
+            zip_path.unlink(missing_ok=True)
+            raise BundleGateError('decrypted bundle could not be written', gate=3) from error
+        return zip_path, kdf_seconds
+
+    @staticmethod
+    def _derive_key(password: str, header: fmt.BundleHeader) -> bytes:
+        """Argon2id over the NFC-normalised password, with the header's own cost parameters.
+
+        The password is normalised and **not** trimmed, matching the producer's `_derive_key`: the
+        same password typed on another OS must derive the same key, while a trailing space is part
+        of the secret.
+        """
+        normalized = unicodedata.normalize('NFC', password)
+        return sodium.crypto_pwhash_alg(
+            fmt.KEY_SIZE,
+            normalized.encode('utf-8'),
+            header.salt,
+            header.kdf_opslimit,
+            header.kdf_memlimit_bytes,
+            sodium.crypto_pwhash_ALG_ARGON2ID13,
+        )
+
+    @staticmethod
+    def _pull_stream(encrypted: Path, key: bytes, zip_path: Path) -> None:
+        """Decrypt the chunk stream into `zip_path`, requiring the FINAL tag."""
+        chunk_size = fmt.CHUNK_SIZE + fmt.ABYTES
+        state = sodium.crypto_secretstream_xchacha20poly1305_state()
+        with encrypted.open('rb') as source, zip_path.open('wb') as target:
+            source.seek(fmt.HEADER_SIZE)
+            stream_header = source.read(fmt.STREAM_HEADER_SIZE)
+            sodium.crypto_secretstream_xchacha20poly1305_init_pull(state, stream_header, key)
+            last_tag: int | None = None
+            while chunk := source.read(chunk_size):
+                plaintext, last_tag = sodium.crypto_secretstream_xchacha20poly1305_pull(
+                    state, chunk
+                )
+                target.write(plaintext)
+        if last_tag != sodium.crypto_secretstream_xchacha20poly1305_TAG_FINAL:
+            # A stream that stops without its FINAL tag was cut short. The digest already catches
+            # truncation of a stored file; this catches a stream that was framed to look complete.
+            raise BundleGateError(_DECRYPT_FAILURE_MESSAGE, gate=3)
+
+    # ------------------------------------------------------------------ gate 4
+
+    def _gate_4_inspect(self, zip_path: Path) -> None:
+        """Inspect the central directory only. Nothing is written to disk by this gate."""
+        try:
+            with zipfile.ZipFile(zip_path) as archive:
+                entries = archive.infolist()
+        except zipfile.BadZipFile as error:
+            raise BundleGateError('bundle content is not a readable archive', gate=4) from error
+
+        if len(entries) > self.limits.max_entries:
+            raise BundleGateError(
+                f'archive declares {len(entries)} entries, above the '
+                f'{self.limits.max_entries} limit',
+                gate=4,
+            )
+        if not entries:
+            raise BundleGateError('archive is empty', gate=4)
+
+        uncompressed = sum(entry.file_size for entry in entries)
+        compressed = sum(entry.compress_size for entry in entries)
+        if uncompressed > self.limits.max_uncompressed_bytes:
+            raise BundleGateError(
+                f'archive declares {uncompressed} uncompressed bytes, above the '
+                f'{self.limits.max_uncompressed_bytes} limit',
+                gate=4,
+            )
+        ratio = uncompressed / max(compressed, 1)
+        if ratio > self.limits.max_compression_ratio:
+            raise BundleGateError(
+                f'archive compression ratio {ratio:.1f} is above the '
+                f'{self.limits.max_compression_ratio} limit',
+                gate=4,
+            )
+
+        for entry in entries:
+            self._check_entry_name(entry.filename)
+            self._check_entry_mode(entry)
+
+    @staticmethod
+    def _check_entry_name(name: str) -> None:
+        """Reject absolute names and traversal, from the declared name alone."""
+        normalised = name.replace('\\', '/')
+        if normalised.startswith('/') or (len(normalised) > 1 and normalised[1] == ':'):
+            raise BundleGateError('archive declares an absolute entry path', gate=4)
+        if any(part == '..' for part in normalised.split('/')):
+            raise BundleGateError('archive declares an entry escaping its own tree', gate=4)
+
+    @staticmethod
+    def _check_entry_mode(entry: zipfile.ZipInfo) -> None:
+        """Reject symlinks and anything that is neither a regular file nor a directory."""
+        mode = entry.external_attr >> 16
+        if stat.S_IFMT(mode) == 0:
+            # No file-type bits stored. This is the normal case for two kinds of entry: a zip
+            # written by a tool that records no Unix mode at all (`mode == 0`), and
+            # `ZipFile.writestr`, which stores permissions only — the producer's `metadata.json`
+            # arrives as `0o600 << 16`. Nothing to check: the member is extracted as a plain file
+            # with the worker's own permissions either way.
+            return
+        if stat.S_ISLNK(mode):
+            raise BundleGateError('archive declares a symbolic link', gate=4)
+        if not (stat.S_ISREG(mode) or stat.S_ISDIR(mode)):
+            raise BundleGateError('archive declares an entry that is not a regular file', gate=4)
+
+    # ------------------------------------------------------------------ gate 5
+
+    def _gate_5_extract(self, zip_path: Path, root: Path) -> None:
+        """Extract into a fresh directory, containing every member by resolved path.
+
+        A pre-existing extraction root is a failure rather than something to clear: it means
+        another import, or a previous attempt, owns that path.
+        """
+        if root.exists():
+            raise BundleGateError('extraction directory already exists', gate=5)
+        root.mkdir(parents=True)
+        resolved_root = root.resolve()
+        try:
+            with zipfile.ZipFile(zip_path) as archive:
+                for entry in archive.infolist():
+                    self._extract_member(archive, entry, root, resolved_root)
+        except BundleGateError:
+            shutil.rmtree(root, ignore_errors=True)
+            raise
+        except (OSError, zipfile.BadZipFile) as error:
+            shutil.rmtree(root, ignore_errors=True)
+            raise BundleGateError('bundle content could not be unpacked', gate=5) from error
+
+    @staticmethod
+    def _extract_member(
+        archive: zipfile.ZipFile, entry: zipfile.ZipInfo, root: Path, resolved_root: Path
+    ) -> None:
+        """Write one member, re-checking containment on the resolved path."""
+        name = entry.filename.replace('\\', '/')
+        target = root / name
+        resolved = Path(os.path.realpath(target))
+        if resolved != resolved_root and resolved_root not in resolved.parents:
+            raise BundleGateError('archive member resolves outside the extraction root', gate=5)
+        if entry.is_dir():
+            resolved.mkdir(parents=True, exist_ok=True)
+            return
+        if resolved.exists():
+            raise BundleGateError('archive member would overwrite an existing file', gate=5)
+        resolved.parent.mkdir(parents=True, exist_ok=True)
+        # Written as a plain file: the archive's stored permissions are never restored, so no
+        # member can arrive executable.
+        with archive.open(entry) as source, resolved.open('wb') as sink:
+            shutil.copyfileobj(source, sink)
+
+    # ------------------------------------------------------------------ gate 6
+
+    def _gate_6_structure(self, root: Path) -> tuple[dict[str, Any], dict[str, Any]]:
+        """Validate the extracted layout and the metadata schema."""
+        metadata_path = root / METADATA_FILE_NAME
+        artifacts_dir = root / ARTIFACTS_DIR_NAME
+        if not metadata_path.is_file():
+            raise BundleGateError('bundle has no metadata document at its root', gate=6)
+        if not artifacts_dir.is_dir():
+            raise BundleGateError('bundle has no artifacts directory', gate=6)
+
+        document = self._load_metadata_document(metadata_path)
+        metadata = document['metadata']
+        parameters = document['parameters']
+
+        self._check_metadata_fields(metadata)
+        self._check_parameters(parameters)
+        self._check_model_directories(artifacts_dir)
+        return metadata, parameters
+
+    @staticmethod
+    def _load_metadata_document(metadata_path: Path) -> dict[str, Any]:
+        """Read `metadata.json` and check its two top-level keys."""
+        try:
+            document = json.loads(metadata_path.read_text(encoding='utf-8'))
+        except (json.JSONDecodeError, UnicodeDecodeError, OSError) as error:
+            raise BundleGateError('bundle metadata is not readable JSON', gate=6) from error
+        if not isinstance(document, dict) or set(document) != REQUIRED_TOP_LEVEL_KEYS:
+            raise BundleGateError(
+                'bundle metadata does not carry exactly the parameters and metadata blocks', gate=6
+            )
+        if not isinstance(document['metadata'], dict) or not isinstance(
+            document['parameters'], dict
+        ):
+            raise BundleGateError('bundle metadata blocks are not objects', gate=6)
+        return document
+
+    @staticmethod
+    def _check_metadata_fields(metadata: dict[str, Any]) -> None:
+        """Every origin field the import records must be present and usable."""
+        for field_name in REQUIRED_METADATA_FIELDS:
+            value = metadata.get(field_name)
+            if not isinstance(value, str) or not value.strip():
+                raise BundleGateError(f'bundle metadata is missing the {field_name} field', gate=6)
+        experiment_name = metadata['experiment_name']
+        if not (EXPERIMENT_NAME_MIN_LENGTH <= len(experiment_name) <= EXPERIMENT_NAME_MAX_LENGTH):
+            raise BundleGateError(
+                "bundle experiment name does not fit the platform's limits", gate=6
+            )
+
+    @staticmethod
+    def _check_parameters(parameters: dict[str, Any]) -> None:
+        """`target_variable` is what `model_config.target` is built from; nothing else supplies it."""
+        for name in REQUIRED_PARAMETERS:
+            value = parameters.get(name)
+            if value is None or not str(value).strip():
+                raise BundleGateError(f'bundle parameters are missing {name}', gate=6)
+
+    @staticmethod
+    def _check_model_directories(artifacts_dir: Path) -> None:
+        """`prediction_model` and `data_model` must each be a loadable MLflow model directory."""
+        for directory in REQUIRED_MODEL_DIRS:
+            model_dir = artifacts_dir / directory
+            if not model_dir.is_dir():
+                raise BundleGateError(f'bundle has no {directory} artifacts', gate=6)
+            for required in REQUIRED_MODEL_FILES:
+                if not (model_dir / required).is_file():
+                    raise BundleGateError(f'bundle {directory} artifacts are incomplete', gate=6)
+
+    # ------------------------------------------------------------------ gate 7
+
+    def _gate_7_content_policy(self, root: Path) -> None:
+        """Only the file kinds an MLflow artifact tree calls for may be present."""
+        for path in sorted(root.rglob('*')):
+            if path.is_dir():
+                continue
+            if not path.is_file():
+                raise BundleGateError('bundle contains an entry that is not a file', gate=7)
+            if path.name in ALLOWED_FILE_NAMES:
+                continue
+            if path.suffix.lower() in ALLOWED_FILE_SUFFIXES:
+                continue
+            raise BundleGateError('bundle contains a file a model export should not carry', gate=7)

laborious/utils/bundle/steps.py

@@ -0,0 +1,55 @@
+"""The import pipeline's step vocabulary, and the gate-to-step mapping.
+
+`ImportStep` is the join between three things: the machine-readable step written into the import
+record's JSON detail, the key of the error catalog (one code and one sentence per step) and the
+value the frontend switches on. It lives in the bundle package rather than in the activities module
+because the reader raises with it — `laborious/activities/model_import.py` imports it from here so
+there is exactly one definition.
+
+It holds **pipeline** steps only. The two failures that are not steps of the pipeline — failing to
+claim the record and failing to write it — are reported by the failure channel under their own
+markers (`not_claimable`, `status_write`), deliberately outside this enum so the
+catalog-completeness test keeps meaning "every pipeline step has a sentence".
+"""
+
+from __future__ import annotations
+
+from enum import StrEnum
+from typing import Final
+
+
+class ImportStep(StrEnum):
+    """Where an import stopped. Values are what gets written to the record."""
+
+    RECEIVED = 'received'
+    DOWNLOAD = 'download'
+    DECRYPTION = 'decryption'
+    ARCHIVE_INSPECTION = 'archive_inspection'
+    EXTRACTION = 'extraction'
+    STRUCTURE_VALIDATION = 'structure_validation'
+    CONTENT_POLICY = 'content_policy'
+    EXPERIMENT_CREATION = 'experiment_creation'
+    ARTIFACT_UPLOAD = 'artifact_upload'
+    REGISTRATION = 'registration'
+    MODEL_DOCUMENT = 'model_document'
+    CLEANUP = 'cleanup'
+
+
+# The seven gates map onto the first seven steps, with gates 1 and 2 sharing one: both say "these
+# are not the bytes that were sent" — an object over the size ceiling, a key with the wrong shape, a
+# digest that disagrees with what the uploader computed, a wrong magic, an unsupported version, a
+# chunk stream that does not match the header's digest. The catalog's `download` sentence ("The
+# uploaded file could not be read, or it is not the file that was sent. Upload it again.") is true
+# for all of them, and none of them may borrow the `decryption` sentence, which implicates the
+# password.
+GATE_TO_STEP: Final[dict[int, ImportStep]] = {
+    1: ImportStep.DOWNLOAD,
+    2: ImportStep.DOWNLOAD,
+    3: ImportStep.DECRYPTION,
+    4: ImportStep.ARCHIVE_INSPECTION,
+    5: ImportStep.EXTRACTION,
+    6: ImportStep.STRUCTURE_VALIDATION,
+    7: ImportStep.CONTENT_POLICY,
+}
+
+GATE_COUNT: Final[int] = 7

laborious/utils/connectors_config.py

@@ -0,0 +1,171 @@
+import json
+import tempfile
+from os import getenv, path
+from typing import Any
+
+
+def build_mlflow_config() -> dict[str, Any]:
+    """
+    Build MLFlow server configuration from environment variables.
+
+    This function constructs an MLFlow configuration dictionary from
+    environment variables with sensible defaults for local development.
+    It handles server connection and authentication parameters.
+
+    Environment Variables:
+        MLFLOW_HOST: MLFlow server hostname (default: http://localhost)
+        MLFLOW_PORT: MLFlow server port (default: 5080)
+        MLFLOW_USERNAME: MLFlow username (default: aignosi)
+        MLFLOW_PASSWORD: MLFlow password (default: aignosi)
+
+    Returns:
+        dict: MLFlow configuration dictionary with all required parameters
+    """
+    return {
+        'host': getenv('MLFLOW_HOST', 'http://localhost'),
+        'port': int(getenv('MLFLOW_PORT', '5080')),
+        'username': getenv('MLFLOW_USERNAME', 'aignosi'),
+        'password': getenv('MLFLOW_PASSWORD', 'aignosi'),
+    }
+
+
+def build_opc_config() -> dict[str, Any]:
+    """
+    Build OPC server configuration from environment variables.
+
+    This function constructs an OPC server configuration dictionary from
+    environment variables. It supports both single server and multi-server
+    configurations with flexible parameter handling.
+
+    Environment Variables:
+        OPC_CONFIG: JSON string containing multiple OPC server configurations
+        OPC_ID: OPC server ID (fallback, default: 1)
+        OPC_URL: Single OPC server URL (fallback, default: opc.tcp://localhost:4840)
+        OPC_SERVER_URI: Single OPC server URI (fallback, default: opc.tcp://localhost:4840)
+        OPC_CERT_PATH: Client certificate path (fallback, default: None)
+        OPC_PRIVATE_KEY_PATH: Client private key path (fallback, default: None)
+        OPC_SERVER_CERT_PATH: Server certificate path (fallback, default: None)
+        OPC_RECONNECTION_INTERVAL: Reconnection interval in seconds (fallback, default: 120)
+
+    Returns:
+        dict: OPC server configuration dictionary
+    """
+    opc_raw = getenv('OPC_CONFIG', None)
+
+    if opc_raw:
+        return json.loads(opc_raw)
+
+    return {
+        getenv('OPC_ID', '1'): {
+            'id': getenv('OPC_ID', '1'),
+            'server_name': getenv('OPC_SERVER_NAME', 'default_server'),
+            'url': getenv('OPC_URL', 'opc.tcp://localhost:4840'),
+            'server_uri': getenv('OPC_SERVER_URI', 'opc.tcp://localhost:4840'),
+            'cert_path': getenv('OPC_CERT_PATH', None),
+            'private_key_path': getenv('OPC_PRIVATE_KEY_PATH', None),
+            'server_cert_path': getenv('OPC_SERVER_CERT_PATH', None),
+            'reconnection_interval': int(getenv('OPC_RECONNECTION_INTERVAL', '120')),
+        }
+    }
+
+
+def build_minio_config() -> dict[str, Any]:
+    """
+    Build MinIO (S3-compatible) configuration from environment variables.
+
+    Environment Variables:
+        MINIO_ENDPOINT: MinIO endpoint including scheme (default: http://localhost:9000)
+        MINIO_ACCESS_KEY: Access key (default: minioadmin)
+        MINIO_SECRET_KEY: Secret key (default: minioadmin)
+        MINIO_REGION: Region name for S3 client (default: us-east-1)
+        MINIO_BUCKET_DEFAULT: Default bucket for uploads (default: laborious)
+        MINIO_SECURE: Whether to use HTTPS (default: false)
+    Returns:
+        dict: MinIO configuration dictionary
+    """
+    return {
+        'endpoint_url': getenv('MINIO_ENDPOINT_URL', 'http://localhost:9000'),
+        'access_key': getenv('MINIO_ACCESS_KEY', 'minioadmin'),
+        'secret_key': getenv('MINIO_SECRET_KEY', 'minioadmin'),
+        'default_bucket': getenv('MINIO_DEFAULT_BUCKET', 'laborious'),
+        'retention_hours': int(getenv('MINIO_RETENTION_HOURS', '24')),
+        'secure': getenv('MINIO_SECURE', 'false') == 'true',
+    }
+
+
+def build_import_config() -> dict[str, Any]:
+    """
+    Build the `.sientia` model-import configuration from environment variables.
+
+    The password key is read as the raw base64 string and **not** validated here: the worker must
+    start whether or not `IMPORT_MODEL_ENABLED` is set, and a missing or malformed key is an import
+    failure at the decryption step (`laborious.utils.import_password.load_envelope_key`), not a
+    boot failure.
+
+    Environment Variables:
+        IMPORT_PASSWORD_KEY: base64 of the 32-byte AES-256-GCM key that wraps bundle passwords
+        IMPORT_BUNDLE_BUCKET: bucket holding the uploaded bundles (default: sientia)
+        IMPORT_BUNDLE_PREFIX: key prefix the uploads must sit under (default: imported_models/)
+        IMPORT_WORK_DIR: worker scratch directory (default: <tempdir>/sientia-import)
+        IMPORT_MAX_OBJECT_BYTES: gate 1 object size ceiling (default: 1 GiB)
+        IMPORT_MAX_ARCHIVE_ENTRIES: gate 4 entry count ceiling (default: 5000)
+        IMPORT_MAX_UNCOMPRESSED_BYTES: gate 4 uncompressed size ceiling (default: 4 GiB)
+        IMPORT_MAX_COMPRESSION_RATIO: gate 4 compression ratio ceiling (default: 200)
+        IMPORT_MODELS_COLLECTION: MongoDB collection holding the model listing (default: models)
+        IMPORT_BUNDLE_RETENTION_DAYS: lifecycle expiry for the uploaded bundle (default: 7)
+
+    Returns:
+        dict: import configuration dictionary
+    """
+    prefix = getenv('IMPORT_BUNDLE_PREFIX', 'imported_models/')
+    return {
+        'password_key': getenv('IMPORT_PASSWORD_KEY'),
+        'bucket': getenv('IMPORT_BUNDLE_BUCKET', 'sientia'),
+        'prefix': prefix if prefix.endswith('/') else f'{prefix}/',
+        'work_dir': getenv('IMPORT_WORK_DIR', path.join(tempfile.gettempdir(), 'sientia-import')),
+        'max_object_bytes': int(getenv('IMPORT_MAX_OBJECT_BYTES', str(1024 * 1024 * 1024))),
+        'max_entries': int(getenv('IMPORT_MAX_ARCHIVE_ENTRIES', '5000')),
+        'max_uncompressed_bytes': int(
+            getenv('IMPORT_MAX_UNCOMPRESSED_BYTES', str(4 * 1024 * 1024 * 1024))
+        ),
+        'max_compression_ratio': float(getenv('IMPORT_MAX_COMPRESSION_RATIO', '200')),
+        'models_collection': getenv('IMPORT_MODELS_COLLECTION', 'models'),
+        'retention_days': int(getenv('IMPORT_BUNDLE_RETENTION_DAYS', '7')),
+    }
+
+
+def build_import_status_config() -> dict[str, Any]:
+    """
+    Build the connection to the import log database — the POSTGRES_* server, another database.
+
+    The import log is `public.experiment_run` in the database owned by the Spring Boot BFF
+    (`sientia-core-mlops-bff`), which sits on the same Postgres server as the `sientia` database
+    Laborious itself uses. Only the database name differs, so only the database name has a variable
+    of its own; host, port, user and password come from POSTGRES_*. The table and schema are
+    constants of the code, never inputs.
+
+    The four IMPORT_STATUS_DB_* connection variables remain as overrides, for a deployment where the
+    log genuinely lives on another server — the e2e suite is exactly that, two containers.
+
+    Environment Variables:
+        IMPORT_STATUS_DB_NAME: database (default: sientia-core-mlops-bff)
+        POSTGRES_HOST: server host, unless IMPORT_STATUS_DB_HOST overrides it (default: localhost)
+        POSTGRES_PORT: server port, unless IMPORT_STATUS_DB_PORT overrides it (default: 5432)
+        POSTGRES_USER: user, unless IMPORT_STATUS_DB_USER overrides it (default: sientia)
+        POSTGRES_PASSWORD: password, unless IMPORT_STATUS_DB_PASSWORD overrides it (default: empty)
+
+    Returns:
+        dict: import status database configuration dictionary
+    """
+    # Written as `override or fallback or default` rather than
+    # `getenv(override) or getenv(fallback, default)`: inside an `or`, the two-argument `getenv`
+    # widens to `str | None` for mypy, so the `int(...)` below fails to type-check. Putting the
+    # default last as its own term keeps every value a plain `str` and reads in precedence order.
+    return {
+        'host': getenv('IMPORT_STATUS_DB_HOST') or getenv('POSTGRES_HOST') or 'localhost',
+        'port': int(getenv('IMPORT_STATUS_DB_PORT') or getenv('POSTGRES_PORT') or '5432'),
+        'dbname': getenv('IMPORT_STATUS_DB_NAME') or 'sientia-core-mlops-bff',
+        'user': getenv('IMPORT_STATUS_DB_USER') or getenv('POSTGRES_USER') or 'sientia',
+        'password': getenv('IMPORT_STATUS_DB_PASSWORD') or getenv('POSTGRES_PASSWORD') or '',
+        'schema': 'public',
+    }

laborious/utils/dataframe_debug.py

@@ -0,0 +1,34 @@
+from typing import Any
+
+from pandas import DataFrame
+
+DEFAULT_MAX_DEBUG_DATAFRAME_ROWS = 100
+
+
+def build_dataframe_debug_message(
+    message: str,
+    data: Any,
+    max_rows: int = DEFAULT_MAX_DEBUG_DATAFRAME_ROWS,
+) -> str:
+    """
+    Build a safe debug message for dataframe payloads
+
+    Args:
+     - message (str): Base message to identify the logged payload
+     - data (Any): Payload to evaluate for dataframe-aware logging
+     - max_rows (int): Maximum dataframe row count allowed for full payload logging
+
+    Return:
+     Formatted debug message with full dataframe content or compact summary
+    """
+    if not isinstance(data, DataFrame):
+        return f'{message} {data}'
+
+    rows = data.shape[0]
+    if rows <= max_rows:
+        return f'{message}\n{data.to_csv()}'
+
+    return (
+        f'{message} skipped because dataframe has {rows} rows '
+        f'(max: {max_rows}). Shape: {data.shape}'
+    )

laborious/utils/filters/conditional_filters.py

@@ -0,0 +1,48 @@
+from pandas import DataFrame
+
+
+def filter_specific_variables_null_values(data: DataFrame, config: dict) -> bool:
+    """
+    Filter to check if specific variables contain null values.
+
+    This function examines a DataFrame to determine if any of the specified variables
+    contain null (NaN) values. It returns True if null values are found for any of
+    the specified variables, False otherwise.
+
+    Args:
+        data (DataFrame): The pandas DataFrame to be examined. Must contain columns
+            named 'variable' and 'value'.
+        config (dict): Configuration dictionary containing the following key:
+            - variables (list): List of variable names to check for null values
+
+    Returns:
+        bool: True if any of the specified variables contain null values,
+              False if none of the specified variables contain null values.
+
+    """
+
+    if data.empty:
+        return False
+
+    return not data[data['variable'].isin(config['variables']) & data['value'].isna()].empty
+
+
+def filter_empty_data(data: DataFrame, _config: dict) -> bool:
+    """
+    Filter to check if the DataFrame is empty.
+
+    This function determines whether the provided DataFrame contains any data.
+    It's a simple utility function that can be used in conditional logic to
+    handle cases where no data is available.
+
+    Args:
+        data (DataFrame): The pandas DataFrame to be checked for emptiness.
+        _config (dict): Configuration dictionary (unused in this function).
+            The underscore prefix indicates this parameter is required for
+            interface consistency but not used in the implementation.
+
+    Returns:
+        bool: True if the DataFrame is empty (has no rows), False if it contains data.
+
+    """
+    return data.empty

laborious/utils/filters/mlflow_filters.py

@@ -0,0 +1,64 @@
+import numpy as np
+from pandas import DataFrame
+
+
+def api_error_filter(response: dict, _config: dict) -> bool:
+    """
+    Filter MLFlow API responses for error conditions.
+
+    This function analyzes MLFlow API responses to detect error conditions
+    and determine if the response should be filtered out due to quality
+    or reliability issues.
+
+
+    Args:
+        response: MLFlow API response data (dict)
+        _config: Filter configuration dictionary
+            Required keys:
+                - error_codes (list, optional): List of error codes to detect
+                - error_keywords (list, optional): List of error keywords to detect
+                - check_structure (bool, optional): Whether to validate response structure
+
+    Returns:
+        bool: True if data should be filtered (contains errors), False otherwise
+
+    """
+    if not response:
+        return True
+
+    if not response['success']:
+        return True
+
+    return False
+
+
+def nan_values_filter(predictions: DataFrame, _config: dict) -> bool:
+    """
+    Filter data for NaN (Not a Number) values.
+
+    This function detects NaN values in MLFlow prediction results and
+    determines if the data quality is sufficient for further processing
+    or export operations.
+
+    Args:
+        predictions: DataFrame containing prediction data to check for NaN values
+        _config: Filter configuration dictionary
+            Required keys:
+                - max_nan_ratio (float, optional): Maximum allowed NaN value ratio (0.0 to 1.0)
+                - max_nan_count (int, optional): Maximum allowed NaN value count
+                - check_nested (bool, optional): Whether to check nested data structures
+
+    Returns:
+        bool: True if data should be filtered (too many NaN values), False otherwise
+
+    """
+    data = (
+        predictions.replace({None: np.nan})
+        .drop(columns=['timestamp'], errors='ignore')
+        .infer_objects()
+    )
+
+    if data.isna().all().all():
+        return True
+
+    return False

laborious/utils/import_password.py

@@ -0,0 +1,98 @@
+"""The bundle password envelope: AES-256-GCM, opened inside the activity that uses the password.
+
+The workflow only ever carries `password_envelope`, so no plaintext password reaches the Temporal
+event history. This module is imported by the bundle-opening activity and by nothing else — in
+particular not by `laborious/workflows/import_model.py`, which a test asserts.
+
+Envelope layout, chosen because WebCrypto gives the Angular frontend AES-GCM natively:
+
+    base64( nonce(12 bytes) || ciphertext || GCM tag(16 bytes) )
+
+The key is a 32-byte value held in a Kubernetes Secret and exposed as `IMPORT_PASSWORD_KEY`,
+base64-encoded. Neither the key, the envelope nor the recovered password is ever logged, notified,
+used as a metric label or written to the import record.
+"""
+
+from __future__ import annotations
+
+import base64
+import binascii
+from typing import Final
+
+from cryptography.exceptions import InvalidTag
+from cryptography.hazmat.primitives.ciphers.aead import AESGCM
+
+NONCE_SIZE: Final[int] = 12
+TAG_SIZE: Final[int] = 16
+KEY_SIZE: Final[int] = 32
+MIN_ENVELOPE_SIZE: Final[int] = NONCE_SIZE + TAG_SIZE
+
+
+class EnvelopeError(Exception):
+    """The envelope or its key is unusable.
+
+    Deliberately one type for every cause — missing key, wrong key, malformed base64, altered
+    bytes — because the failure is reported at the decryption step either way. What it must never
+    be confused with is a wrong *bundle* password: that one comes from the reader's gate 3, and the
+    two carry different sentences.
+    """
+
+
+def load_envelope_key(raw_key: str | None) -> bytes:
+    """Decode the configured envelope key.
+
+    Args:
+        raw_key: the value of `IMPORT_PASSWORD_KEY` — base64 of exactly 32 bytes.
+
+    Returns:
+        bytes: the 32-byte key.
+
+    Raises:
+        EnvelopeError: when the value is absent, not base64, or not 32 bytes long. There is no
+            fallback interpretation: an unset key never means "the envelope is plaintext".
+    """
+    if not raw_key:
+        raise EnvelopeError('the import password key is not configured')
+    try:
+        key = base64.b64decode(raw_key, validate=True)
+    except (binascii.Error, ValueError) as error:
+        raise EnvelopeError('the import password key is not valid base64') from error
+    if len(key) != KEY_SIZE:
+        raise EnvelopeError(f'the import password key must be {KEY_SIZE} bytes')
+    return key
+
+
+def decrypt_password_envelope(envelope: str, key: bytes) -> str:
+    """Recover the bundle password from its envelope.
+
+    Args:
+        envelope: base64 of `nonce || ciphertext || tag`.
+        key: the 32-byte key from `load_envelope_key`.
+
+    Returns:
+        str: the plaintext password, exactly as it was typed — not trimmed, not case-folded. NFC
+            normalisation happens at key derivation, where the producer does it.
+
+    Raises:
+        EnvelopeError: malformed envelope, wrong key or altered bytes (AES-GCM authentication).
+    """
+    if not envelope:
+        raise EnvelopeError('the import request carries no password envelope')
+    try:
+        payload = base64.b64decode(envelope, validate=True)
+    except (binascii.Error, ValueError) as error:
+        raise EnvelopeError('the password envelope is not valid base64') from error
+    if len(payload) < MIN_ENVELOPE_SIZE:
+        raise EnvelopeError('the password envelope is too short to hold a nonce and a tag')
+
+    nonce, sealed = payload[:NONCE_SIZE], payload[NONCE_SIZE:]
+    try:
+        plaintext = AESGCM(key).decrypt(nonce, sealed, None)
+    except InvalidTag as error:
+        raise EnvelopeError('the password envelope could not be authenticated') from error
+    except ValueError as error:
+        raise EnvelopeError('the password envelope could not be opened') from error
+    try:
+        return plaintext.decode('utf-8')
+    except UnicodeDecodeError as error:
+        raise EnvelopeError('the recovered password is not valid UTF-8') from error

laborious/utils/models/minio_dataframe_payload.py

@@ -0,0 +1,348 @@
+"""
+MinIO-backed DataFrame payload for Temporal workflows.
+
+Data is never stored as a pandas ``DataFrame`` field on the dataclass.
+Instead, the DataFrame is only provided as an input to:
+`from_dataframe` / `from_dataframe_to_dict`.
+
+At build time, the DataFrame is evaluated for its serialized size; if it exceeds
+the configured threshold, it is serialized to parquet bytes and uploaded to MinIO.
+Otherwise, it is inlined as a Temporal-friendly ``dict``.
+"""
+
+import pickle
+import re
+from collections.abc import Hashable
+from dataclasses import dataclass
+from datetime import datetime
+from io import BytesIO
+from os import getenv
+from typing import Any, Literal
+
+from pandas import DataFrame, read_parquet
+from sientia_do.observability.logger import Logger
+from sientia_do.repository.minio_repository import MinioRepository
+from sientia_do.temporal.constants import DATETIME_FORMAT_FILENAME, DATETIME_FORMAT_WITH_TZ, now
+
+# Keys that are part of the serialized wire format (not arbitrary metadata).
+_SERIALIZED_FIELD_KEYS = frozenset({'data', 'bucket', 'object_key', 'object_prefix', 'uri'})
+
+_OBJECT_TIMESTAMP_PATTERN = re.compile(
+    r'-(?:initial|transform)-(\d{4}-\d{2}-\d{2}_\d{2}-\d{2}-\d{2})\.parquet$'
+)
+
+OFFLOAD_THRESHOLD_BYTES = int(
+    float(getenv('SIENTIA_MINIO_OFFLOAD_THRESHOLD_MEGABYTES', '1.5')) * 1024 * 1024
+)
+
+# Relative prefix used for storing offloaded prediction datasets in MinIO.
+# It is also the root directory for retention cleanup listing.
+PREDICTION_DATASETS_PREFIX = 'prediction_datasets'
+
+OperationKind = Literal['initial', 'transform', 'predict']
+
+
+def _build_object_key(
+    model_name: str, operation: OperationKind, timestamp: str
+) -> tuple[str, str | None]:
+    """
+    Build the MinIO object key and the directory prefix used for retention listing.
+
+    Args:
+        model_name: Registered model name used in the pipeline.
+        operation: Either initial (pre-transform load) or transform (post-MLFlow transform).
+        timestamp: Filename timestamp segment from DATETIME_FORMAT_FILENAME.
+
+    Return:
+        tuple[str, str | None]: Full object key and normalized prefix (or None if at bucket root).
+    """
+    # Naming convention:
+    #   - Directory is always `prediction_datasets/<model_name>`
+    #   - Filename follows the retention-parsing pattern
+    basename = f'{model_name}-{operation}-{timestamp}.parquet'
+    model_dir = model_name.strip().strip('/')
+    prefix = f'{PREDICTION_DATASETS_PREFIX}/{model_dir}'
+    return f'{prefix}/{basename}', prefix
+
+
+@dataclass
+class MinioDataFramePayload:
+    """
+    Serializable payload after a DataFrame was evaluated: inline tabular dict and/or MinIO keys.
+
+    Build from a live DataFrame only via `from_dataframe` / `from_dataframe_to_dict`.
+    Rehydrate from Temporal via `from_dict`. The DataFrame is not a field on this class.
+    """
+
+    last_timestamp: str
+    status: dict[str, Any] | None = None
+    data: dict[Hashable, Any] | None = None
+    bucket: str | None = None
+    object_key: str | None = None
+    object_prefix: str | None = None
+    uri: str | None = None
+
+    @staticmethod
+    def _debug(
+        logger: Logger | None,
+        message: str,
+        metadata: dict[str, Any] | None = None,
+    ) -> None:
+        """
+        Emit debug logs only when logger is provided
+
+        Args:
+         - logger (Logger | None): Logger instance used for debug messages
+         - message (str): Message to be logged
+         - metadata (dict[str, Any] | None): Optional workflow metadata context
+        """
+        if logger is None:
+            return
+        logger.custom_debug(message, metadata)
+
+    @classmethod
+    def from_dict(cls, raw: 'dict[str, Any] | MinioDataFramePayload') -> 'MinioDataFramePayload':
+        """
+        Reconstruct a MinioDataFramePayload from a plain dict produced by Temporal serialization.
+
+        Temporal converts dataclass return values into plain dicts when crossing
+        workflow/activity boundaries. This method rebuilds the typed instance so
+        that methods like ``retrieve``, ``cleanup_prefix`` and ``has_data`` are
+        available on the receiving side.
+
+        If the argument is already a MinioDataFramePayload, it is returned as-is.
+
+        Args:
+            raw: Dict with keys matching the dataclass fields
+                (last_timestamp, status, data, bucket, object_key, object_prefix, uri),
+                or an existing MinioDataFramePayload instance.
+
+        Return:
+            MinioDataFramePayload: Reconstructed (or original) instance.
+        """
+        if isinstance(raw, MinioDataFramePayload):
+            return raw
+        return cls(
+            last_timestamp=raw['last_timestamp'],
+            status=raw.get('status'),
+            data=raw.get('data'),
+            bucket=raw.get('bucket'),
+            object_key=raw.get('object_key'),
+            object_prefix=raw.get('object_prefix'),
+            uri=raw.get('uri'),
+        )
+
+    @staticmethod
+    def estimate_size_bytes(
+        df: DataFrame,
+        metadata: dict[str, Any] | None = None,
+        logger: Logger | None = None,
+    ) -> int:
+        """
+        Approximate serialized size of the DataFrame as the default-orient dict.
+
+        Args:
+            df: DataFrame whose tabular content size is estimated.
+
+        Return:
+            int: Estimated size in bytes (pickle of dict representation).
+        """
+        try:
+            size = len(pickle.dumps(df.to_dict()))
+        except Exception:
+            size = len(pickle.dumps(df))
+
+        MinioDataFramePayload._debug(
+            logger,
+            f'DataFrame size: {size} bytes',
+            metadata,
+        )
+        return size
+
+    @staticmethod
+    def parse_object_timestamp(object_key: str) -> datetime | None:
+        """
+        Parse the timestamp embedded in the object key basename (before .parquet).
+
+        Args:
+            object_key: S3/MinIO object key whose basename follows
+                ``{model}-{initial|transform}-{DATETIME_FORMAT_FILENAME}.parquet``.
+
+        Return:
+            datetime | None: Parsed UTC-naive datetime from the key, or None if not matched.
+        """
+        basename = object_key.rsplit('/', 1)[-1]
+        match = _OBJECT_TIMESTAMP_PATTERN.search(basename)
+        if not match:
+            return None
+        try:
+            return datetime.strptime(match.group(1), DATETIME_FORMAT_FILENAME)
+        except ValueError:
+            return None
+
+    def cleanup_prefix(self) -> str | None:
+        """
+        Return True if cleanup is enabled for this payload.
+        """
+        if self.object_key is not None and self.data is None:
+            return self.object_prefix
+        return None
+
+    def has_data(self) -> bool:
+        """
+        Return True if the payload has some data internally or in MinIO.
+        """
+        return (self.data is not None and self.data != {}) or self.object_key is not None
+
+    @classmethod
+    async def from_dataframe(
+        cls,
+        dataframe: DataFrame | None,
+        minio_repo: MinioRepository,
+        model_name: str,
+        operation: OperationKind,
+        status: dict[str, Any] | None = None,
+        workflow_metadata: dict | None = None,
+        last_timestamp: str | None = None,
+        logger: Logger | None = None,
+    ) -> 'MinioDataFramePayload':
+        """
+        Evaluate the DataFrame size, then either inline dict or upload parquet to MinIO.
+
+        The DataFrame is not stored on the returned instance.
+
+        Args:
+            dataframe: Tabular data to evaluate and persist (inline or MinIO).
+            metadata: Small metadata dict merged into the payload (e.g. success, message).
+            minio_repo: sientia_do MinioRepository (or compatible) with `upload_file()`.
+            workflow_metadata: Metadata passed to MinIO store for logging/metrics.
+            model_name: Registered model name used in the object basename.
+            operation: Either ``initial`` (query load) or ``transform`` (post-transform).
+            key_prefix: Backward-compatible parameter (currently ignored for object naming).
+            size_threshold_bytes: Byte limit before offload. When None, the module-level
+                environment-derived default is used.
+
+        Return:
+            MinioDataFramePayload: Instance with data and/or MinIO fields set.
+        """
+
+        if dataframe is None or dataframe.empty:
+            cls._debug(
+                logger,
+                'MinioDataFramePayload.from_dataframe received empty dataframe, returning empty payload',
+                workflow_metadata,
+            )
+            return cls(
+                data=None, last_timestamp=now().strftime(DATETIME_FORMAT_WITH_TZ), status=status
+            )
+
+        if last_timestamp is None:
+            last_timestamp = max(dataframe['timestamp'].values.tolist())
+
+        dataframe_size = cls.estimate_size_bytes(dataframe, workflow_metadata, logger)
+        cls._debug(
+            logger,
+            (
+                f'MinioDataFramePayload.from_dataframe estimated size: {dataframe_size} bytes '
+                f'(threshold: {OFFLOAD_THRESHOLD_BYTES} bytes)'
+            ),
+            workflow_metadata,
+        )
+
+        if dataframe_size <= OFFLOAD_THRESHOLD_BYTES:
+            cls._debug(
+                logger,
+                'MinioDataFramePayload.from_dataframe using inline payload',
+                workflow_metadata,
+            )
+            return cls(data=dataframe.to_dict(), last_timestamp=last_timestamp, status=status)
+
+        timestamp = now().strftime(DATETIME_FORMAT_FILENAME)
+        object_key, object_prefix = _build_object_key(model_name, operation, timestamp)
+        cls._debug(
+            logger,
+            (
+                'MinioDataFramePayload.from_dataframe offloading payload to MinIO '
+                f'with key {object_key}'
+            ),
+            workflow_metadata,
+        )
+
+        # Upload using the relative object key. The upstream repository will
+        # prefix it internally under its MinIO namespace.
+        parquet_buffer = BytesIO()
+        dataframe.to_parquet(parquet_buffer, engine='pyarrow', index=True)
+        file_bytes = parquet_buffer.getvalue()
+
+        upload_result = await minio_repo.upload_file(
+            file_bytes=file_bytes,
+            relative_key=object_key,
+            metadata=workflow_metadata,
+        )
+
+        bucket = minio_repo.bucket
+        object_key_full = upload_result.get('minio_object_name', object_key)
+        uri = f's3://{bucket}/{object_key_full}' if bucket else None
+        cls._debug(
+            logger,
+            f'MinioDataFramePayload.from_dataframe upload completed: {uri}',
+            workflow_metadata,
+        )
+
+        return cls(
+            data=None,
+            bucket=bucket,
+            object_key=object_key_full,
+            object_prefix=object_prefix,
+            uri=uri,
+            last_timestamp=last_timestamp,
+            status=status,
+        )
+
+    async def retrieve(
+        self,
+        minio_repo: MinioRepository,
+        workflow_metadata: dict[str, Any] | None = None,
+        logger: Logger | None = None,
+    ) -> DataFrame:
+        """
+        Load parquet from MinIO when object_key is set and populate inline data.
+
+        Args:
+            minio_repo: sientia_do MinioRepository (or compatible) with download_file().
+            workflow_metadata: Metadata passed to MinIO read for logging/metrics.
+
+        Return:
+            dict[str, Any]: Flat dict with data filled (same keys as to_dict after load).
+        """
+        if self.data is not None:
+            self._debug(
+                logger,
+                'MinioDataFramePayload.retrieve using inline payload data',
+                workflow_metadata,
+            )
+            return DataFrame(self.data)
+
+        if not self.has_data():
+            self._debug(
+                logger,
+                'MinioDataFramePayload.retrieve found no payload data, returning empty dataframe',
+                workflow_metadata,
+            )
+            return DataFrame()
+
+        self._debug(
+            logger,
+            f'MinioDataFramePayload.retrieve downloading object from MinIO: {self.object_key}',
+            workflow_metadata,
+        )
+        file_bytes = await minio_repo.download_file(
+            object_name=self.object_key, metadata=workflow_metadata
+        )
+        df = read_parquet(BytesIO(file_bytes))
+        self._debug(
+            logger,
+            f'MinioDataFramePayload.retrieve loaded dataframe from MinIO with shape {df.shape}',
+            workflow_metadata,
+        )
+        return df

laborious/utils/repository/minio_manager.py

@@ -0,0 +1,32 @@
+from sientia_do.notifications.handlers import NotificationHandler
+from sientia_do.observability.logger import Logger
+from sientia_do.observability.metrics_controller import MetricsController
+from sientia_do.observability.sientia_monitoring import SientiaMonitoring
+from sientia_do.repository.minio_repository import MinioRepository
+
+
+class MinioManager(SientiaMonitoring):
+    minio_repository: MinioRepository | None = None
+
+    def __init__(
+        self,
+        minio_repository: MinioRepository | None = None,
+        logger: Logger | None = None,
+        notification_handler: NotificationHandler | None = None,
+        metrics_controller: MetricsController | None = None,
+    ):
+        if self.minio_repository is None:
+            self.minio_repository = minio_repository
+        SientiaMonitoring.__init__(self, logger, notification_handler, metrics_controller)
+
+    def close(self) -> None:
+        """
+        Close the MinioManager and clean up resources.
+        """
+        if self.minio_repository is not None:
+            try:
+                self.minio_repository.close()
+            finally:
+                self.minio_repository = None
+
+        SientiaMonitoring.shutdown(self)

laborious/utils/repository/opc_repository.py

@@ -0,0 +1,866 @@
+import asyncio
+import json
+import time
+import traceback
+from datetime import datetime
+from pathlib import Path
+from typing import Any
+
+from asyncua import Client
+from asyncua.crypto.security_policies import SecurityPolicyBasic256
+from asyncua.ua import DataValue, Variant, VariantType
+from asyncua.ua.uaerrors import UaStatusCodeError
+from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler
+from sientia_do.notifications.models import NotificationLevel
+from sientia_do.observability.logger import Logger
+from sientia_do.observability.metrics_controller import MetricsController
+from sientia_do.observability.sientia_monitoring import SientiaMonitoring
+
+from laborious import metrics
+
+# Requested session and secure channel lifetime (ms) before server revision; 10 minutes.
+OPC_UA_SESSION_AND_CHANNEL_TIMEOUT_MS = 10 * 60 * 1000
+
+
+class OpcClientAlreadyExistsError(RuntimeError):
+    """Raised when _create_client is called while self.client is already set."""
+
+
+class OpcSessionAlreadyConnectedError(RuntimeError):
+    """Raised when _open_session is called while a UA session is already open."""
+
+
+class OpcClientNotInitializedError(RuntimeError):
+    """Raised when _open_session is called before _create_client."""
+
+
+RECONNECTABLE_OPC_BAD_NAMES: frozenset[str] = frozenset(
+    {
+        'BadSessionIdInvalid',
+        'BadSessionClosed',
+        'BadSessionNotActivated',
+        'BadSecureChannelIdInvalid',
+        'BadSecureChannelClosed',
+        'BadSecureChannelTokenUnknown',
+        'BadTcpSecureChannelUnknown',
+        'BadServerNotConnected',
+        'BadConnectionClosed',
+        'BadDisconnect',
+        'BadConnectionRejected',
+        'BadCommunicationError',
+        'BadRequestInterrupted',
+        'BadUnknownResponse',
+        'BadTimeout',
+        'BadRequestTimeout',
+        'BadSequenceNumberInvalid',
+        'BadSequenceNumberUnknown',
+        'BadSecurityModeInsufficient',
+        'BadRequestHeaderInvalid',
+        'BadInvalidState',
+    }
+)
+
+
+def _opc_authentication_token_str(client: Client | None) -> str:
+    """
+    Serialize the current OPC UA authentication token (session handle) for logging and metrics.
+
+    Return:
+        str: Token string, or "unknown" if unavailable.
+    """
+    if client is None:
+        return 'unknown'
+    try:
+        proto = client.uaclient.protocol
+        if proto is None:
+            return 'unknown'
+        tok = getattr(proto, 'authentication_token', None)
+        if tok is None:
+            return 'unknown'
+        return str(tok)
+    except Exception:
+        return 'unknown'
+
+
+def _opc_status_from_exception(exc: BaseException) -> str:
+    """
+    Resolve OPC UA status name from an exception, including chained UaStatusCodeError causes.
+
+    Args:
+        exc (BaseException): Raised error from asyncua.
+
+    Return:
+        str: Status class name or generic Python exception name.
+    """
+    current: BaseException | None = exc
+    while current is not None:
+        if isinstance(current, UaStatusCodeError):
+            return type(current).__name__
+        current = current.__cause__
+    return type(exc).__name__
+
+
+def is_reconnectable_opcua_bad(exc: BaseException) -> bool:
+    """
+    Return whether the exception is a Tier-1 OPC UA Bad* that should trigger reconnect.
+
+    Args:
+        exc (BaseException): Raised error from get_node or write_value.
+
+    Return:
+        bool: True if reconnect should be scheduled.
+    """
+    return _opc_status_from_exception(exc) in RECONNECTABLE_OPC_BAD_NAMES
+
+
+def _model_labels_from_write_metadata(metadata: dict[str, Any] | None) -> dict[str, str]:
+    """
+    Extract model_id and model_name from write metadata for Prometheus labels.
+
+    Args:
+        metadata (dict[str, Any] | None): Context passed into write_data; may omit keys.
+
+    Return:
+        dict[str, str]: Labels model_id and model_name, defaulting to "unknown".
+    """
+    if not metadata:
+        return {'model_id': 'unknown', 'model_name': 'unknown'}
+    return {
+        'model_id': str(metadata.get('model_id', 'unknown')),
+        'model_name': str(metadata.get('model_name', 'unknown')),
+    }
+
+
+data_type_map = {
+    'float': {
+        'converter': float,
+        'opc_type': VariantType.Float,
+    },
+    'double': {
+        'converter': float,
+        'opc_type': VariantType.Double,
+    },
+    'int': {
+        'converter': int,
+        'opc_type': VariantType.Int32,
+    },
+    'bool': {
+        'converter': bool,
+        'opc_type': VariantType.Boolean,
+    },
+    'str': {
+        'converter': str,
+        'opc_type': VariantType.String,
+    },
+}
+
+
+class OpcRepository(SientiaMonitoring):
+    def __init__(
+        self,
+        opc_id: str,
+        url: str,
+        server_name: str,
+        logger: Logger,
+        notification_handler: NotificationHandler,
+        metrics_controller: MetricsController,
+        reconnection_interval: int = 60,
+        server_uri: str | None = None,
+        cert_path: str | None = None,
+        private_key_path: str | None = None,
+        server_cert_path: str | None = None,
+    ):
+        self.url = url
+        self.id = opc_id
+        self.server_name = server_name
+        self.server_uri = server_uri
+        self.cert_path = cert_path
+        self.private_key_path = private_key_path
+        self.server_cert_path = server_cert_path
+        self.reconnection_interval = reconnection_interval
+        self.last_reconnection_time: None | datetime = None
+        self.disconnection_interval = 10.0
+        self.notification_handler = notification_handler
+        self.client: None | Client = None
+
+        SientiaMonitoring.__init__(self, logger, notification_handler, metrics_controller)
+
+        self.metadata = {
+            'model_name': '-',
+            'model_id': '-',
+            'workflow_name': 'opc_repository',
+            'schedule_name': '-',
+        }
+        self._last_write_mono: float | None = None
+        self._connection_lock = asyncio.Lock()
+        self._session_ready = asyncio.Event()
+        self._reconnect_task: asyncio.Task[None] | None = None
+        self._allow_reconnect = True
+
+    def _opc_debug_tags(self, session_id: str) -> dict[str, str]:
+        """
+        Build Prometheus/log label tags for OPC session-scoped metrics.
+
+        Args:
+            session_id (str): OPC UA session token string.
+
+        Return:
+            dict[str, str]: Labels pod_id, server_name, runtime, opc_server_id, session_id.
+        """
+        return {
+            'pod_id': str(getattr(self, 'pod_id', 'unknown')),
+            'server_name': self.server_name,
+            'runtime': str(getattr(self, 'runtime', 'unknown')),
+            'opc_server_id': self.id,
+            'session_id': session_id,
+        }
+
+    def _is_session_open(self) -> bool:
+        """
+        Return whether the asyncua client has an open transport session.
+
+        Return:
+            bool: True when protocol exists and is not closed.
+        """
+        if self.client is None:
+            return False
+        try:
+            proto = self.client.uaclient.protocol
+            return proto is not None and proto.state != 'closed'
+        except Exception:
+            return False
+
+    def _reconnection_window_elapsed(self) -> bool:
+        """
+        Return whether enough time has passed since the last reconnect attempt.
+
+        Return:
+            bool: True if a new reconnect is allowed.
+        """
+        if self.last_reconnection_time is None:
+            return True
+        return (
+            datetime.now() - self.last_reconnection_time
+        ).total_seconds() > self.reconnection_interval
+
+    def _not_connected_error(self) -> dict[str, Any]:
+        """
+        Build the standard error payload when validate_connection finds no open protocol.
+
+        Return:
+            dict[str, Any]: Notification fields for OPC_CONNECTION_NOT_READY.
+        """
+        return {
+            'notification_id': f'OPC_CONNECTION_NOT_READY_{self.id}',
+            'message': f'OPC server {self.id} is not connected',
+            'block': 'opc_repository',
+            'level': NotificationLevel.WARNING,
+        }
+
+    async def set_security(self) -> None:
+        """
+        Configure certificates and timeouts on the asyncua client.
+
+        Raises:
+            ValueError: If cert paths or client are missing.
+        """
+        if self.cert_path is None or self.private_key_path is None:
+            raise ValueError(
+                'Certificate and private key paths must be provided for secure connection.'
+            )
+
+        cert = Path(self.cert_path)
+        private_key = Path(self.private_key_path)
+        server_cert = Path(self.server_cert_path) if self.server_cert_path else None
+
+        if self.client is None:
+            raise ValueError('Client must be initialized before setting security')
+
+        self.client.application_uri = self.server_uri
+        self.info('Setting security...', self.metadata)
+        await self.client.set_security(
+            SecurityPolicyBasic256,
+            certificate=str(cert),
+            private_key=str(private_key),
+            server_certificate=str(server_cert) if server_cert else None,
+        )
+        self.client.secure_channel_timeout = OPC_UA_SESSION_AND_CHANNEL_TIMEOUT_MS
+        self.client.session_timeout = OPC_UA_SESSION_AND_CHANNEL_TIMEOUT_MS
+
+    async def _create_client(self) -> None:
+        """
+        Instantiate the asyncua Client and apply security when configured.
+
+        Caller must hold _connection_lock. Does not open a UA session.
+
+        Raises:
+            OpcClientAlreadyExistsError: If self.client is already set.
+        """
+        if self.client is not None:
+            raise OpcClientAlreadyExistsError(
+                f'OPC client already exists for server {self.id}; '
+                'call disconnect() before creating a new client'
+            )
+
+        self.client = Client(self.url, timeout=10, watchdog_intervall=50)  # type: ignore[attr-defined]
+        self.client.name = self.pod_id
+        self.client.application_name = self.pod_id
+        pod_uri = self.pod_id.replace('-', ':')
+        self.client.application_uri = pod_uri
+        self.client.product_uri = pod_uri
+        if self.cert_path:
+            await self.set_security()
+
+    async def _open_session(self) -> tuple[bool, dict[str, Any]]:
+        """
+        Open the OPC UA session on the existing client.
+
+        Caller must hold _connection_lock.
+
+        Raises:
+            OpcClientNotInitializedError: If self.client is None.
+            OpcSessionAlreadyConnectedError: If a session is already open.
+
+        Return:
+            tuple[bool, dict[str, Any]]: Success flag and error payload on connect failure.
+        """
+        if self.client is None:
+            raise OpcClientNotInitializedError(
+                f'OPC client is not initialized for server {self.id}; '
+                'call _create_client() before opening a session'
+            )
+        if self._is_session_open():
+            raise OpcSessionAlreadyConnectedError(
+                f'OPC session already connected for server {self.id}; '
+                'call disconnect() before connecting again'
+            )
+
+        tags = {
+            'pod_id': self.pod_id,
+            'server_name': self.server_name,
+        }
+        await self.emit_metric(metrics.OPC_CONNECTIONS_TOTAL, tags)
+
+        try:
+            await self.client.connect()
+
+            session_id = _opc_authentication_token_str(self.client)
+            revised_session_timeout_ms = int(self.client.session_timeout)
+            revised_secure_channel_timeout_ms = int(self.client.secure_channel_timeout)
+            self.info(
+                f'OPC new session connected opc_server_id={self.id} session_id={session_id} '
+                f'revised_session_timeout_ms={revised_session_timeout_ms} '
+                f'revised_secure_channel_timeout_ms={revised_secure_channel_timeout_ms}',
+                self.metadata,
+            )
+            await self.emit_metric(
+                metrics.OPC_SESSION_CREATED_TOTAL, self._opc_debug_tags(session_id)
+            )
+            await self.emit_metric(
+                metric_object=metrics.OPC_SESSION_REVISED_TIMEOUT_MS,
+                method='set',
+                tags=self._opc_debug_tags(session_id),
+                value=revised_session_timeout_ms,
+            )
+            await self.emit_metric(
+                metric_object=metrics.OPC_CONNECTION_STATUS,
+                method='set',
+                tags={**tags, 'server_url': self.url},
+                value=1,
+            )
+
+            self._last_write_mono = None
+            self._session_ready.set()
+            return True, {}
+
+        except Exception as e:
+            await self._disconnect_locked()
+            trace = traceback.format_exc()
+            self.error(trace, self.metadata)
+            await self.emit_metric(metrics.OPC_CONNECTIONS_FAILED, tags)
+            return False, {
+                'notification_id': f'OPC_CONNECTION_ERROR_{self.id}',
+                'message': f'Failed to connect to OPC server: {e}',
+                'block': 'opc_repository',
+                'level': NotificationLevel.ERROR,
+                'attachment_content': trace,
+            }
+
+    async def _connect_locked(self) -> tuple[bool, dict[str, Any]]:
+        """
+        Create the client when absent, then open a UA session.
+
+        Caller must hold _connection_lock.
+
+        Raises:
+            OpcSessionAlreadyConnectedError: If a session is already open.
+
+        Return:
+            tuple[bool, dict[str, Any]]: Result from _open_session on connect failure.
+        """
+        if self._is_session_open():
+            raise OpcSessionAlreadyConnectedError(
+                f'OPC session already connected for server {self.id}; '
+                'call disconnect() before connecting again'
+            )
+        if self.client is None:
+            await self._create_client()
+        return await self._open_session()
+
+    async def _disconnection_fallback(self) -> list[dict[str, Any]]:
+        """
+        Try up to five times to disconnect from the OPC UA server.
+        """
+        assert self.client is not None
+        error_stack: list[dict[str, Any]] = []
+        for i in range(5):
+            try:
+                self.info(
+                    f'Disconnecting from OPC UA server, attempt {i + 1} of 5',
+                    self.metadata,
+                )
+                await self.client.disconnect()
+                return []
+            except Exception as e:
+                self.error(
+                    f'Failed to disconnect from OPC UA server in attempt {i + 1} of 5: {e}',
+                    self.metadata,
+                )
+                error_stack.append(
+                    {
+                        'attempt': i + 1,
+                        'error': str(e),
+                        'traceback': traceback.format_exc(),
+                    }
+                )
+                await asyncio.sleep(self.disconnection_interval * i)
+        return error_stack
+
+    async def _disconnect_locked(self) -> None:
+        """
+        Tear down the current session and client.
+
+        Caller must hold _connection_lock.
+        """
+        self._last_write_mono = None
+        self._session_ready.clear()
+
+        if self.client is None:
+            return
+
+        session_id = _opc_authentication_token_str(self.client)
+        self.info(
+            f'OPC disconnecting opc_server_id={self.id} session_id={session_id}',
+            self.metadata,
+        )
+        await self.emit_metric(metrics.OPC_SESSION_CLOSED_TOTAL, self._opc_debug_tags(session_id))
+
+        errors = await self._disconnection_fallback()
+        if errors:
+            await self.send_notification_async(
+                metadata=self.metadata,
+                notification_id=f'OPC_DISCONNECTION_ERROR_{self.id}',
+                message='Failed to disconnect from OPC server in 5 attempts.',
+                block='opc_repository',
+                level=NotificationLevel.ERROR,
+                attachment_content=json.dumps(errors, indent=4),
+            )
+        else:
+            self.warning(f'Disconnected from OPC server {self.id} successfully', self.metadata)
+
+        await self.emit_metric(
+            metric_object=metrics.OPC_CONNECTION_STATUS,
+            method='set',
+            tags={
+                'pod_id': self.pod_id,
+                'server_name': self.server_name,
+                'server_url': self.url,
+            },
+            value=0,
+        )
+        self.client = None
+
+    async def _reconnect_locked(self) -> tuple[bool, dict[str, Any]]:
+        """
+        Close the current session and open a new one.
+
+        Caller must hold _connection_lock. Records last_reconnection_time for interval gating.
+
+        Return:
+            tuple[bool, dict[str, Any]]: Result from _connect_locked after teardown.
+        """
+        self.last_reconnection_time = datetime.now()
+        await self._disconnect_locked()
+        return await self._connect_locked()
+
+    async def connect(self) -> tuple[bool, dict[str, Any]]:
+        """
+        Open an OPC UA session under the connection lock (worker initialization).
+        """
+        async with self._connection_lock:
+            self.info(
+                f'Starting connection to OPC server {self.id}:{self.server_name}...',
+                self.metadata,
+            )
+            return await self._connect_locked()
+
+    async def disconnect(self) -> None:
+        """
+        Gracefully disconnect from the OPC server under the connection lock.
+
+        Disables background reconnect so late writes during worker shutdown do not
+        respawn sessions.
+        """
+        async with self._connection_lock:
+            self._allow_reconnect = False
+            await self._disconnect_locked()
+
+    async def validate_connection(self) -> tuple[bool, dict[str, Any]]:
+        """
+        Read-only check that the asyncua protocol is open.
+
+        Caller must ensure _session_ready before writing. Does not connect or reconnect.
+
+        Return:
+            tuple[bool, dict[str, Any]]: (True, {}) when open, otherwise (False, error).
+        """
+        if self._is_session_open():
+            return True, {}
+        self.error(f'OPC server {self.id} is not connected', self.metadata)
+        return False, self._not_connected_error()
+
+    def _reconnect_task_in_progress(self) -> bool:
+        """
+        Return whether a background reconnect task is currently running.
+
+        Return:
+            bool: True when a reconnect task exists and has not finished.
+        """
+        return self._reconnect_task is not None and not self._reconnect_task.done()
+
+    async def _start_reconnect(self, reason: str, session_id: str) -> None:
+        """
+        Schedule a background reconnect when allowed by interval and task state.
+
+        Clears _session_ready before starting the task. No-op when _allow_reconnect is
+        False, the reconnection window has not elapsed, or a reconnect is already running.
+
+        Args:
+            reason (str): Trigger for reconnect (OPC status name or synthetic reason).
+            session_id (str): Session token before failure.
+        """
+        if not self._allow_reconnect:
+            return
+        if not self._reconnection_window_elapsed():
+            self.warning(
+                f'OPC reconnect skipped reason=reconnection_window opc_server_id={self.id} '
+                f'reconnect_reason={reason}',
+                self.metadata,
+            )
+            return
+        if self._reconnect_task_in_progress():
+            self.warning(
+                f'OPC reconnect skipped reason=in_progress opc_server_id={self.id} '
+                f'reconnect_reason={reason}',
+                self.metadata,
+            )
+            return
+
+        self._session_ready.clear()
+        self.info(
+            f'OPC reconnect scheduled reconnect_reason={reason} opc_server_id={self.id} '
+            f'old_session_id={session_id}',
+            self.metadata,
+        )
+        self._reconnect_task = asyncio.create_task(self._run_reconnect(reason, session_id))
+
+    async def _run_reconnect(self, reason: str, session_id: str) -> None:
+        """
+        Background task that tears down and re-establishes the OPC UA session.
+
+        Args:
+            reason (str): Trigger for reconnect (OPC status or ProtocolClosed).
+            session_id (str): Previous session token string for logging.
+        """
+        try:
+            async with self._connection_lock:
+                self.info(
+                    f'OPC reconnect started reconnect_reason={reason} opc_server_id={self.id} '
+                    f'old_session_id={session_id}',
+                    self.metadata,
+                )
+                success, error = await self._reconnect_locked()
+            if not success:
+                self.error(
+                    f'OPC reconnect failed reconnect_reason={reason} opc_server_id={self.id}',
+                    self.metadata,
+                )
+                if error:
+                    self.error(error.get('message', ''), self.metadata)
+        except Exception:
+            self.error(
+                f'OPC reconnect task failed opc_server_id={self.id} reconnect_reason={reason}',
+                self.metadata,
+            )
+            self.error(traceback.format_exc(), self.metadata)
+
+    async def _log_write_inter_arrival(self, session_id: str, node: str) -> None:
+        """
+        Log elapsed wall time since the previous successful OPC write on this repository.
+
+        Args:
+            session_id (str): Current OPC UA session token string.
+            node (str): Node id written in this operation.
+        """
+        now = time.monotonic()
+        if self._last_write_mono is not None:
+            delta_s = now - self._last_write_mono
+            self.info(
+                f'OPC write inter-arrival_s={delta_s:.6f} opc_server_id={self.id} '
+                f'session_id={session_id} node={node}',
+                self.metadata,
+            )
+            if self.client is not None:
+                session_timeout_ms = float(self.client.session_timeout)
+                if session_timeout_ms > 0 and delta_s > (session_timeout_ms / 1000.0):
+                    await self.emit_metric(
+                        metrics.OPC_WRITE_INTER_ARRIVAL_OVER_SESSION_TIMEOUT_TOTAL,
+                        self._opc_debug_tags(session_id),
+                    )
+        self._last_write_mono = now
+
+    async def _emit_opc_write_metric(
+        self, session_id: str, result: str, metadata: dict[str, Any] | None
+    ) -> None:
+        """
+        Emit opc_write_attempts_total for a single write attempt outcome.
+
+        Args:
+            session_id (str): OPC UA session token string, or "unknown".
+            result (str): Outcome label (OK, OPC status name, ProtocolClosed, etc.).
+            metadata (dict[str, Any] | None): Write context for model_id/model_name labels.
+        """
+        await self.emit_metric(
+            metrics.OPC_WRITE_ATTEMPTS_TOTAL,
+            {
+                **self._opc_debug_tags(session_id),
+                **_model_labels_from_write_metadata(metadata),
+                'result': result,
+            },
+        )
+
+    def _write_failure_payload(
+        self,
+        notification_id: str,
+        message: str,
+        level: NotificationLevel = NotificationLevel.ERROR,
+        attachment_content: str | None = None,
+        opc_error_kind: str | None = None,
+        opc_status: str | None = None,
+    ) -> dict[str, Any]:
+        """
+        Build a structured error dict returned from failed write_data paths.
+
+        Args:
+            notification_id (str): Stable notification identifier.
+            message (str): Human-readable failure message.
+            level (NotificationLevel): Severity for downstream notifications.
+            attachment_content (str | None): Optional traceback or diagnostic text.
+            opc_error_kind (str | None): Classifier (session_bad, connection_lost, etc.).
+            opc_status (str | None): OPC UA status name or synthetic reason.
+
+        Return:
+            dict[str, Any]: Error payload consumed by the OPC activity layer.
+        """
+        payload: dict[str, Any] = {
+            'notification_id': notification_id,
+            'message': message,
+            'block': 'opc_repository',
+            'level': level,
+        }
+        if attachment_content is not None:
+            payload['attachment_content'] = attachment_content
+        if opc_error_kind is not None:
+            payload['opc_error_kind'] = opc_error_kind
+        if opc_status is not None:
+            payload['opc_status'] = opc_status
+        return payload
+
+    async def _handle_tier1_bad(
+        self,
+        exc: BaseException,
+        session_id: str,
+        node: str,
+        metadata: dict[str, Any],
+        phase: str,
+    ) -> tuple[bool, dict[str, Any]]:
+        """
+        Record metrics/logs and schedule reconnect after a Tier-1 Bad* error.
+
+        Args:
+            exc (BaseException): Tier-1 OPC UA error.
+            session_id (str): Session token at failure time.
+            node (str): Node id being written.
+            metadata (dict[str, Any]): Write context.
+            phase (str): get_node or write_value.
+
+        Return:
+            tuple[bool, dict[str, Any]]: Always (False, error payload).
+        """
+        opc_status = _opc_status_from_exception(exc)
+        trace = traceback.format_exc()
+        self.error(trace, metadata)
+        await self._emit_opc_write_metric(session_id, opc_status, metadata)
+        self.error(
+            f'OPC write failed opc_status={opc_status} opc_server_id={self.id} '
+            f'session_id={session_id} model_id={metadata.get("model_id", "unknown")} '
+            f'model_name={metadata.get("model_name", "unknown")} node={node} phase={phase}',
+            metadata,
+        )
+        await self._start_reconnect(opc_status, session_id)
+        return False, self._write_failure_payload(
+            notification_id=f'OPC_WRITE_DATA_ERROR_{self.id}',
+            message=f'Failed to {phase} on OPC server: {exc} | metadata: {metadata}',
+            attachment_content=trace,
+            opc_error_kind='session_bad',
+            opc_status=opc_status,
+        )
+
+    async def _write_reconnect_in_progress(
+        self, metadata: dict[str, Any]
+    ) -> tuple[bool, dict[str, Any]]:
+        """
+        Fail a write because a background reconnect task is already running.
+
+        Args:
+            metadata (dict[str, Any]): Write context passed through to the activity.
+
+        Return:
+            tuple[bool, dict[str, Any]]: (False, error info with opc_error_kind reconnect_in_progress).
+        """
+        await self._emit_opc_write_metric('unknown', 'ReconnectInProgress', metadata)
+        self.warning(
+            f'OPC write rejected reconnect_in_progress opc_server_id={self.id} '
+            f'model_id={metadata.get("model_id", "unknown")} '
+            f'model_name={metadata.get("model_name", "unknown")}',
+            metadata,
+        )
+        return False, {
+            'notification_id': f'OPC_WRITE_RECONNECT_IN_PROGRESS_{self.id}',
+            'message': f'OPC write skipped: reconnect in progress | metadata: {metadata}',
+            'block': 'opc_repository',
+            'level': NotificationLevel.WARNING,
+            'opc_error_kind': 'reconnect_in_progress',
+        }
+
+    async def _write_connection_lost(
+        self, metadata: dict[str, Any], opc_status: str
+    ) -> tuple[bool, dict[str, Any]]:
+        """
+        Fail a write after scheduling reconnect for a closed or stale session.
+
+        Args:
+            metadata (dict[str, Any]): Write context passed through to the activity.
+            opc_status (str): Synthetic reason (ProtocolClosed, SessionNotReady).
+
+        Return:
+            tuple[bool, dict[str, Any]]: (False, error info with opc_error_kind connection_lost).
+        """
+        await self._emit_opc_write_metric('unknown', opc_status, metadata)
+        return False, self._write_failure_payload(
+            notification_id=f'OPC_WRITE_CONNECTION_LOST_{self.id}',
+            message=f'OPC write skipped: connection lost ({opc_status}) | metadata: {metadata}',
+            level=NotificationLevel.WARNING,
+            opc_error_kind='connection_lost',
+            opc_status=opc_status,
+        )
+
+    async def write_data(
+        self, node: str, value: Any, data_type: str, metadata: dict[str, Any]
+    ) -> tuple[bool, dict[str, Any]]:
+        """
+        Write data to OPC server with a single attempt and background reconnect scheduling.
+
+        Reconnect is scheduled on Tier-1 Bad*, closed protocol, or stale session readiness.
+        There is no retry within the same call.
+
+        Args:
+            node (str): OPC UA node id to write.
+            value (Any): Value to convert and send.
+            data_type (str): Logical type key (float, int, bool, str, double).
+            metadata (dict[str, Any]): Activity context (model_id, model_name, etc.).
+
+        Return:
+            tuple[bool, dict[str, Any]]: (True, {response_time}) on success, or
+                (False, structured error info) on failure.
+        """
+        if self._reconnect_task_in_progress():
+            return await self._write_reconnect_in_progress(metadata)
+
+        if not self._session_ready.is_set():
+            session_id = _opc_authentication_token_str(self.client)
+            await self._start_reconnect('SessionNotReady', session_id)
+            if self._reconnect_task_in_progress():
+                return await self._write_reconnect_in_progress(metadata)
+            return await self._write_connection_lost(metadata, 'SessionNotReady')
+
+        is_connected, _error = await self.validate_connection()
+        if not is_connected:
+            session_id = _opc_authentication_token_str(self.client)
+            await self._start_reconnect('ProtocolClosed', session_id)
+            return await self._write_connection_lost(metadata, 'ProtocolClosed')
+
+        start_time = time.time()
+        session_id = _opc_authentication_token_str(self.client)
+
+        try:
+            node_obj = self.client.get_node(node)  # type: ignore[union-attr]
+        except Exception as e:
+            if is_reconnectable_opcua_bad(e):
+                return await self._handle_tier1_bad(e, session_id, node, metadata, 'get_node')
+            trace = traceback.format_exc()
+            self.error(trace, metadata)
+            await self._emit_opc_write_metric(
+                session_id, f'GetNodeError:{type(e).__name__}', metadata
+            )
+            return False, self._write_failure_payload(
+                notification_id=f'OPC_WRITE_GET_NODE_ERROR_{self.id}',
+                message=f'Failed to get node from OPC server: {e} | metadata: {metadata}',
+                attachment_content=trace,
+            )
+
+        if data_type not in data_type_map:
+            await self._emit_opc_write_metric(session_id, 'UnsupportedDataType', metadata)
+            return False, self._write_failure_payload(
+                notification_id=f'OPC_WRITE_DATA_TYPE_ERROR_{self.id}',
+                message=f'Unsupported data type: {data_type} | metadata: {metadata}',
+            )
+
+        data = data_type_map[data_type]['converter'](value)
+        self.info(f'Writing {data} - {type(data)} to {node}', metadata)
+        ua_data = DataValue(
+            Variant(data, data_type_map[data_type]['opc_type']),
+        )
+
+        try:
+            await node_obj.write_value(ua_data)
+            end_time = time.time()
+            response_time = end_time - start_time
+        except Exception as e:
+            if is_reconnectable_opcua_bad(e):
+                return await self._handle_tier1_bad(e, session_id, node, metadata, 'write_value')
+            trace = traceback.format_exc()
+            self.error(trace, metadata)
+            await self._emit_opc_write_metric(session_id, type(e).__name__, metadata)
+            return False, self._write_failure_payload(
+                notification_id=f'OPC_WRITE_DATA_ERROR_{self.id}',
+                message=f'Failed to write data to OPC server: {e} | metadata: {metadata}',
+                attachment_content=trace,
+            )
+
+        await self._emit_opc_write_metric(session_id, 'OK', metadata)
+        await self._log_write_inter_arrival(session_id, node)
+
+        return True, {
+            'response_time': response_time,
+        }