feat: add durable background experiment runs

CONTEXT.md

@@ -0,0 +1,35 @@
+# AnteLab Experiment Execution
+
+This context names the durable lifecycle around one scientifically reproducible
+AnteLab experiment.
+
+## Language
+
+**Experiment Run**:
+A durable, observable execution of one immutable validated experiment request.
+_Avoid_: Job, task, process
+
+**Run Request**:
+The immutable scientific inputs and version bindings that identify what an
+Experiment Run must execute.
+_Avoid_: Job spec, payload
+
+**Run Attempt**:
+One actual computation of an Experiment Run. Recovery may create another Run
+Attempt without creating another Experiment Run.
+_Avoid_: Retry job, resumed run
+
+**Run Snapshot**:
+An immutable revision of an Experiment Run's current lifecycle, attempt, and
+progress facts.
+_Avoid_: Job status, mutable run
+
+**Published Artifact**:
+The immutable, scientifically validated result attached to a completed
+Experiment Run and safe for callers to retrieve.
+_Avoid_: Output file, partial result
+
+**Deterministic Replay**:
+Recovery that starts a new Run Attempt from tick zero using the same Run
+Request. It is not checkpoint resume.
+_Avoid_: Resume, continue

RUNNING.md

@@ -33,5 +33,5 @@ bash scripts/loops/verify-antelab-kernel-closure.sh
 ```
 
 The local closure budget is 5,000 ticks in at most 120 seconds and 512 MiB peak
-RSS. The stricter 30-second target remains a public V1 release gate; Phase A
-does not revise or claim to meet it.
+RSS. The 30-second target is non-blocking stretch telemetry: record it, but do
+not fail a run or block release solely because it exceeds 30 seconds.

antelab/runs/__init__.py

@@ -0,0 +1,27 @@
+"""Durable background Experiment Runs."""
+
+from antelab.runs.artifacts import InMemoryArtifactRepository
+from antelab.runs.errors import (
+    ArtifactIntegrityError,
+    ArtifactNotReadyError,
+    IdempotencyConflictError,
+    RunNotFoundError,
+    RunStateConflictError,
+)
+from antelab.runs.manager import ExperimentRuns
+from antelab.runs.model import RunId, RunRequest, RunSnapshot
+from antelab.runs.repository import InMemoryRunRepository
+
+__all__ = [
+    "ArtifactIntegrityError",
+    "ArtifactNotReadyError",
+    "ExperimentRuns",
+    "IdempotencyConflictError",
+    "InMemoryArtifactRepository",
+    "InMemoryRunRepository",
+    "RunId",
+    "RunNotFoundError",
+    "RunRequest",
+    "RunSnapshot",
+    "RunStateConflictError",
+]

antelab/runs/artifacts.py

@@ -0,0 +1,318 @@
+"""ArtifactRepository Interface and in-memory Adapter."""
+
+from __future__ import annotations
+
+import json
+import os
+from hashlib import sha256
+from pathlib import Path
+from typing import Protocol, cast
+from uuid import uuid4
+
+from antelab.artifacts import RunArtifact
+from antelab.runs.errors import ArtifactIntegrityError
+from antelab.runs.model import ArtifactRef
+
+
+class ArtifactRepository(Protocol):
+    def stage(self, artifact: RunArtifact) -> ArtifactRef: ...
+
+    def publish(self, reference: ArtifactRef) -> ArtifactRef: ...
+
+    def load(self, reference: ArtifactRef) -> RunArtifact: ...
+
+    def exists(self, reference: ArtifactRef) -> bool: ...
+
+
+class InMemoryArtifactRepository:
+    def __init__(self) -> None:
+        self._artifacts: dict[str, tuple[ArtifactRef, bytes]] = {}
+        self._staged: dict[str, tuple[ArtifactRef, bytes]] = {}
+
+    def stage(self, artifact: RunArtifact) -> ArtifactRef:
+        artifact.validate()
+        payload = (
+            json.dumps(
+                artifact.to_dict(),
+                ensure_ascii=True,
+                sort_keys=True,
+                separators=(",", ":"),
+                allow_nan=False,
+            )
+            + "\n"
+        ).encode("utf-8")
+        digest = sha256(payload).hexdigest()
+        reference = ArtifactRef(
+            artifact_id=f"sha256/{digest}",
+            sha256=digest,
+            byte_size=len(payload),
+            schema_version=artifact.schema_version,
+        )
+        parsed = _artifact_from_mapping(
+            cast(dict[str, object], json.loads(payload))
+        )
+        parsed.validate()
+        if _canonical_bytes(parsed) != payload:
+            raise ArtifactIntegrityError("staged artifact is not canonical")
+        self._staged[reference.artifact_id] = (reference, payload)
+        return reference
+
+    def publish(self, reference: ArtifactRef) -> ArtifactRef:
+        try:
+            staged_ref, payload = self._staged[reference.artifact_id]
+        except KeyError as error:
+            raise ArtifactIntegrityError("staged artifact is missing") from error
+        if staged_ref != reference or sha256(payload).hexdigest() != reference.sha256:
+            raise ArtifactIntegrityError("staged artifact metadata differs")
+        existing = self._artifacts.get(reference.artifact_id)
+        if existing is not None and existing[0] != reference:
+            raise ArtifactIntegrityError("immutable artifact identity conflicts")
+        self._artifacts[reference.artifact_id] = (reference, payload)
+        return reference
+
+    def load(self, reference: ArtifactRef) -> RunArtifact:
+        try:
+            stored_ref, payload = self._artifacts[reference.artifact_id]
+        except KeyError as error:
+            raise ArtifactIntegrityError("published artifact is missing") from error
+        if stored_ref != reference:
+            raise ArtifactIntegrityError("published artifact metadata differs")
+        if sha256(payload).hexdigest() != reference.sha256:
+            raise ArtifactIntegrityError("published artifact digest differs")
+        try:
+            artifact = _artifact_from_mapping(
+                cast(dict[str, object], json.loads(payload))
+            )
+            artifact.validate()
+        except (KeyError, TypeError, ValueError) as error:
+            raise ArtifactIntegrityError("published artifact is invalid") from error
+        if _canonical_bytes(artifact) != payload:
+            raise ArtifactIntegrityError("published artifact is not canonical")
+        return artifact
+
+    def exists(self, reference: ArtifactRef) -> bool:
+        return reference.artifact_id in self._artifacts
+
+
+class FilesystemArtifactRepository:
+    def __init__(self, root: Path) -> None:
+        self._root = root.resolve()
+        self._staging = self._root / ".staging"
+        self._published = self._root / "sha256"
+        self._staged: dict[str, Path] = {}
+
+    def stage(self, artifact: RunArtifact) -> ArtifactRef:
+        artifact.validate()
+        payload = _canonical_bytes(artifact)
+        parsed = _artifact_from_mapping(
+            cast(dict[str, object], json.loads(payload))
+        )
+        parsed.validate()
+        if _canonical_bytes(parsed) != payload:
+            raise ArtifactIntegrityError("staged artifact is not canonical")
+        digest = sha256(payload).hexdigest()
+        reference = ArtifactRef(
+            artifact_id=f"sha256/{digest}",
+            sha256=digest,
+            byte_size=len(payload),
+            schema_version=artifact.schema_version,
+        )
+        self._staging.mkdir(parents=True, exist_ok=True)
+        path = self._staging / f"{digest}.{uuid4().hex}.tmp"
+        with path.open("xb") as stream:
+            stream.write(payload)
+            stream.flush()
+            os.fsync(stream.fileno())
+        self._staged[reference.artifact_id] = path
+        return reference
+
+    def publish(self, reference: ArtifactRef) -> ArtifactRef:
+        final = self._path(reference)
+        if final.exists():
+            self._verify_bytes(final.read_bytes(), reference)
+            self._detach_published_aliases(final, reference.sha256)
+            return reference
+        staged = self._staged.get(reference.artifact_id)
+        if staged is None or not staged.exists():
+            candidates = sorted(self._staging.glob(f"{reference.sha256}.*.tmp"))
+            if not candidates:
+                raise ArtifactIntegrityError("staged artifact is missing")
+            staged = candidates[0]
+        payload = staged.read_bytes()
+        self._verify_bytes(payload, reference)
+        final.parent.mkdir(parents=True, exist_ok=True)
+        publishing = final.parent / f".{reference.sha256}.{uuid4().hex}.publishing"
+        with publishing.open("xb") as stream:
+            stream.write(payload)
+            stream.flush()
+            os.fsync(stream.fileno())
+        publishing.chmod(0o444)
+        linked = False
+        try:
+            final.hardlink_to(publishing)
+            linked = True
+        except FileExistsError:
+            self._verify_bytes(final.read_bytes(), reference)
+        directory_fd = os.open(final.parent, os.O_RDONLY)
+        try:
+            os.fsync(directory_fd)
+        finally:
+            os.close(directory_fd)
+        if linked:
+            self._detach_published_aliases(final, reference.sha256)
+            final.chmod(0o644)
+            staged.unlink()
+            self._staged.pop(reference.artifact_id, None)
+            staging_fd = os.open(self._staging, os.O_RDONLY)
+            try:
+                os.fsync(staging_fd)
+            finally:
+                os.close(staging_fd)
+        else:
+            publishing.unlink()
+            directory_fd = os.open(final.parent, os.O_RDONLY)
+            try:
+                os.fsync(directory_fd)
+            finally:
+                os.close(directory_fd)
+        return reference
+
+    def _detach_published_aliases(self, final: Path, digest: str) -> None:
+        candidates = (
+            *self._staging.glob(f"{digest}.*.tmp"),
+            *final.parent.glob(f".{digest}.*.publishing"),
+        )
+        changed_directories: set[Path] = set()
+        for candidate in candidates:
+            try:
+                if candidate.samefile(final):
+                    candidate.unlink()
+                    changed_directories.add(candidate.parent)
+            except FileNotFoundError:
+                continue
+        for directory in changed_directories:
+            directory_fd = os.open(directory, os.O_RDONLY)
+            try:
+                os.fsync(directory_fd)
+            finally:
+                os.close(directory_fd)
+
+    def exists(self, reference: ArtifactRef) -> bool:
+        final = self._path(reference)
+        if not final.exists():
+            return False
+        self._verify_bytes(final.read_bytes(), reference)
+        return True
+
+    def load(self, reference: ArtifactRef) -> RunArtifact:
+        path = self._path(reference)
+        if not path.exists():
+            raise ArtifactIntegrityError("published artifact is missing")
+        payload = path.read_bytes()
+        self._verify_bytes(payload, reference)
+        try:
+            raw = json.loads(payload)
+            artifact = _artifact_from_mapping(cast(dict[str, object], raw))
+            artifact.validate()
+        except (KeyError, TypeError, ValueError) as error:
+            raise ArtifactIntegrityError("published artifact is invalid") from error
+        if _canonical_bytes(artifact) != payload:
+            raise ArtifactIntegrityError("published artifact is not canonical")
+        return artifact
+
+    def reference_for(self, artifact_id: str) -> ArtifactRef:
+        if not artifact_id.startswith("sha256/"):
+            raise ArtifactIntegrityError("artifact identity is not content addressed")
+        digest = artifact_id.removeprefix("sha256/")
+        path = (self._root / artifact_id).resolve()
+        if not path.is_relative_to(self._root):
+            raise ArtifactIntegrityError("artifact path escapes managed root")
+        if not path.exists():
+            raise ArtifactIntegrityError("published artifact is missing")
+        payload = path.read_bytes()
+        reference = ArtifactRef(
+            artifact_id=artifact_id,
+            sha256=digest,
+            byte_size=len(payload),
+            schema_version=1,
+        )
+        self._verify_bytes(payload, reference)
+        return reference
+
+    def recoverable_reference_for(self, artifact_id: str) -> ArtifactRef:
+        if not artifact_id.startswith("sha256/"):
+            raise ArtifactIntegrityError("artifact identity is not content addressed")
+        digest = artifact_id.removeprefix("sha256/")
+        if len(digest) != 64:
+            raise ArtifactIntegrityError("artifact digest is invalid")
+        final = (self._root / artifact_id).resolve()
+        if not final.is_relative_to(self._root):
+            raise ArtifactIntegrityError("artifact path escapes managed root")
+        if final.exists():
+            return self.reference_for(artifact_id)
+        candidates = sorted(self._staging.glob(f"{digest}.*.tmp"))
+        if not candidates:
+            raise ArtifactIntegrityError("staged artifact is missing")
+        payload = candidates[0].read_bytes()
+        reference = ArtifactRef(
+            artifact_id=artifact_id,
+            sha256=digest,
+            byte_size=len(payload),
+            schema_version=1,
+        )
+        self._verify_bytes(payload, reference)
+        return reference
+
+    def _path(self, reference: ArtifactRef) -> Path:
+        if reference.artifact_id != f"sha256/{reference.sha256}":
+            raise ArtifactIntegrityError("artifact identity is not content addressed")
+        path = (self._root / reference.artifact_id).resolve()
+        if not path.is_relative_to(self._root):
+            raise ArtifactIntegrityError("artifact path escapes managed root")
+        return path
+
+    @staticmethod
+    def _verify_bytes(payload: bytes, reference: ArtifactRef) -> None:
+        if len(payload) != reference.byte_size:
+            raise ArtifactIntegrityError("artifact byte size differs")
+        if sha256(payload).hexdigest() != reference.sha256:
+            raise ArtifactIntegrityError("artifact digest differs")
+
+
+def _canonical_bytes(artifact: RunArtifact) -> bytes:
+    return (
+        json.dumps(
+            artifact.to_dict(),
+            ensure_ascii=True,
+            sort_keys=True,
+            separators=(",", ":"),
+            allow_nan=False,
+        )
+        + "\n"
+    ).encode("utf-8")
+
+
+def _artifact_from_mapping(raw: dict[str, object]) -> RunArtifact:
+    return RunArtifact(
+        schema_version=cast(int, raw["schema_version"]),
+        engine_version=cast(str, raw["engine_version"]),
+        run_id=cast(str, raw["run_id"]),
+        condition=cast(str, raw["condition"]),
+        seed=cast(int, raw["seed"]),
+        ticks_requested=cast(int, raw["ticks_requested"]),
+        ticks_completed=cast(int, raw["ticks_completed"]),
+        status=cast(str, raw["status"]),
+        config=cast(dict[str, object], raw["config"]),
+        ancestor_genome=cast(dict[str, object], raw["ancestor_genome"]),
+        summary=cast(dict[str, int], raw["summary"]),
+        metrics=tuple(cast(list[dict[str, int | str]], raw["metrics"])),
+        lineage=tuple(cast(list[dict[str, int]], raw["lineage"])),
+        genome_distribution=tuple(
+            cast(list[dict[str, object]], raw["genome_distribution"])
+        ),
+        notable_events=tuple(
+            cast(list[dict[str, int | str]], raw["notable_events"])
+        ),
+        final_state=cast(dict[str, object], raw["final_state"]),
+        final_checksum=cast(str, raw["final_checksum"]),
+    )