perf(observatory): pack camera epoch source transfer

2026-08-31 17:59:18 +03:00
parent 1766c0bdb2
commit 4cd94b5805
5 changed files with 1164 additions and 7 deletions
@@ -16,11 +16,12 @@ import os
 import re
 import secrets
 import stat
+import tarfile
 from collections.abc import AsyncIterable, Mapping, Sequence
 from contextlib import suppress
 from dataclasses import dataclass
 from pathlib import Path
-from typing import Final, Literal
+from typing import BinaryIO, Final, Literal, cast

 from k1link.observatory.portable_result_contract import (
    OBSERVATION_ONLY_AUTHORITY,
@@ -59,6 +60,16 @@ from k1link.sessions.store import SessionStore
 PORTABLE_SOURCE_MATERIALIZATION_SCHEMA: Final = (
    "missioncore.observatory-portable-source-materialization/v1"
 )
+PORTABLE_CAMERA_EPOCH_ARCHIVE_SCHEMA: Final = (
+    "missioncore.observatory-portable-camera-epoch-archive/v1"
+)
+PORTABLE_CAMERA_EPOCH_ARCHIVE_BINDING_SCHEMA: Final = (
+    "missioncore.observatory-portable-camera-epoch-archive-binding/v1"
+)
+PORTABLE_CAMERA_EPOCH_ARCHIVE_MEDIA_TYPE: Final = "application/x-tar"
+PORTABLE_CAMERA_EPOCH_ARCHIVE_ID_HEADER: Final = (
+    "X-Mission-Core-Camera-Epoch-Archive-Id"
+)
 PORTABLE_RESULT_UPLOAD_PLAN_SCHEMA: Final = (
    "missioncore.observatory-portable-result-upload-plan/v1"
 )
@@ -73,6 +84,10 @@ PORTABLE_RESULT_STAGING_DIRECTORY: Final = "observatory-worker-result-staging"

 MAX_SOURCE_MEMBERS: Final = 100_000
 MAX_SOURCE_BYTES: Final = 2 * 1024 * 1024 * 1024 * 1024
+MAX_CAMERA_EPOCH_ARCHIVE_BYTES: Final = (
+    MAX_SOURCE_BYTES + MAX_SOURCE_MEMBERS * 1024 + tarfile.RECORDSIZE
+)
+MAX_CAMERA_EPOCH_ARCHIVE_BINDING_BYTES: Final = 8 * 1024 * 1024
 MAX_RESULT_MEMBER_BYTES: Final = 64 * 1024 * 1024 * 1024
 MAX_RESULT_PACKAGE_BYTES: Final = 256 * 1024 * 1024 * 1024
 MAX_RESULT_MANIFEST_BYTES: Final = 1024 * 1024
@@ -228,6 +243,53 @@ class PortableSourceMaterializationManifest:
        }


+@dataclass(frozen=True, slots=True)
+class PortableCameraEpochArchive:
+    archive_id: str
+    job_identity_sha256: str
+    source_bundle_sha256: str
+    artifact_id: str
+    camera_epoch: int
+    member_ids: tuple[str, ...]
+    byte_length: int
+    sha256: str
+
+    def __post_init__(self) -> None:
+        _digest(self.archive_id, "camera epoch archive id")
+        _digest(self.job_identity_sha256, "camera epoch archive job identity")
+        _digest(self.source_bundle_sha256, "camera epoch archive source bundle")
+        _session_id(self.artifact_id)
+        _positive_int(self.camera_epoch, "camera epoch archive ordinal")
+        if (
+            not 2 <= len(self.member_ids) <= MAX_SOURCE_MEMBERS
+            or len(set(self.member_ids)) != len(self.member_ids)
+        ):
+            raise ValueError("camera epoch archive members are invalid")
+        for member_id in self.member_ids:
+            _digest(member_id, "camera epoch archive member id")
+        if not 1 <= self.byte_length <= MAX_CAMERA_EPOCH_ARCHIVE_BYTES:
+            raise ValueError("camera epoch archive byte length is invalid")
+        _digest(self.sha256, "camera epoch archive sha256")
+        if self.archive_id != portable_camera_epoch_archive_id(
+            job_identity_sha256=self.job_identity_sha256,
+            source_bundle_sha256=self.source_bundle_sha256,
+            artifact_id=self.artifact_id,
+            camera_epoch=self.camera_epoch,
+            member_ids=self.member_ids,
+            byte_length=self.byte_length,
+            sha256=self.sha256,
+        ):
+            raise ValueError("camera epoch archive identity is invalid")
+
+    def as_dict(self) -> dict[str, object]:
+        return {
+            "archive_id": self.archive_id,
+            "media_type": PORTABLE_CAMERA_EPOCH_ARCHIVE_MEDIA_TYPE,
+            "byte_length": self.byte_length,
+            "sha256": self.sha256,
+        }
+
+
@dataclass(frozen=True, slots=True)
 class PortableResultUploadMember:
    member_id: str
@@ -443,6 +505,39 @@ class PortableObservatoryArtifactTransport:
        )
        return member, destination

+    def materialize_camera_epoch_archive(
+        self,
+        *,
+        job_id: str,
+        claim_token: str,
+        claim_generation: int,
+        claimant_id: str,
+    ) -> tuple[PortableCameraEpochArchive, Path]:
+        """Publish one claim-bound deterministic tar for the admitted camera epoch."""
+
+        job = self._authorize(
+            job_id=job_id,
+            claim_token=claim_token,
+            claim_generation=claim_generation,
+            claimant_id=claimant_id,
+            allowed_states=("claimed", "running"),
+        )
+        manifest = self._resolve_source_manifest(job)
+        members = _ordered_camera_epoch_members(manifest.members)
+        archive, path = _publish_camera_epoch_archive(
+            self._source_cas,
+            job=job,
+            members=members,
+        )
+        self._authorize(
+            job_id=job_id,
+            claim_token=claim_token,
+            claim_generation=claim_generation,
+            claimant_id=claimant_id,
+            allowed_states=("claimed", "running"),
+        )
+        return archive, path
+
    def stage_result_manifest(
        self,
        *,
@@ -1033,6 +1128,418 @@ class PortableObservatoryArtifactTransport:
        return receipt


+def portable_camera_epoch_archive_id(
+    *,
+    job_identity_sha256: str,
+    source_bundle_sha256: str,
+    artifact_id: str,
+    camera_epoch: int,
+    member_ids: Sequence[str],
+    byte_length: int,
+    sha256: str,
+) -> str:
+    """Return the fixed identity of one packed, content-sealed camera epoch."""
+
+    _digest(job_identity_sha256, "camera epoch archive job identity")
+    _digest(source_bundle_sha256, "camera epoch archive source bundle")
+    _session_id(artifact_id)
+    _positive_int(camera_epoch, "camera epoch archive ordinal")
+    canonical_member_ids = tuple(member_ids)
+    if (
+        not 2 <= len(canonical_member_ids) <= MAX_SOURCE_MEMBERS
+        or len(set(canonical_member_ids)) != len(canonical_member_ids)
+    ):
+        raise ValueError("camera epoch archive members are invalid")
+    for member_id in canonical_member_ids:
+        _digest(member_id, "camera epoch archive member id")
+    if not 1 <= byte_length <= MAX_CAMERA_EPOCH_ARCHIVE_BYTES:
+        raise ValueError("camera epoch archive byte length is invalid")
+    _digest(sha256, "camera epoch archive sha256")
+    return hashlib.sha256(
+        canonical_json(
+            {
+                "schema_version": PORTABLE_CAMERA_EPOCH_ARCHIVE_SCHEMA,
+                "job_identity_sha256": job_identity_sha256,
+                "source_bundle_sha256": source_bundle_sha256,
+                "artifact_id": artifact_id,
+                "camera_epoch": camera_epoch,
+                "member_ids": list(canonical_member_ids),
+                "media_type": PORTABLE_CAMERA_EPOCH_ARCHIVE_MEDIA_TYPE,
+                "byte_length": byte_length,
+                "sha256": sha256,
+            }
+        )
+    ).hexdigest()
+
+
+def _ordered_camera_epoch_members(
+    members: Sequence[PortableSourceMember],
+) -> tuple[PortableSourceMember, ...]:
+    inits = tuple(member for member in members if member.kind == "camera-init")
+    segments = tuple(member for member in members if member.kind == "camera-segment")
+    if len(inits) != 1 or not segments:
+        raise PortableArtifactTransportIntegrityError(
+            "portable camera epoch member inventory is incomplete"
+        )
+    init = inits[0]
+    if init.artifact_id is None or init.camera_epoch is None:
+        raise PortableArtifactTransportIntegrityError(
+            "portable camera epoch identity is incomplete"
+        )
+    ordered_segments = tuple(
+        sorted(segments, key=lambda member: member.camera_sequence or 0)
+    )
+    if tuple(member.camera_sequence for member in ordered_segments) != tuple(
+        range(1, len(ordered_segments) + 1)
+    ) or any(
+        member.artifact_id != init.artifact_id
+        or member.camera_epoch != init.camera_epoch
+        for member in ordered_segments
+    ):
+        raise PortableArtifactTransportIntegrityError(
+            "portable camera epoch member order changed"
+        )
+    return (init, *ordered_segments)
+
+
+def _camera_archive_relative_name(member: PortableSourceMember) -> str:
+    if member.kind == "camera-init":
+        return "init.mp4"
+    if member.kind != "camera-segment" or member.camera_sequence is None:
+        raise PortableArtifactTransportIntegrityError(
+            "portable camera archive member role is invalid"
+        )
+    return f"segments/{member.camera_sequence}.m4s"
+
+
+def _camera_epoch_archive_byte_length(
+    members: Sequence[PortableSourceMember],
+) -> int:
+    content_bytes = sum(
+        512 + ((member.byte_length + 511) // 512) * 512 for member in members
+    )
+    logical_bytes = content_bytes + 1024
+    return ((logical_bytes + tarfile.RECORDSIZE - 1) // tarfile.RECORDSIZE) * (
+        tarfile.RECORDSIZE
+    )
+
+
+def _camera_epoch_archive_binding_identity(
+    job: ObservatoryRecordedJob,
+    members: Sequence[PortableSourceMember],
+) -> dict[str, object]:
+    ordered = _ordered_camera_epoch_members(members)
+    init = ordered[0]
+    assert init.artifact_id is not None
+    assert init.camera_epoch is not None
+    return {
+        "schema_version": PORTABLE_CAMERA_EPOCH_ARCHIVE_BINDING_SCHEMA,
+        "job_identity_sha256": job.identity_sha256,
+        "source_bundle_sha256": job.source_bundle_sha256,
+        "artifact_id": init.artifact_id,
+        "camera_epoch": init.camera_epoch,
+        "member_ids": [member.member_id for member in ordered],
+    }
+
+
+def _load_camera_epoch_archive_binding(
+    path: Path,
+    *,
+    binding_id: str,
+    identity: Mapping[str, object],
+    archive_root: Path,
+) -> tuple[PortableCameraEpochArchive, Path]:
+    try:
+        payload = _read_exact_regular_file(
+            path,
+            maximum_bytes=MAX_CAMERA_EPOCH_ARCHIVE_BINDING_BYTES,
+        )
+        document = object_document(
+            json.loads(payload.decode("utf-8")),
+            "camera epoch archive binding",
+        )
+        if canonical_json(document) != payload:
+            raise PortableArtifactTransportIntegrityError(
+                "camera epoch archive binding is not canonical"
+            )
+    except PortableArtifactTransportError:
+        raise
+    except (
+        OSError,
+        UnicodeDecodeError,
+        json.JSONDecodeError,
+        PortableResultPackageIntegrityError,
+        ValueError,
+    ) as exc:
+        raise PortableArtifactTransportIntegrityError(
+            "camera epoch archive binding is unreadable"
+        ) from exc
+    if (
+        set(document) != {*identity, "binding_id", "archive", "authority"}
+        or any(document.get(key) != value for key, value in identity.items())
+        or document.get("binding_id") != binding_id
+        or document.get("authority") != OBSERVATION_ONLY_AUTHORITY
+    ):
+        raise PortableArtifactTransportIntegrityError(
+            "camera epoch archive binding identity changed"
+        )
+    try:
+        archive_document = object_document(
+            document.get("archive"),
+            "camera epoch archive",
+        )
+    except PortableResultPackageIntegrityError as exc:
+        raise PortableArtifactTransportIntegrityError(
+            "camera epoch archive seal is invalid"
+        ) from exc
+    if (
+        set(archive_document)
+        != {"archive_id", "media_type", "byte_length", "sha256"}
+        or archive_document.get("media_type")
+        != PORTABLE_CAMERA_EPOCH_ARCHIVE_MEDIA_TYPE
+    ):
+        raise PortableArtifactTransportIntegrityError(
+            "camera epoch archive seal changed"
+        )
+    try:
+        member_ids_value = identity["member_ids"]
+        if not isinstance(member_ids_value, list) or any(
+            not isinstance(member_id, str) for member_id in member_ids_value
+        ):
+            raise ValueError("camera epoch archive member ids are invalid")
+        result = PortableCameraEpochArchive(
+            archive_id=string(archive_document.get("archive_id"), "archive id"),
+            job_identity_sha256=string(
+                identity["job_identity_sha256"],
+                "archive job identity",
+            ),
+            source_bundle_sha256=string(
+                identity["source_bundle_sha256"],
+                "archive source bundle",
+            ),
+            artifact_id=string(identity["artifact_id"], "archive artifact id"),
+            camera_epoch=_positive_int(
+                identity["camera_epoch"],
+                "archive camera epoch",
+            ),
+            member_ids=tuple(member_ids_value),
+            byte_length=_positive_int(
+                archive_document.get("byte_length"),
+                "archive byte length",
+            ),
+            sha256=string(archive_document.get("sha256"), "archive sha256"),
+        )
+    except (PortableResultPackageIntegrityError, ValueError) as exc:
+        raise PortableArtifactTransportIntegrityError(
+            "camera epoch archive binding is invalid"
+        ) from exc
+    archive_path = archive_root / result.sha256[:2] / result.sha256
+    if not _matches_exact_file(
+        archive_path,
+        expected_sha256=result.sha256,
+        expected_byte_length=result.byte_length,
+    ):
+        raise PortableArtifactTransportIntegrityError(
+            "camera epoch archive differs from its immutable binding"
+        )
+    return result, archive_path
+
+
+def _publish_camera_epoch_archive(
+    root: Path,
+    *,
+    job: ObservatoryRecordedJob,
+    members: Sequence[PortableSourceMember],
+) -> tuple[PortableCameraEpochArchive, Path]:
+    ordered = _ordered_camera_epoch_members(members)
+    init = ordered[0]
+    assert init.artifact_id is not None
+    assert init.camera_epoch is not None
+    archive_root = _prepare_secure_root(root / "camera-epoch-archives")
+    binding_identity = _camera_epoch_archive_binding_identity(job, ordered)
+    binding_id = hashlib.sha256(canonical_json(binding_identity)).hexdigest()
+    binding_root = _prepare_secure_root(archive_root / "bindings")
+    binding_path = binding_root / f"{binding_id}.json"
+    if binding_path.exists():
+        return _load_camera_epoch_archive_binding(
+            binding_path,
+            binding_id=binding_id,
+            identity=binding_identity,
+            archive_root=archive_root,
+        )
+    for member in ordered:
+        if member._source_path is None:
+            raise PortableArtifactTransportIntegrityError(
+                "portable camera archive member has no admitted source"
+            )
+    temporary = archive_root / f".tmp-{secrets.token_hex(16)}"
+    descriptor = os.open(
+        temporary,
+        os.O_WRONLY
+        | os.O_CREAT
+        | os.O_EXCL
+        | getattr(os, "O_NOFOLLOW", 0),
+        0o600,
+    )
+    try:
+        with os.fdopen(descriptor, "wb") as stream:
+            descriptor = -1
+            with tarfile.open(
+                fileobj=stream,
+                mode="w",
+                format=tarfile.USTAR_FORMAT,
+            ) as archive:
+                for member in ordered:
+                    info = tarfile.TarInfo(_camera_archive_relative_name(member))
+                    info.size = member.byte_length
+                    info.mode = 0o600
+                    info.mtime = 0
+                    info.uid = 0
+                    info.gid = 0
+                    info.uname = ""
+                    info.gname = ""
+                    assert member._source_path is not None
+                    _add_exact_camera_archive_member(
+                        archive,
+                        info,
+                        member._source_path,
+                        expected_sha256=member.sha256,
+                        expected_byte_length=member.byte_length,
+                    )
+            stream.flush()
+            os.fsync(stream.fileno())
+        sha256, byte_length = _hash_regular_file(
+            temporary,
+            maximum_bytes=MAX_CAMERA_EPOCH_ARCHIVE_BYTES,
+        )
+        if byte_length != _camera_epoch_archive_byte_length(ordered):
+            raise PortableArtifactTransportIntegrityError(
+                "portable camera epoch archive size is not deterministic"
+            )
+        result = PortableCameraEpochArchive(
+            archive_id=portable_camera_epoch_archive_id(
+                job_identity_sha256=job.identity_sha256,
+                source_bundle_sha256=job.source_bundle_sha256,
+                artifact_id=init.artifact_id,
+                camera_epoch=init.camera_epoch,
+                member_ids=tuple(member.member_id for member in ordered),
+                byte_length=byte_length,
+                sha256=sha256,
+            ),
+            job_identity_sha256=job.identity_sha256,
+            source_bundle_sha256=job.source_bundle_sha256,
+            artifact_id=init.artifact_id,
+            camera_epoch=init.camera_epoch,
+            member_ids=tuple(member.member_id for member in ordered),
+            byte_length=byte_length,
+            sha256=sha256,
+        )
+        destination_root = _prepare_secure_root(
+            archive_root / result.sha256[:2]
+        )
+        destination = destination_root / result.sha256
+        _publish_uploaded_file(
+            temporary,
+            destination,
+            expected_sha256=result.sha256,
+            expected_byte_length=result.byte_length,
+        )
+        _write_immutable_file(
+            binding_path,
+            canonical_json(
+                {
+                    **binding_identity,
+                    "binding_id": binding_id,
+                    "archive": result.as_dict(),
+                    "authority": dict(OBSERVATION_ONLY_AUTHORITY),
+                }
+            ),
+        )
+        return result, destination
+    except PortableArtifactTransportError:
+        raise
+    except (OSError, tarfile.TarError, ValueError) as exc:
+        raise PortableArtifactTransportUnavailableError(
+            "portable camera epoch archive could not be materialized"
+        ) from exc
+    finally:
+        if descriptor >= 0:
+            os.close(descriptor)
+        with suppress(FileNotFoundError):
+            temporary.unlink()
+
+
+class _ExactArchiveSource:
+    def __init__(self, stream: BinaryIO, *, expected_byte_length: int) -> None:
+        self._stream = stream
+        self._expected_byte_length = expected_byte_length
+        self._digest = hashlib.sha256()
+        self.byte_length = 0
+
+    def read(self, size: int = -1) -> bytes:
+        payload = self._stream.read(size)
+        self.byte_length += len(payload)
+        if self.byte_length > self._expected_byte_length:
+            raise PortableArtifactTransportIntegrityError(
+                "portable camera archive source grew while read"
+            )
+        if not payload and self.byte_length != self._expected_byte_length:
+            raise PortableArtifactTransportIntegrityError(
+                "portable camera archive source ended early"
+            )
+        self._digest.update(payload)
+        return payload
+
+    @property
+    def sha256(self) -> str:
+        return self._digest.hexdigest()
+
+
+def _add_exact_camera_archive_member(
+    archive: tarfile.TarFile,
+    info: tarfile.TarInfo,
+    source: Path,
+    *,
+    expected_sha256: str,
+    expected_byte_length: int,
+) -> None:
+    descriptor = os.open(
+        source,
+        os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0),
+    )
+    with os.fdopen(descriptor, "rb") as source_stream:
+        before = os.fstat(source_stream.fileno())
+        if not stat.S_ISREG(before.st_mode) or before.st_size != expected_byte_length:
+            raise PortableArtifactTransportIntegrityError(
+                "portable camera archive source is not the admitted regular file"
+            )
+        verified = _ExactArchiveSource(
+            source_stream,
+            expected_byte_length=expected_byte_length,
+        )
+        archive.addfile(info, cast(BinaryIO, verified))
+        after = os.fstat(source_stream.fileno())
+        stable = (
+            before.st_dev,
+            before.st_ino,
+            before.st_size,
+            before.st_mtime_ns,
+        ) == (
+            after.st_dev,
+            after.st_ino,
+            after.st_size,
+            after.st_mtime_ns,
+        )
+        if (
+            verified.byte_length != expected_byte_length
+            or verified.sha256 != expected_sha256
+            or not stable
+        ):
+            raise PortableArtifactTransportIntegrityError(
+                "portable camera archive source changed while read"
+            )
+
+
 def _source_member(
    job: ObservatoryRecordedJob,
    *,
@@ -14,21 +14,28 @@ import json
 import os
 import re
 import secrets
+import shutil
 import stat
+import tarfile
+import tempfile
 import threading
 from collections.abc import Mapping
 from contextlib import suppress
 from dataclasses import dataclass
 from pathlib import Path
-from typing import Final, Literal, cast
+from typing import BinaryIO, Final, Literal, cast
 from urllib.parse import urlsplit

 import httpx

 from k1link.observatory.portable_artifact_transport import (
+    MAX_CAMERA_EPOCH_ARCHIVE_BYTES,
+    PORTABLE_CAMERA_EPOCH_ARCHIVE_ID_HEADER,
+    PORTABLE_CAMERA_EPOCH_ARCHIVE_MEDIA_TYPE,
    PORTABLE_RESULT_UPLOAD_PLAN_SCHEMA,
    PORTABLE_RESULT_UPLOAD_RECEIPT_SCHEMA,
    PORTABLE_SOURCE_MATERIALIZATION_SCHEMA,
+    portable_camera_epoch_archive_id,
 )
 from k1link.observatory.portable_result_contract import (
    OBSERVATION_ONLY_AUTHORITY,
@@ -54,6 +61,7 @@ from k1link.observatory.worker_agent import (
 WORKER_HTTP_MAX_JSON_BYTES: Final = 16 * 1024 * 1024
 WORKER_HTTP_COPY_CHUNK_BYTES: Final = 1024 * 1024
 WORKER_HTTP_DEFAULT_TIMEOUT_SECONDS: Final = 30.0
+WORKER_HTTP_CAMERA_ARCHIVE_READ_TIMEOUT_SECONDS: Final = 30 * 60.0

 _SHA256 = re.compile(r"^[a-f0-9]{64}$")
 _JOB_ID = re.compile(r"^observatory-run-[a-f0-9]{32}$")
@@ -136,6 +144,7 @@ class ObservatoryWorkerHttpGateway(ObservatoryWorkerTransport):
            _CONTOUR_HEADER: contour_id,
            "Accept": "application/json",
        }
+        self._timeout_seconds = timeout_seconds
        self._work_root = _secure_directory(work_root)
        self._client = httpx.Client(
            base_url=self._base_url,
@@ -304,6 +313,21 @@ class ObservatoryWorkerHttpGateway(ObservatoryWorkerTransport):
                    "source materialization members select the same local role"
                )
            destinations[destination] = member
+        camera_members = _ordered_camera_epoch_members(members)
+        if not all(
+            _matches_file(
+                _source_destination(root, member),
+                member.sha256,
+                member.byte_length,
+            )
+            for member in camera_members
+        ):
+            self._download_camera_epoch_archive(
+                job=job,
+                context=context,
+                members=camera_members,
+                root=root,
+            )
        for destination, member in destinations.items():
            if _matches_file(destination, member.sha256, member.byte_length):
                continue
@@ -463,6 +487,123 @@ class ObservatoryWorkerHttpGateway(ObservatoryWorkerTransport):
            result_sha256=draft.result_sha256,
        )

+    def _download_camera_epoch_archive(
+        self,
+        *,
+        job: SealedObservatoryRecordedJob,
+        context: _ClaimContext,
+        members: tuple[_SourceMember, ...],
+        root: Path,
+    ) -> None:
+        transfer_root = _secure_directory(self._work_root / ".source-transfers")
+        temporary = transfer_root / f".camera-epoch-{secrets.token_hex(16)}.tar"
+        descriptor = -1
+        try:
+            with self._client.stream(
+                "GET",
+                self._job_path(job.job_id, "source-camera-epoch-archive"),
+                headers=self._claim_headers(context),
+                timeout=httpx.Timeout(
+                    connect=self._timeout_seconds,
+                    read=WORKER_HTTP_CAMERA_ARCHIVE_READ_TIMEOUT_SECONDS,
+                    write=self._timeout_seconds,
+                    pool=self._timeout_seconds,
+                ),
+            ) as response:
+                if response.status_code == 404:
+                    return
+                self._raise_for_status(response)
+                if (
+                    response.headers.get("content-type")
+                    != PORTABLE_CAMERA_EPOCH_ARCHIVE_MEDIA_TYPE
+                ):
+                    raise ObservatoryWorkerHttpError(
+                        "camera epoch archive media type changed"
+                    )
+                declared = response.headers.get("content-length")
+                if (
+                    declared is None
+                    or not declared.isascii()
+                    or not declared.isdigit()
+                    or len(declared) > len(str(MAX_CAMERA_EPOCH_ARCHIVE_BYTES))
+                    or str(int(declared)) != declared
+                ):
+                    raise ObservatoryWorkerHttpError(
+                        "camera epoch archive length is invalid"
+                    )
+                expected_bytes = int(declared)
+                expected_sha256 = response.headers.get(_CONTENT_SHA_HEADER, "")
+                archive_id = response.headers.get(
+                    PORTABLE_CAMERA_EPOCH_ARCHIVE_ID_HEADER,
+                    "",
+                )
+                if (
+                    not 1 <= expected_bytes <= MAX_CAMERA_EPOCH_ARCHIVE_BYTES
+                    or _SHA256.fullmatch(expected_sha256) is None
+                    or _SHA256.fullmatch(archive_id) is None
+                ):
+                    raise ObservatoryWorkerHttpError(
+                        "camera epoch archive seal is invalid"
+                    )
+                descriptor = os.open(
+                    temporary,
+                    os.O_WRONLY
+                    | os.O_CREAT
+                    | os.O_EXCL
+                    | getattr(os, "O_NOFOLLOW", 0),
+                    0o600,
+                )
+                digest = hashlib.sha256()
+                byte_length = 0
+                with os.fdopen(descriptor, "wb") as stream:
+                    descriptor = -1
+                    for chunk in response.iter_bytes(WORKER_HTTP_COPY_CHUNK_BYTES):
+                        byte_length += len(chunk)
+                        if byte_length > expected_bytes:
+                            raise ObservatoryWorkerHttpError(
+                                "camera epoch archive exceeds its declared length"
+                            )
+                        digest.update(chunk)
+                        stream.write(chunk)
+                    stream.flush()
+                    os.fsync(stream.fileno())
+            if byte_length != expected_bytes or digest.hexdigest() != expected_sha256:
+                raise ObservatoryWorkerHttpError(
+                    "camera epoch archive content differs from its seal"
+                )
+            init = members[0]
+            assert init.artifact_id is not None
+            assert init.camera_epoch is not None
+            expected_archive_id = portable_camera_epoch_archive_id(
+                job_identity_sha256=job.identity_sha256,
+                source_bundle_sha256=job.source_bundle_sha256,
+                artifact_id=init.artifact_id,
+                camera_epoch=init.camera_epoch,
+                member_ids=tuple(member.member_id for member in members),
+                byte_length=byte_length,
+                sha256=expected_sha256,
+            )
+            if archive_id != expected_archive_id:
+                raise ObservatoryWorkerHttpError(
+                    "camera epoch archive identity changed"
+                )
+            _extract_camera_epoch_archive(
+                temporary,
+                root=root,
+                members=members,
+            )
+        except ObservatoryWorkerHttpError:
+            raise
+        except httpx.HTTPError as exc:
+            raise ObservatoryWorkerHttpError(
+                "camera epoch archive transport is unavailable"
+            ) from exc
+        finally:
+            if descriptor >= 0:
+                os.close(descriptor)
+            with suppress(FileNotFoundError):
+                temporary.unlink()
+
    def _download_member(
        self,
        *,
@@ -868,6 +1009,192 @@ def _source_member(value: object) -> _SourceMember:
    )


+def _ordered_camera_epoch_members(
+    members: tuple[_SourceMember, ...],
+) -> tuple[_SourceMember, ...]:
+    inits = tuple(member for member in members if member.kind == "camera-init")
+    segments = tuple(member for member in members if member.kind == "camera-segment")
+    if len(inits) != 1 or not segments:
+        raise ObservatoryWorkerHttpError(
+            "source materialization camera epoch is incomplete"
+        )
+    init = inits[0]
+    if init.artifact_id is None or init.camera_epoch is None:
+        raise ObservatoryWorkerHttpError(
+            "source materialization camera identity is incomplete"
+        )
+    ordered_segments = tuple(
+        sorted(segments, key=lambda member: member.camera_sequence or 0)
+    )
+    if tuple(member.camera_sequence for member in ordered_segments) != tuple(
+        range(1, len(ordered_segments) + 1)
+    ) or any(
+        member.artifact_id != init.artifact_id
+        or member.camera_epoch != init.camera_epoch
+        for member in ordered_segments
+    ):
+        raise ObservatoryWorkerHttpError(
+            "source materialization camera member order changed"
+        )
+    return (init, *ordered_segments)
+
+
+def _camera_archive_relative_name(member: _SourceMember) -> str:
+    if member.kind == "camera-init":
+        return "init.mp4"
+    if member.kind != "camera-segment" or member.camera_sequence is None:
+        raise ObservatoryWorkerHttpError("camera archive member role is invalid")
+    return f"segments/{member.camera_sequence}.m4s"
+
+
+def _camera_epoch_archive_byte_length(members: tuple[_SourceMember, ...]) -> int:
+    content_bytes = sum(
+        512 + ((member.byte_length + 511) // 512) * 512 for member in members
+    )
+    logical_bytes = content_bytes + 1024
+    return ((logical_bytes + tarfile.RECORDSIZE - 1) // tarfile.RECORDSIZE) * (
+        tarfile.RECORDSIZE
+    )
+
+
+def _extract_camera_epoch_archive(
+    archive_path: Path,
+    *,
+    root: Path,
+    members: tuple[_SourceMember, ...],
+) -> None:
+    expected_bytes = _camera_epoch_archive_byte_length(members)
+    try:
+        archive_metadata = archive_path.lstat()
+    except OSError as exc:
+        raise ObservatoryWorkerHttpError(
+            "camera epoch archive is unavailable"
+        ) from exc
+    if (
+        stat.S_ISLNK(archive_metadata.st_mode)
+        or not stat.S_ISREG(archive_metadata.st_mode)
+        or archive_metadata.st_size != expected_bytes
+    ):
+        raise ObservatoryWorkerHttpError(
+            "camera epoch archive size differs from its member inventory"
+        )
+    transfer_root = _secure_directory(archive_path.parent)
+    staging = Path(
+        tempfile.mkdtemp(prefix=".camera-epoch-extract-", dir=transfer_root)
+    )
+    staged: list[Path] = []
+    archive_descriptor = -1
+    try:
+        archive_descriptor = os.open(
+            archive_path,
+            os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0),
+        )
+        with os.fdopen(archive_descriptor, "rb") as archive_stream:
+            archive_descriptor = -1
+            with tarfile.open(fileobj=archive_stream, mode="r:") as archive:
+                for member in members:
+                    info = archive.next()
+                    if info is None:
+                        raise ObservatoryWorkerHttpError(
+                            "camera epoch archive member inventory changed"
+                        )
+                    relative_name = _camera_archive_relative_name(member)
+                    if (
+                        info.name != relative_name
+                        or info.type != tarfile.REGTYPE
+                        or info.size != member.byte_length
+                        or info.mode != 0o600
+                        or info.mtime != 0
+                        or info.uid != 0
+                        or info.gid != 0
+                        or info.uname != ""
+                        or info.gname != ""
+                        or info.linkname != ""
+                        or info.pax_headers
+                    ):
+                        raise ObservatoryWorkerHttpError(
+                            "camera epoch archive contains an unsafe member"
+                        )
+                    source = archive.extractfile(info)
+                    if source is None:
+                        raise ObservatoryWorkerHttpError(
+                            "camera epoch archive member is unreadable"
+                        )
+                    target = staging.joinpath(*relative_name.split("/"))
+                    _secure_directory(target.parent)
+                    with source:
+                        _copy_camera_archive_member(
+                            cast(BinaryIO, source),
+                            target,
+                            expected_sha256=member.sha256,
+                            expected_byte_length=member.byte_length,
+                        )
+                    staged.append(target)
+                if archive.next() is not None:
+                    raise ObservatoryWorkerHttpError(
+                        "camera epoch archive member inventory changed"
+                    )
+        for temporary, member in zip(staged, members, strict=True):
+            _publish_local_file(
+                temporary,
+                _source_destination(root, member),
+                member.sha256,
+                member.byte_length,
+            )
+    except ObservatoryWorkerHttpError:
+        raise
+    except (OSError, tarfile.TarError) as exc:
+        raise ObservatoryWorkerHttpError(
+            "camera epoch archive could not be extracted"
+        ) from exc
+    finally:
+        if archive_descriptor >= 0:
+            os.close(archive_descriptor)
+        shutil.rmtree(staging, ignore_errors=True)
+
+
+def _copy_camera_archive_member(
+    source: BinaryIO,
+    destination: Path,
+    *,
+    expected_sha256: str,
+    expected_byte_length: int,
+) -> None:
+    descriptor = os.open(
+        destination,
+        os.O_WRONLY
+        | os.O_CREAT
+        | os.O_EXCL
+        | getattr(os, "O_NOFOLLOW", 0),
+        0o600,
+    )
+    digest = hashlib.sha256()
+    byte_length = 0
+    try:
+        with os.fdopen(descriptor, "wb") as stream:
+            descriptor = -1
+            while chunk := source.read(WORKER_HTTP_COPY_CHUNK_BYTES):
+                byte_length += len(chunk)
+                if byte_length > expected_byte_length:
+                    raise ObservatoryWorkerHttpError(
+                        "camera epoch archive member exceeds its declared length"
+                    )
+                digest.update(chunk)
+                stream.write(chunk)
+            stream.flush()
+            os.fsync(stream.fileno())
+        if byte_length != expected_byte_length or digest.hexdigest() != expected_sha256:
+            raise ObservatoryWorkerHttpError(
+                "camera epoch archive member content changed"
+            )
+    finally:
+        if descriptor >= 0:
+            os.close(descriptor)
+        if not _matches_file(destination, expected_sha256, expected_byte_length):
+            with suppress(OSError):
+                destination.unlink()
+
+
 def _source_destination(root: Path, member: _SourceMember) -> Path:
    if member.kind == "source-bundle":
        return root / "source-bundle.json"
@@ -26,6 +26,8 @@ from pydantic import BaseModel, ConfigDict, Field

 from k1link.observatory.portable_artifact_transport import (
    MAX_RESULT_MANIFEST_BYTES,
+    PORTABLE_CAMERA_EPOCH_ARCHIVE_ID_HEADER,
+    PORTABLE_CAMERA_EPOCH_ARCHIVE_MEDIA_TYPE,
    PortableArtifactTransportError,
    PortableArtifactTransportIntegrityError,
    PortableArtifactTransportUnavailableError,
@@ -381,6 +383,41 @@ def build_observatory_worker_router(
                .as_dict()
            )

+        @router.get("/recorded-jobs/{job_id}/source-camera-epoch-archive")
+        def source_camera_epoch_archive(
+            job_id: Annotated[str, ApiPath(pattern=_JOB_ID_PATTERN)],
+            claim_token: Annotated[
+                str,
+                Header(
+                    alias=OBSERVATORY_WORKER_CLAIM_TOKEN_HEADER,
+                    pattern=_CLAIM_TOKEN_PATTERN,
+                ),
+            ],
+            claim_generation: Annotated[
+                int,
+                Header(alias=OBSERVATORY_WORKER_CLAIM_GENERATION_HEADER, ge=1),
+            ],
+        ) -> FileResponse:
+            archive, path = _artifact_call(
+                lambda: artifact_transport.materialize_camera_epoch_archive(
+                    job_id=job_id,
+                    claim_token=claim_token,
+                    claim_generation=claim_generation,
+                    claimant_id=authentication.contour_id,
+                )
+            )
+            return FileResponse(
+                path,
+                media_type=PORTABLE_CAMERA_EPOCH_ARCHIVE_MEDIA_TYPE,
+                headers={
+                    "ETag": f'"{archive.sha256}"',
+                    "Cache-Control": "private, no-store",
+                    "X-Content-Type-Options": "nosniff",
+                    "X-Mission-Core-Content-Sha256": archive.sha256,
+                    PORTABLE_CAMERA_EPOCH_ARCHIVE_ID_HEADER: archive.archive_id,
+                },
+            )
+
        @router.get("/recorded-jobs/{job_id}/source-members/{member_id}")
        def source_member(
            job_id: Annotated[str, ApiPath(pattern=_JOB_ID_PATTERN)],