diff --git a/README.md b/README.md index 8ada012..5832395 100644 --- a/README.md +++ b/README.md @@ -128,6 +128,11 @@ dispatcher.dispatch(my_adapter, record.record_id, MyTarget("docs")) dispatcher.invalidate(record.record_id, "source_changed") # removes delivered data ``` +Artifacts made outside a Stele sandbox (for example a document an application +has validated) are recorded with `stele.ledger.external.record_external_artifact`. +They are archived and sealed the same way, marked `backend="external"`, and +never replayed. See [docs/ledger.md](docs/ledger.md#recording-an-artifact-made-outside-a-sandbox). + ### Packaged parsers MinerU, Marker and Docling run in pinned container images with their model diff --git a/docs/ledger.md b/docs/ledger.md index 6f5a659..d4b4035 100644 --- a/docs/ledger.md +++ b/docs/ledger.md @@ -69,7 +69,7 @@ A second record for the same `run_id` raises `DuplicateRunError`. | `source_path`, `source_id` | The descriptive **Source** of that input (a locator) and its id in the archive. Descriptive only, never evidence. | | `parser` | `ParserIdentity`: `name`, `version`, and the executable's digest: `image_digest` (OCI backend) or `module_sha256` (Wasm backends). The digests are measured by the run, not asserted by the caller. | | `parser_config` | The configuration passed to the parser, stored as canonical JSON | -| `backend` | Sandbox backend that ran the parser | +| `backend` | Sandbox backend that ran the parser, or `external` for an artifact recorded outside a sandbox | | `run_conditions` | `RunConditions`: the `device` (`cpu` or `gpu`) and the `memory`, `cpus`, `pids_limit` and `timeout_seconds` the run executed under. Recorded by Stele's runner (`record_parser_run`), so a replay runs the parser the same way (#30). Every stated condition must match the limits the backend reported applying (`SandboxResult.telemetry.limits`); a mismatch, or a condition the backend does not report, raises `ProvenanceError` and nothing is recorded. `None` for runs recorded without them and for records made before schema version 5. | | `artifact_dir` | Host path of the run's output directory (a location, not evidence) | | `artifact_manifest` | `{relative POSIX path: sha256}` for every artifact | @@ -116,6 +116,56 @@ Before creating the record, the transaction refuses: If the bundle on disk no longer matches what the run archived, the record is created and immediately marked `failed`. +## Recording an artifact made outside a sandbox + +Some evidence is produced by an application rather than a sandboxed parser, +for example a document it validated and wants kept (AgentSync's promoted +`SKILL.md` files). `record_external_artifact` is the supported way to ledger +it. Callers should not drive `create_pending` / `seal` themselves. + +```python +from stele.ledger.external import record_external_artifact + +record = record_external_artifact( + ledger, + run_id=run_id, # the caller's UUID + artifact_dir=out_dir, + artifact_paths=[out_dir / "SKILL.md"], + producer=ParserIdentity("agentsync-kanon", "0.1.0"), + producer_config={"validation_level": "promote"}, + input_snapshot=None, # or an archived Snapshot +) +assert record.state is ArtifactState.SEALED +``` + +- **Same integrity.** The bundle is hashed, archived and verified as for a + sandbox run, so `sealed` means the same thing. +- **Asserted, not measured, provenance.** The producer is whatever the caller + says. Such records have `backend = "external"` (`EXTERNAL_BACKEND`), a + name no sandbox backend uses; `ledger_transaction` refuses a run that + claims it. The producer can't carry `image_digest` or `module_sha256`, + since those are measured by a sandbox run (`ProvenanceError`). +- **Never replayed.** There is no parser run to repeat, so replay reports + these records `UNREPLAYABLE` and never invalidates them. +- **Never read as a sandbox parser.** An external record keeps its + producer's name, but the parser-specific consumers refuse it: the + extraction normalizers (`NormalizeError`) and `ChatGPTExportAdapter` + (`MalformedExportError`). A producer named like a sandbox parser can't + pass its bytes off as that parser's output. `SealedBundle.backend` and + `SealedBundle.external` let other adapters make the same check. +- **Input is optional.** An `input_snapshot` must already be in the ledger's + archive (`ProvenanceError` otherwise). +- **`run_id` is the idempotency key.** Calling again for a recorded `run_id` + with the same bundle, producer, config, input Snapshot and Source returns + the sealed record, or seals a record an earlier call left `pending`. A + retry may come after the caller removed some or all of its artifact files: + the archive vouches for those, and every file still present must hash as + recorded. Anything else + for that `run_id`, including a record that is `failed`, raises + `DuplicateRunError`: record it under a new `run_id`. +- If sealing fails, the record is marked `failed` and the error propagates, + as with `ledger_transaction`. + ## Sealing `seal()` stores every artifact in the archive (content that is already stored @@ -212,12 +262,14 @@ if there is any problem. - `stele/ledger/hashing.py` — `sha256_file`, `sha256_manifest`, `build_manifest` - `stele/ledger/store.py` — `LedgerStore` (SQLite, WAL mode) - `stele/ledger/transaction.py` — `ledger_transaction`, `record_run` +- `stele/ledger/external.py` — `record_external_artifact` for artifacts made outside a sandbox - `stele/ledger/migration.py` — migration from schema versions 0, 2, 3, 4, 5, 6 and 7 - `stele/ledger/events.py` — the hash-chained event log, `verify_ledger`, CLI - `stele/ledger/delivery.py` — the delivery log (#13) - `tests/test_ledger.py` — state machine, hashing, per-run records - `tests/test_ledger_provenance.py` — Snapshot provenance, parser identity, sealing, migration - `tests/test_event_log.py` — chaining, verification, tamper detection, anchors, migration, CLI +- `tests/test_external_artifacts.py` — external records: sealing, retries, provenance guards, replay ## Completion criteria diff --git a/docs/replay.md b/docs/replay.md index bb4b297..de9817f 100644 --- a/docs/replay.md +++ b/docs/replay.md @@ -73,7 +73,7 @@ Five named outcomes, never merged into each other: | `REPRODUCED` | Byte-identical output from a deterministic replay. Only parsers whose spec requires a `DETERMINISTIC` backend (Wasm, #7) can produce it. | | `EQUIVALENT` | Accepted under the parser's comparison policy. **Not** proof of reproduction, and never reported as `REPRODUCED`, even when the bytes happen to match. | | `DIVERGED` | The replay produced complete output and it is outside policy. For a parser without a policy, any byte difference. | -| `UNREPLAYABLE` | The parser cannot be run as recorded: no parser identity (a migrated pre-#12 record), no spec in the catalog, a missing or different Wasm module or image, a missing input Snapshot, a record the spec cannot build a run from (e.g. no original file name for a parser that reads by suffix), no capable backend, or a non-deterministic parser with no comparison policy. | +| `UNREPLAYABLE` | The parser cannot be run as recorded: no parser identity (a migrated pre-#12 record), a record made outside a Stele sandbox (`backend = "external"`, see [ledger.md](ledger.md#recording-an-artifact-made-outside-a-sandbox)), no spec in the catalog, a missing or different Wasm module or image, a missing input Snapshot, a record the spec cannot build a run from (e.g. no original file name for a parser that reads by suffix), no capable backend, or a non-deterministic parser with no comparison policy. | | `FAILED` | The parser was run as recorded, but the run did not complete (non-zero exit, timeout, resource limit, crash), so no output was compared. It says nothing about the sealed evidence: replay again. | `DIVERGED` feeds invalidation: `invalidate_diverged(dispatcher, results)` diff --git a/stele/adapters/chatgpt.py b/stele/adapters/chatgpt.py index ea1a1d4..c410903 100644 --- a/stele/adapters/chatgpt.py +++ b/stele/adapters/chatgpt.py @@ -55,6 +55,11 @@ def transform(self, bundle: SealedBundle) -> list[SteleChunk]: def iter_chunks(self, bundle: SealedBundle) -> Iterator[SteleChunk]: """Chunks for every conversation, reading one conversation at a time.""" + if bundle.external: + raise MalformedExportError( + f"record {bundle.record_id} was recorded outside a Stele sandbox; " + f"only {SPLITTER} runs are read as ChatGPT exports" + ) if bundle.parser is not None and bundle.parser.name != SPLITTER: raise MalformedExportError( f"expected a {SPLITTER} bundle, got one from {bundle.parser.name}" diff --git a/stele/contracts/adapter.py b/stele/contracts/adapter.py index 38d1c91..959315f 100644 --- a/stele/contracts/adapter.py +++ b/stele/contracts/adapter.py @@ -28,7 +28,7 @@ from ..archive.records import SnapshotKind, canonical_json from ..archive.store import BlobStore -from ..ledger.models import ArtifactRecord, ParserIdentity +from ..ledger.models import EXTERNAL_BACKEND, ArtifactRecord, ParserIdentity # --------------------------------------------------------------------------- @@ -85,6 +85,14 @@ class SealedBundle: source_kind: SnapshotKind | None source_path: str | None # descriptive locator of the input, not a readable path _archive: BlobStore = field(repr=False, compare=False) + # The sandbox backend that made the record; EXTERNAL_BACKEND for artifacts + # recorded with record_external_artifact, whose producer is asserted. + backend: str | None = None + + @property + def external(self) -> bool: + """True if the bundle was recorded outside a Stele sandbox.""" + return self.backend == EXTERNAL_BACKEND @classmethod def from_record(cls, record: ArtifactRecord, archive: BlobStore) -> "SealedBundle": @@ -102,6 +110,7 @@ def from_record(cls, record: ArtifactRecord, archive: BlobStore) -> "SealedBundl source_kind=record.source_kind, source_path=record.source_path, _archive=archive, + backend=record.backend, ) def paths(self) -> list[str]: diff --git a/stele/extraction/normalizers.py b/stele/extraction/normalizers.py index c8c58af..ea5e816 100644 --- a/stele/extraction/normalizers.py +++ b/stele/extraction/normalizers.py @@ -55,6 +55,13 @@ def normalize(bundle: SealedBundle) -> Extraction: def normalizer_for(bundle: SealedBundle) -> Callable[[SealedBundle], Extraction]: name = bundle.parser.name if bundle.parser is not None else None + if bundle.external: + # A normalizer reads a sandbox parser's output format; an external + # producer's name is asserted, not measured, so it selects nothing. + raise NormalizeError( + f"record {bundle.record_id} was recorded outside a Stele sandbox " + f"(producer {name!r}); parser normalizers apply only to sandbox runs" + ) if name in MARKDOWN_PARSERS: return normalize_markdown if name == SPLITTER: diff --git a/stele/ledger/external.py b/stele/ledger/external.py new file mode 100644 index 0000000..590c400 --- /dev/null +++ b/stele/ledger/external.py @@ -0,0 +1,209 @@ +""" +Stele — recording artifacts produced outside a Stele sandbox. + +ledger_transaction / record_run record a sandboxed parser run: Stele ran the +parser, measured its identity, archived its input and captured its output. +Some evidence is produced elsewhere, by an application that wants the result +kept with the same guarantees (AgentSync's promoted SKILL.md files are the +first such caller). record_external_artifact is the supported way to ledger +it; callers should not drive create_pending / seal themselves. + +Usage:: + + archive = BlobStore(archive_root) + ledger = LedgerStore(db_path, archive) + + record = record_external_artifact( + ledger, + run_id=run_id, # the caller's UUID + artifact_dir=out_dir, + artifact_paths=[out_dir / "SKILL.md"], + producer=ParserIdentity("agentsync-kanon", "0.1.0"), + producer_config={"validation_level": "promote"}, + ) + assert record.state is ArtifactState.SEALED + +What the record claims, and what it does not: + + - The bundle is hashed, archived and verified exactly as for a sandbox run, + so SEALED means the same thing: the archive holds these bytes. + - The producer is asserted by the caller, not measured. The record's + backend is EXTERNAL_BACKEND ("external"), which no sandbox backend uses, + and replay reports such records UNREPLAYABLE. A producer cannot carry an + image_digest or module_sha256: those are digests a sandbox run measures. + - An input is optional. If given, its Snapshot must already be in the + ledger's archive, as for a sandbox run. + +run_id is the caller's idempotency key. Calling again with a run_id the +ledger already holds, for the same bundle, producer, config, input and +source, returns the SEALED record (or seals a record an earlier call left +PENDING). A retry may come after the caller removed some or all of its +artifact files: those are vouched for by the archive, which sealing already +verified, and every file still present must hash as recorded. Anything else for that run_id +raises DuplicateRunError: a run has one record. + +External records carry their producer's name like any record, but the +parser-specific consumers (extraction normalizers, the ChatGPT adapter) +refuse them: bytes an application wrote are never read as a sandbox +parser's output, even under the same name. +""" +from __future__ import annotations + +import uuid +from pathlib import Path +from typing import Any, Mapping + +from ..archive.records import Snapshot, Source +from ..archive.store import ArchiveError +from .hashing import artifact_relative_path, build_manifest +from .models import EXTERNAL_BACKEND, ArtifactRecord, ArtifactState, ParserIdentity +from .store import DuplicateRunError, LedgerStore, ProvenanceError, canonical_parser_config +from .transaction import _fail_quietly, _state_or_none + + +def record_external_artifact( + store: LedgerStore, + *, + run_id: str | uuid.UUID, + artifact_dir: Path, + artifact_paths: list[Path], + producer: ParserIdentity, + producer_config: Mapping[str, Any], + input_snapshot: Snapshot | None = None, + source: Source | None = None, +) -> ArtifactRecord: + """Record a bundle produced outside a Stele sandbox and seal it. + + Returns the SEALED record. Raises ValueError for a run_id that is not a + UUID, ProvenanceError for a producer carrying measured digests (or an + input the archive does not hold), DuplicateRunError when run_id already + has a different record, and the sealing error (after marking the record + FAILED) when the bundle cannot be archived and verified. + """ + run_id = _run_id(run_id) + if not isinstance(producer, ParserIdentity): + raise TypeError("producer must be a ParserIdentity") + if producer.image_digest is not None or producer.module_sha256 is not None: + raise ProvenanceError( + "an external producer cannot assert image_digest or module_sha256; " + "those are measured by a sandbox run" + ) + + existing = store.get_by_run_id(run_id) + if existing is None: + try: + record = store.create_pending( + run_id=run_id, + artifact_dir=artifact_dir, + artifact_paths=artifact_paths, + parser=producer, + parser_config=producer_config, + input_snapshot=input_snapshot, + source=source, + backend=EXTERNAL_BACKEND, + ) + except DuplicateRunError: + # Another caller recorded this run_id first; treat it as a retry. + existing = store.get_by_run_id(run_id) + if existing is None: + raise + else: + return _seal(store, record) + + _require_same_run( + store, existing, artifact_dir, artifact_paths, producer, producer_config, + input_snapshot, source, + ) + if existing.state is ArtifactState.SEALED: + return existing + if existing.state is ArtifactState.PENDING: + return _seal(store, existing) + raise DuplicateRunError( + f"run {run_id} already has record {existing.record_id} " + f"(state={existing.state.value}); record it under a new run_id" + ) + + +def _run_id(run_id: str | uuid.UUID) -> str: + try: + return str(run_id if isinstance(run_id, uuid.UUID) else uuid.UUID(run_id)) + except (ValueError, TypeError, AttributeError): + raise ValueError(f"run_id must be a UUID; got {run_id!r}") from None + + +def _require_same_run( + store: LedgerStore, + record: ArtifactRecord, + artifact_dir: Path, + artifact_paths: list[Path], + producer: ParserIdentity, + producer_config: Mapping[str, Any], + input_snapshot: Snapshot | None, + source: Source | None, +) -> None: + """Raise DuplicateRunError unless this call describes the recorded run. + + Everything that does not need the caller's files is compared first, so a + run_id that belongs to another record is reported as such even when the + files are gone. + """ + differences = [] + if record.backend != EXTERNAL_BACKEND: + differences.append(f"it was recorded by backend {record.backend!r}") + if record.parser != producer: + differences.append("the producer differs") + if canonical_parser_config(record.parser_config or {}) != canonical_parser_config( + producer_config + ): + differences.append("the producer config differs") + if not _same_input(store, record, input_snapshot): + differences.append("the input differs") + if record.source_id != (source.source_id if source is not None else None): + differences.append("the source differs") + if not differences: + relative = sorted( + artifact_relative_path(artifact_dir, p).as_posix() for p in artifact_paths + ) + if relative != sorted(record.artifact_manifest): + differences.append("the artifacts differ") + else: + # Files the caller removed after an earlier call are vouched for + # by the archive (sealing verifies it); every file still present + # is compared, so a changed file is never taken for the recorded one. + present = [p for p in artifact_paths if Path(p).exists()] + manifest = build_manifest(artifact_dir, present) + if any(record.artifact_manifest.get(rel) != digest + for rel, digest in manifest.items()): + differences.append("the artifacts differ") + if differences: + raise DuplicateRunError( + f"run {record.run_id} already has record {record.record_id}, and " + + "; ".join(differences) + ) + + +def _same_input(store: LedgerStore, record: ArtifactRecord, snapshot: Snapshot | None) -> bool: + """True if snapshot is the record's input: same digest and kind, and the + same Snapshot the archive holds.""" + if snapshot is None: + return record.source_hash is None + if (record.source_hash, record.source_kind) != (snapshot.digest, snapshot.kind): + return False + try: + return store.archive.get_snapshot(snapshot.digest, snapshot.kind) == snapshot + except ArchiveError: + return False + + +def _seal(store: LedgerStore, record: ArtifactRecord) -> ArtifactRecord: + """Seal record, marking it FAILED (and re-raising) if sealing fails.""" + try: + return store.seal(record.record_id) + except BaseException as exc: + if isinstance(exc, Exception) and _state_or_none(store, record.record_id) is ( + ArtifactState.SEALED + ): + # The SEALED write is durable; only the read-back after it failed. + return store.get(record.record_id) + _fail_quietly(store, record.record_id, exc) + raise diff --git a/stele/ledger/models.py b/stele/ledger/models.py index ce48768..3bf2956 100644 --- a/stele/ledger/models.py +++ b/stele/ledger/models.py @@ -22,6 +22,11 @@ class ArtifactState(str, Enum): INVALIDATED = "invalidated" # retroactively withdrawn (see replay/); terminal +# The backend recorded for artifacts produced outside a Stele sandbox +# (stele.ledger.external). No sandbox backend uses this name. +EXTERNAL_BACKEND = "external" + + @dataclass(frozen=True) class ParserIdentity: """Which parser produced a run's artifacts. diff --git a/stele/ledger/transaction.py b/stele/ledger/transaction.py index e27cf4a..0459731 100644 --- a/stele/ledger/transaction.py +++ b/stele/ledger/transaction.py @@ -38,7 +38,9 @@ from ..archive.records import Source from ..containment.result import SandboxResult -from .models import ArtifactRecord, ArtifactState, ParserIdentity, RunConditions +from .models import ( + EXTERNAL_BACKEND, ArtifactRecord, ArtifactState, ParserIdentity, RunConditions, +) from .store import LedgerStore, ProvenanceError @@ -187,6 +189,11 @@ def ledger_transaction( "nothing to ledger" ) + if sandbox_result.backend == EXTERNAL_BACKEND: + raise ProvenanceError( + f"backend {EXTERNAL_BACKEND!r} is reserved for artifacts recorded with " + "stele.ledger.external.record_external_artifact" + ) _check_input(sandbox_result) _check_run_conditions(sandbox_result, run_conditions) identity = measured_parser_identity(parser, sandbox_result) diff --git a/stele/replay/engine.py b/stele/replay/engine.py index 36e88b6..f07be32 100644 --- a/stele/replay/engine.py +++ b/stele/replay/engine.py @@ -18,7 +18,7 @@ DIVERGED outside policy (or, without a policy, any byte difference). Feeds invalidation: invalidate_diverged(). UNREPLAYABLE the parser cannot be run as recorded: no parser identity or - spec, a missing or different Wasm module or image, a missing + spec, a record made outside a Stele sandbox, a missing or different Wasm module or image, a missing input Snapshot, no capable backend, or a non-deterministic parser without a comparison policy. FAILED the parser was run as recorded but the run did not complete @@ -48,7 +48,7 @@ from ..archive.records import canonical_json from ..containment.backend import UnsupportedBackendError from ..containment.wasm import wasm_module_sha256 -from ..ledger.models import ArtifactRecord, ArtifactState +from ..ledger.models import EXTERNAL_BACKEND, ArtifactRecord, ArtifactState from ..ledger.events import append_event, write_transaction from ..ledger.store import LedgerStore, connect from .models import InvalidationReason @@ -222,6 +222,11 @@ def _spec(self) -> ParserSpec: parser = self.record.parser if parser is None: raise _Unreplayable("the record has no parser identity (migrated from a pre-#12 ledger)") + if self.record.backend == EXTERNAL_BACKEND: + raise _Unreplayable( + f"recorded outside a Stele sandbox (producer {parser.name} {parser.version}); " + "there is no parser run to repeat" + ) spec = self.catalog.get(parser.name, parser.version) if spec is None: raise _Unreplayable(f"parser {parser.name} {parser.version} is not in the catalog") diff --git a/tests/test_external_artifacts.py b/tests/test_external_artifacts.py new file mode 100644 index 0000000..6820cda --- /dev/null +++ b/tests/test_external_artifacts.py @@ -0,0 +1,357 @@ +""" +Recording artifacts produced outside a Stele sandbox (stele.ledger.external). + + - a bundle is recorded, archived and sealed; the record says "external" + - the same call again is idempotent; a PENDING record left behind is sealed + - anything else for a recorded run_id is refused + - a producer cannot assert measured digests; sandbox runs cannot claim + the external backend + - an input Snapshot must already be archived + - replay reports external records UNREPLAYABLE, and the event log verifies + +Run: + uv run pytest tests/test_external_artifacts.py -v +""" +from __future__ import annotations + +import hashlib +import uuid +from pathlib import Path + +import pytest + +from stele.adapters import ChatGPTExportAdapter, MalformedExportError +from stele.archive.records import Snapshot, SnapshotKind, Source +from stele.contracts.adapter import SealedBundle +from stele.extraction.normalizers import NormalizeError, normalize +from stele.containment.result import SandboxResult +from stele.ledger.events import verify_ledger +from stele.ledger.external import record_external_artifact +from stele.ledger.hashing import sha256_manifest +from stele.ledger.models import EXTERNAL_BACKEND, ArtifactState, ParserIdentity +from stele.ledger.store import ( + ArtifactDriftError, + DuplicateRunError, + LedgerStore, + ProvenanceError, +) +from stele.ledger.transaction import ledger_transaction +from stele.replay.engine import ReplayOutcome, replay_record +from stele.replay.parsers import ParserCatalog +from tests.ledger_helpers import PROVENANCE, open_ledger + +PRODUCER = ParserIdentity(name="agentsync-kanon", version="0.1.0") +CONFIG = {"validation_level": "promote"} + + +@pytest.fixture +def store(tmp_path: Path) -> LedgerStore: + return open_ledger(tmp_path / "ledger.db") + + +@pytest.fixture +def bundle(tmp_path: Path) -> tuple[Path, list[Path]]: + out = tmp_path / "out" + out.mkdir() + skill = out / "SKILL.md" + skill.write_text("# a skill\n", encoding="utf-8") + return out, [skill] + + +def _snapshot(data: bytes) -> Snapshot: + return Snapshot(SnapshotKind.FILE, hashlib.sha256(data).hexdigest(), len(data), 1) + + +def _record(store, bundle, run_id=None, **overrides): + artifact_dir, paths = bundle + kwargs = dict( + run_id=run_id or str(uuid.uuid4()), + artifact_dir=artifact_dir, + artifact_paths=paths, + producer=PRODUCER, + producer_config=CONFIG, + ) + kwargs.update(overrides) + return record_external_artifact(store, **kwargs) + + +# --------------------------------------------------------------------------- +# Recording +# --------------------------------------------------------------------------- + +def test_records_and_seals_bundle(store, bundle) -> None: + run_id = str(uuid.uuid4()) + record = _record(store, bundle, run_id) + + assert record.state is ArtifactState.SEALED + assert record.run_id == run_id + assert record.backend == EXTERNAL_BACKEND + assert record.parser == PRODUCER + assert record.parser_config == CONFIG + assert record.source_hash is None + assert record.artifact_manifest == {"SKILL.md": record.artifact_manifest["SKILL.md"]} + assert record.artifact_hash == sha256_manifest(record.artifact_manifest) + assert store.get_by_run_id(run_id) == record + + +def test_sealed_bundle_survives_removal_from_disk(store, bundle) -> None: + """Sealing archived the bytes: the staging directory is not the evidence.""" + record = _record(store, bundle) + for path in bundle[1]: + path.unlink() + assert store.archive.read_tree(record.artifact_hash) == record.artifact_manifest + + +def test_accepts_uuid_objects(store, bundle) -> None: + run_id = uuid.uuid4() + assert _record(store, bundle, run_id).run_id == str(run_id) + + +@pytest.mark.parametrize("bad", ["", "not-a-uuid", None, 42]) +def test_run_id_must_be_uuid(store, bundle, bad) -> None: + with pytest.raises(ValueError, match="UUID"): + record_external_artifact( + store, run_id=bad, artifact_dir=bundle[0], artifact_paths=bundle[1], + producer=PRODUCER, producer_config=CONFIG, + ) + + +def test_records_input_snapshot(store, bundle) -> None: + store.archive.put_bytes(b"input document") + snapshot = store.archive.put_snapshot(_snapshot(b"input document")) + + record = _record(store, bundle, input_snapshot=snapshot) + + assert record.source_hash == snapshot.digest + + +def test_unarchived_input_is_refused(store, bundle) -> None: + with pytest.raises(ProvenanceError, match="not in the ledger's archive"): + _record(store, bundle, input_snapshot=_snapshot(b"never archived")) + assert store.list_by_states(list(ArtifactState)) == [] + + +# --------------------------------------------------------------------------- +# Provenance guards +# --------------------------------------------------------------------------- + +@pytest.mark.parametrize("field", ["image_digest", "module_sha256"]) +def test_producer_cannot_assert_measured_digests(store, bundle, field) -> None: + producer = ParserIdentity(name="x", version="1", **{field: "a" * 64}) + with pytest.raises(ProvenanceError, match="measured"): + _record(store, bundle, producer=producer) + + +def test_sandbox_run_cannot_claim_external_backend(store, bundle) -> None: + artifact_dir, paths = bundle + result = SandboxResult( + run_id=uuid.uuid4(), exit_code=0, stdout=b"", stderr=b"", wall_time_seconds=0.1, + timed_out=False, artifact_dir=artifact_dir, artifact_paths=paths, + backend=EXTERNAL_BACKEND, + ) + with pytest.raises(ProvenanceError, match="reserved"): + with ledger_transaction(store, result, **PROVENANCE): + pass + assert store.get_by_run_id(str(result.run_id)) is None + + +# --------------------------------------------------------------------------- +# Retries: run_id is the idempotency key +# --------------------------------------------------------------------------- + +def test_same_call_again_returns_sealed_record(store, bundle) -> None: + run_id = str(uuid.uuid4()) + first = _record(store, bundle, run_id) + again = _record(store, bundle, run_id) + + assert again == first + assert len(store.list_by_states(list(ArtifactState))) == 1 + + +def test_pending_record_is_sealed_on_retry(store, bundle) -> None: + run_id = str(uuid.uuid4()) + pending = store.create_pending( + run_id=run_id, artifact_dir=bundle[0], artifact_paths=bundle[1], + parser=PRODUCER, parser_config=CONFIG, backend=EXTERNAL_BACKEND, + ) + + record = _record(store, bundle, run_id) + + assert record.record_id == pending.record_id + assert record.state is ArtifactState.SEALED + + +@pytest.mark.parametrize("change", ["content", "producer", "config"]) +def test_different_call_for_recorded_run_is_refused(store, bundle, change) -> None: + run_id = str(uuid.uuid4()) + first = _record(store, bundle, run_id) + overrides = {} + if change == "content": + bundle[1][0].write_text("# edited\n", encoding="utf-8") + elif change == "producer": + overrides["producer"] = ParserIdentity(name="agentsync-kanon", version="0.2.0") + else: + overrides["producer_config"] = {"validation_level": "draft"} + + with pytest.raises(DuplicateRunError, match="already has record"): + _record(store, bundle, run_id, **overrides) + assert store.get(first.record_id) == first + + +def test_sandbox_record_is_not_reused(store, bundle) -> None: + artifact_dir, paths = bundle + result = SandboxResult( + run_id=uuid.uuid4(), exit_code=0, stdout=b"", stderr=b"", wall_time_seconds=0.1, + timed_out=False, artifact_dir=artifact_dir, artifact_paths=paths, backend="bubblewrap", + ) + with ledger_transaction(store, result, parser=PRODUCER, parser_config=CONFIG): + pass + + with pytest.raises(DuplicateRunError, match="backend 'bubblewrap'"): + _record(store, bundle, str(result.run_id)) + + +def test_retry_after_files_were_removed_returns_sealed_record(store, bundle) -> None: + """The first call sealed the record but its answer was lost; the caller + cleaned up its files and retries with the same run_id.""" + run_id = str(uuid.uuid4()) + first = _record(store, bundle, run_id) + for path in bundle[1]: + path.unlink() + + assert _record(store, bundle, run_id) == first + + +def test_retry_after_files_were_removed_still_compares_paths(store, bundle) -> None: + run_id = str(uuid.uuid4()) + _record(store, bundle, run_id) + for path in bundle[1]: + path.unlink() + + with pytest.raises(DuplicateRunError, match="the artifacts differ"): + _record(store, bundle, run_id, artifact_paths=[bundle[0] / "OTHER.md"]) + + +def test_retry_with_some_files_removed_still_compares_the_rest(store, tmp_path) -> None: + """One removed file must not excuse another that changed.""" + out = tmp_path / "two" + out.mkdir() + a, b = out / "a.md", out / "b.md" + a.write_text("original A", encoding="utf-8") + b.write_text("original B", encoding="utf-8") + run_id = str(uuid.uuid4()) + first = _record(store, (out, [a, b]), run_id) + a.unlink() + + assert _record(store, (out, [a, b]), run_id) == first # b unchanged: a retry + b.write_text("different B", encoding="utf-8") + with pytest.raises(DuplicateRunError, match="the artifacts differ"): + _record(store, (out, [a, b]), run_id) + + +def test_sandbox_record_is_not_reused_after_files_were_removed(store, bundle) -> None: + artifact_dir, paths = bundle + result = SandboxResult( + run_id=uuid.uuid4(), exit_code=0, stdout=b"", stderr=b"", wall_time_seconds=0.1, + timed_out=False, artifact_dir=artifact_dir, artifact_paths=paths, backend="bubblewrap", + ) + with ledger_transaction(store, result, parser=PRODUCER, parser_config=CONFIG): + pass + for path in paths: + path.unlink() + + with pytest.raises(DuplicateRunError, match="backend 'bubblewrap'"): + _record(store, bundle, str(result.run_id)) + + +def _archived_input(store, data: bytes, kind=SnapshotKind.FILE) -> Snapshot: + store.archive.put_bytes(data) + return store.archive.put_snapshot( + Snapshot(kind, hashlib.sha256(data).hexdigest(), len(data), 1) + ) + + +def test_retry_with_a_different_source_is_refused(store, bundle) -> None: + run_id = str(uuid.uuid4()) + snapshot = _archived_input(store, b"input document") + _record(store, bundle, run_id, input_snapshot=snapshot, source=Source(locator="a.md")) + + with pytest.raises(DuplicateRunError, match="the source differs"): + _record(store, bundle, run_id, input_snapshot=snapshot, source=Source(locator="b.md")) + with pytest.raises(DuplicateRunError, match="the source differs"): + _record(store, bundle, run_id, input_snapshot=snapshot) + assert _record( + store, bundle, run_id, input_snapshot=snapshot, source=Source(locator="a.md") + ).state is ArtifactState.SEALED + + +def test_retry_with_a_different_snapshot_is_refused(store, bundle) -> None: + run_id = str(uuid.uuid4()) + snapshot = _archived_input(store, b"input document") + _record(store, bundle, run_id, input_snapshot=snapshot) + + same_digest_other_size = Snapshot(snapshot.kind, snapshot.digest, snapshot.size + 1, 1) + with pytest.raises(DuplicateRunError, match="the input differs"): + _record(store, bundle, run_id, input_snapshot=same_digest_other_size) + + +# --------------------------------------------------------------------------- +# An external producer is never read as a sandbox parser +# --------------------------------------------------------------------------- + +def test_external_record_under_a_parser_name_is_not_read_as_that_parser(store, bundle) -> None: + impostor = ParserIdentity(name="chatgpt-export-split", version="1") + record = _record(store, bundle, producer=impostor, producer_config={}) + sealed = SealedBundle.from_record(record, store.archive) + assert sealed.external + + with pytest.raises(NormalizeError, match="outside a Stele sandbox"): + normalize(sealed) + with pytest.raises(MalformedExportError, match="outside a Stele sandbox"): + ChatGPTExportAdapter().transform(sealed) + + +def test_failed_record_is_not_resumed(store, bundle) -> None: + run_id = str(uuid.uuid4()) + pending = store.create_pending( + run_id=run_id, artifact_dir=bundle[0], artifact_paths=bundle[1], + parser=PRODUCER, parser_config=CONFIG, backend=EXTERNAL_BACKEND, + ) + store.fail(pending.record_id, error="gave up") + + with pytest.raises(DuplicateRunError, match="new run_id"): + _record(store, bundle, run_id) + + +def test_drift_before_seal_marks_record_failed(store, bundle, monkeypatch) -> None: + run_id = str(uuid.uuid4()) + real_seal = store.seal + + def drifting_seal(record_id): + bundle[1][0].write_text("# changed underneath\n", encoding="utf-8") + return real_seal(record_id) + + monkeypatch.setattr(store, "seal", drifting_seal) + with pytest.raises(ArtifactDriftError): + _record(store, bundle, run_id) + assert store.get_by_run_id(run_id).state is ArtifactState.FAILED + + +# --------------------------------------------------------------------------- +# Replay and the event log +# --------------------------------------------------------------------------- + +def test_replay_reports_external_record_unreplayable(store, bundle) -> None: + record = _record(store, bundle) + + result = replay_record(store, ParserCatalog(), record) + + assert result.outcome is ReplayOutcome.UNREPLAYABLE + assert "outside a Stele sandbox" in result.reason + assert store.get(record.record_id).state is ArtifactState.SEALED + + +def test_event_log_verifies_external_records(store, bundle) -> None: + _record(store, bundle) + report = verify_ledger(store) + assert report.ok, report.problems