diff --git a/docs/adapter.md b/docs/adapter.md index e91ffdd..f841663 100644 --- a/docs/adapter.md +++ b/docs/adapter.md @@ -178,15 +178,21 @@ Delivery is never a ledger state. conversation, each holding the exact bytes of its element of `conversations.json`. -- **Streamed.** `iter_chunks()` reads and verifies one conversation at a time, - so memory is bounded by the largest conversation, not the export. - `transform()` is `list(iter_chunks())`. +- **Streamed.** `iter_chunks()` reads, verifies and parses one conversation + at a time, so it holds one parsed conversation at once. `transform()`, which + the Dispatcher calls, is `list(iter_chunks())`: it returns every chunk, so + it holds the text of the whole export. - **Every branch.** A conversation is a tree. Editing a message or regenerating a response keeps the old version as a sibling, and `current_node` only marks the leaf the UI shows. The adapter walks the whole `mapping` iteratively, so any depth works. Every node whose message has - content becomes one chunk, `:`, stable across - re-exports. + content becomes one chunk, `:`, stable across + re-exports. Both parts are percent-encoded, so neither contains `:` or + `#`. The `#` given to the n-th copy of a conversation repeated in one + export can't collide with a real conversation id, and two different + (conversation, node) pairs never share a chunk id. ChatGPT's UUID ids are + unchanged by the encoding. A message whose `author` isn't an object is + malformed. - **Rebuildable.** Chunk metadata rebuilds each branch: - `parent_node_id`, the nearest ancestor that has a chunk, so empty system nodes don't break the chain; diff --git a/docs/ledger.md b/docs/ledger.md index 6f5a659..5e6069d 100644 --- a/docs/ledger.md +++ b/docs/ledger.md @@ -169,11 +169,11 @@ the two can't diverge: | Event | When | |---|---| -| `record.created` | `create_pending`: run, artifact hash, input Snapshot, Source, parser identity and config, backend | +| `record.created` | `create_pending`: run, artifact hash, input Snapshot, Source, parser identity and config, run conditions, backend, creation time | | `record.sealed` / `record.failed` / `record.invalidated` | each state transition, with its error or reason | | `delivery.opened` / `delivery.event` | each delivery and each intent, receipt, failure or removal in the delivery log | | `replay.logged` | each replay and its outcome | -| `*.imported` | the state found when a ledger was migrated to version 5 | +| `*.imported` | the state found when a ledger was migrated to the event log (version 7) | **The chain.** - Each event's hash is SHA-256 over the canonical JSON of its sequence number, @@ -188,9 +188,16 @@ the two can't diverge: 1. It verifies the chain: every hash, every link, no gaps in the sequence. 2. It replays the chain to rebuild what the ledger's tables must contain (each record's state and identity, each delivery and its events, each - replay) and reports every difference. This catches edits to + replay) and reports every difference. Every field an event commits to is + compared, including a record's input Snapshot kind, Source, backend, + creation time and its failure or invalidation reason, and a replay's + reason, policy and platform. A record's `source_path` is checked against + the archived Source its `source_id` names. This catches edits to `artifact_records`, the one table whose rows legitimately change, such as - an `invalidated` record flipped back to `sealed` by hand. + an `invalidated` record flipped back to `sealed` by hand. `artifact_dir` + is only where the run's output was on disk, not evidence, and isn't + checked. Records created before the chain carried `created_at` have their + creation time on the row alone. A tampered event body is reported, never a crash. @@ -206,6 +213,17 @@ python -m stele.ledger.events ledger.db archive/ [--anchor SEQ:HASH] This prints `ok`, the chain length, the head and every problem, and exits 1 if there is any problem. +**Verifiers never migrate.** This command and the other read-only tools +(`python -m stele.identity`, `python -m stele.extraction`, and the `report`, +`sign-off`, `check` and `list` commands of `python -m stele.verification`) +open the ledger with `LedgerStore(..., migrate=False)`. A ledger older than +this Stele is refused with `LedgerSchemaError` and left exactly as it was, +instead of being migrated. Migrating a ledger that predates the event log +would chain its current tables as `*.imported` events, and the verifier would +then report tables it had just vouched for as verified. Open such a ledger +the normal way (for example by recording a run) to migrate it first; there +is no chain to verify what it held before then. + ## Implementation - `stele/ledger/models.py` — `ArtifactRecord`, `ArtifactState`, `ParserIdentity` diff --git a/stele/adapters/chatgpt.py b/stele/adapters/chatgpt.py index ea1a1d4..e95245f 100644 --- a/stele/adapters/chatgpt.py +++ b/stele/adapters/chatgpt.py @@ -5,10 +5,11 @@ chatgpt-export-split, stele/extractors): index.jsonl plus one conversation-NNNNNN.json per conversation, each the exact bytes of one element of the export's conversations.json. The adapter reads them through the -SealedBundle, so every byte it sees is re-verified against the sealed digest, -and it never holds more than one conversation in memory: iter_chunks() is a -generator that parses a conversation only when the previous one's chunks are -consumed. +SealedBundle, so every byte it sees is re-verified against the sealed digest. +iter_chunks() is a generator that parses a conversation only when the +previous one's chunks are consumed, so it holds one parsed conversation at a +time. transform(), which the Dispatcher calls, returns the full chunk list +(the adapter contract), so it holds the text of every conversation at once. A ChatGPT conversation is a tree, not a list. Its `mapping` holds every node the user ever produced: when a message is edited or a response regenerated, @@ -24,8 +25,13 @@ - per-conversation counts of nodes, leaves (= branches) and chunks, and the byte range of the conversation in the original export. -Chunk ids are ":", stable across re-exports, so a -re-delivered conversation updates rather than duplicates. +Chunk ids are ":", stable across re-exports, so a +re-delivered conversation updates rather than duplicates. Both parts are +percent-encoded (every character but unreserved ones), so neither contains +":" or "#": the "#" given to the n-th copy of a conversation repeated in +one export can never collide with a real conversation id, and the ":" +between the parts is unambiguous. ChatGPT's ids are UUIDs, which encoding +leaves unchanged. Traversal is iterative and order is deterministic (roots and siblings in export order), so the same bundle always yields the same chunks, as #13 @@ -36,6 +42,7 @@ import json from typing import Any, Iterator +from urllib.parse import quote from ..contracts.adapter import SealedBundle, SteleChunk, make_chunk @@ -70,11 +77,12 @@ def iter_chunks(self, bundle: SealedBundle) -> Iterator[SteleChunk]: raise MalformedExportError(f"{entry['path']} is not JSON: {exc}") from exc if not isinstance(conversation, dict): raise MalformedExportError(f"{entry['path']} is not a JSON object") - conv_id = str( + conv_id = quote(str( conversation.get("conversation_id") or conversation.get("id") or f"conversation-{entry['index']:06d}" - ) + ), safe="") # The same conversation twice in one export keeps both copies apart. + # An encoded id never contains "#", so the suffix cannot collide. seen_ids[conv_id] = seen_ids.get(conv_id, 0) + 1 if seen_ids[conv_id] > 1: conv_id = f"{conv_id}#{seen_ids[conv_id]}" @@ -158,9 +166,15 @@ def children_of(node_id: str) -> list[str]: emitted_as = chunk_parent if text: - author = message.get("author") or {} + author = message.get("author") + if author is None: + author = {} + elif not isinstance(author, dict): + raise MalformedExportError( + f"{conv_id}: node {node_id!r} has an author that is not an object" + ) yield make_chunk( - f"{conv_id}:{node_id}", + f"{conv_id}:{quote(node_id, safe='')}", text, **base, node_id=node_id, diff --git a/stele/extraction/__main__.py b/stele/extraction/__main__.py index 3a6842c..21a8e3d 100644 --- a/stele/extraction/__main__.py +++ b/stele/extraction/__main__.py @@ -15,7 +15,7 @@ from ..archive import BlobStore from ..contracts.adapter import SealedBundle from ..ledger.models import ArtifactState -from ..ledger.store import LedgerStore +from ..ledger.store import LedgerSchemaError, LedgerStore from .contract import Extraction, ExtractionFormatError from .normalizers import normalize from .resolver import ResolutionError, Resolver @@ -37,7 +37,12 @@ def main(argv: list[str] | None = None) -> int: ve.add_argument("--allow-invalidated", action="store_true") args = ap.parse_args(argv) - ledger = LedgerStore(args.ledger, BlobStore(args.archive)) + try: + # Read-only: a verifier never migrates the ledger it reads. + ledger = LedgerStore(args.ledger, BlobStore(args.archive), migrate=False) + except LedgerSchemaError as exc: + print(f"cannot open the ledger: {exc}", file=sys.stderr) + return 1 try: if args.command == "extract": record = ledger.get(args.record_id) diff --git a/stele/identity/__main__.py b/stele/identity/__main__.py index 19408eb..0dd6917 100644 --- a/stele/identity/__main__.py +++ b/stele/identity/__main__.py @@ -17,7 +17,7 @@ from ..archive import BlobStore, Snapshot, Source from ..ledger.events import EventLog from ..ledger.models import ArtifactRecord -from ..ledger.store import LedgerStore, RecordNotFoundError +from ..ledger.store import LedgerSchemaError, LedgerStore, RecordNotFoundError from .refs import Observation, event_ref, observation_of, record_ref from .resolver import IdentityResolver, ResolveError @@ -53,7 +53,12 @@ def main(argv: list[str] | None = None) -> int: rf.add_argument("record_id") args = ap.parse_args(argv) - ledger = LedgerStore(args.ledger, BlobStore(args.archive)) + try: + # Read-only: a verifier never migrates the ledger it reads. + ledger = LedgerStore(args.ledger, BlobStore(args.archive), migrate=False) + except LedgerSchemaError as exc: + print(f"cannot open the ledger: {exc}", file=sys.stderr) + return 1 try: if args.command == "refs": try: diff --git a/stele/identity/resolver.py b/stele/identity/resolver.py index b5c984f..3da7340 100644 --- a/stele/identity/resolver.py +++ b/stele/identity/resolver.py @@ -117,6 +117,7 @@ def _resolve_ObservationRef(self, ref: ObservationRef) -> Any: observation = observation_of(record) except ValueError as exc: raise ResolveError(f"{ref}: {exc}") from exc + self._check_observed_facts(ref, record) if observation.digest != ref.digest: raise ResolveError( f"{ref}: the record's observation has digest {observation.digest}, not {ref.digest}" @@ -126,6 +127,41 @@ def _resolve_ObservationRef(self, ref: ObservationRef) -> Any: self._resolve_SourceRef(SourceRef(observation.source_id)) return observation + def _check_observed_facts(self, ref: ObservationRef, record: ArtifactRecord) -> None: + """The record's row must say what its creation event in the chain says. + + An observation is rebuilt from the row, so it must not resolve from a + row edited after the fact. source_hash, source_kind and source_id are + chained with every record; created_at is chained for records created + since the event log first committed to it (older records' creation + time rests on the row alone). + """ + log = EventLog(self.ledger) + try: + created = next( + (e for e in log.for_subject(record.record_id) + if e.kind in ("record.created", "record.imported")), + None, + ) + finally: + log.close() + if created is None: + raise ResolveError(f"{ref}: record {record.record_id} has no creation event in the chain") + body = created.body + facts = { + "source_hash": record.source_hash, + "source_kind": record.source_kind.value if record.source_kind else None, + "source_id": record.source_id, + } + if "created_at" in body: + facts["created_at"] = record.created_at.isoformat() + for key, value in facts.items(): + if body.get(key) != value: + raise ResolveError( + f"{ref}: the record's {key} is {value!r}, but its creation event " + f"(event {created.seq}) says {body.get(key)!r}" + ) + def _resolve_AnchorRef(self, ref: AnchorRef) -> str: record = self._resolve_RecordRef(ref.record) try: diff --git a/stele/ledger/events.py b/stele/ledger/events.py index cbae4f8..c436a04 100644 --- a/stele/ledger/events.py +++ b/stele/ledger/events.py @@ -225,8 +225,12 @@ def verify_ledger(ledger: LedgerStore, *, anchor: tuple[int, str] | None = None) Replays the chain to derive every record's state and identity, every delivery and its events, and every replay, and reports each row that is - missing from the chain, missing from the tables, or different. + missing from the chain, missing from the tables, or different. Every + field an event commits to is compared; a record's source_path is checked + against the archived Source its source_id names. artifact_dir is only + where the run's output was on disk, not evidence, and is not checked. """ + from ..archive.store import ArchiveError from .hashing import sha256_manifest log = EventLog(ledger) @@ -273,8 +277,10 @@ def verify_ledger(ledger: LedgerStore, *, anchor: tuple[int, str] | None = None) continue actual = { "run_id": row["run_id"], "artifact_hash": row["artifact_hash"], - "source_hash": row["source_hash"], "parser": parser, + "source_hash": row["source_hash"], "source_kind": row["source_kind"], + "source_id": row["source_id"], "backend": row["backend"], "parser": parser, "parser_config": config, "run_conditions": conditions, "state": row["state"], + "error": row["error"], "created_at": row["created_at"], } for key in sorted(want): if actual[key] != want[key]: @@ -283,6 +289,19 @@ def verify_ledger(ledger: LedgerStore, *, anchor: tuple[int, str] | None = None) ) if manifest_hash != row["artifact_hash"]: problems.append(f"record {record_id}: artifact_manifest no longer matches artifact_hash") + # source_path is the locator of the Source that source_id hashes; + # the chain commits to source_id, so the archive vouches for it. + if row["source_id"] is not None: + try: + locator = ledger.archive.get_source(row["source_id"]).locator + except ArchiveError as exc: + problems.append(f"record {record_id}: its Source is unreadable: {exc}") + else: + if row["source_path"] != locator: + problems.append( + f"record {record_id}: source_path is {row['source_path']!r}, " + f"its Source says {locator!r}" + ) rows = {r["dispatch_id"]: r for r in conn.execute("SELECT * FROM deliveries")} for dispatch_id in sorted(set(rows) | set(deliveries)): @@ -306,8 +325,17 @@ def verify_ledger(ledger: LedgerStore, *, anchor: tuple[int, str] | None = None) if row is None or want is None: where = "the chain" if row is None else "the ledger" problems.append(f"replay {replay_id} is only in {where}") - elif {k: row[k] for k in want} != want: - problems.append(f"replay {replay_id} differs from the chain") + continue + try: + actual = _replay_row(row) + except (ValueError, TypeError) as exc: + problems.append(f"replay {replay_id} has a malformed column: {exc!r}") + continue + for key in sorted(want): + if actual[key] != want[key]: + problems.append( + f"replay {replay_id}: {key} is {actual[key]!r}, the chain says {want[key]!r}" + ) return LedgerReport(chain=chain, problems=tuple(problems)) finally: log.close() @@ -323,22 +351,50 @@ def _apply( """Fold one event into the state the ledger's tables must hold.""" b = ev.body if ev.kind in ("record.created", "record.imported"): - records[ev.subject] = { + want = { "run_id": b["run_id"], "artifact_hash": b["artifact_hash"], "source_hash": b.get("source_hash"), + "source_kind": b.get("source_kind"), "source_id": b.get("source_id"), + "backend": b.get("backend"), "parser": b.get("parser"), "parser_config": b.get("parser_config"), "run_conditions": b.get("run_conditions"), "state": b.get("state", "pending"), + "error": b.get("error"), } + # Chained since the event log first committed to it; older events + # leave the record's creation time unchecked. + if "created_at" in b: + want["created_at"] = b["created_at"] + records[ev.subject] = want elif ev.kind in _RECORD_STATE: if ev.subject in records: - records[ev.subject]["state"] = _RECORD_STATE[ev.kind] + want = records[ev.subject] + want["state"] = _RECORD_STATE[ev.kind] + if ev.kind == "record.failed": + want["error"] = b["error"] + elif ev.kind == "record.invalidated": + want["error"] = f"INVALIDATED: {b['reason']}" elif ev.kind in ("delivery.opened", "delivery.imported"): deliveries[ev.subject] = {k: b[k] for k in ("record_id", "target_kind", "target")} elif ev.kind in ("delivery.event", "delivery_event.imported"): delivery_events.setdefault(ev.subject, []).append(_event_tuple(b)) elif ev.kind in ("replay.logged", "replay.imported"): - replays[ev.subject] = {k: b.get(k) for k in ("record_id", "outcome", "replay_artifact_hash")} + # Every field the event carries (replay.imported carries fewer). + replays[ev.subject] = { + k: b[k] for k in _REPLAY_FIELDS if k in b + } + + +_REPLAY_FIELDS = ( + "record_id", "outcome", "reason", "replay_artifact_hash", "policy", "platform", +) + + +def _replay_row(row: sqlite3.Row) -> dict[str, Any]: + """A replays row in the form the chain commits to (policy as an object).""" + out = {k: row[k] for k in _REPLAY_FIELDS} + out["policy"] = None if row["policy"] is None else json.loads(row["policy"]) + return out def _event_tuple(e: Any) -> tuple: @@ -373,6 +429,8 @@ def record_body(row: sqlite3.Row | dict[str, Any], *, with_state: bool = False) "run_conditions": ( None if row["run_conditions"] is None else json.loads(row["run_conditions"]) ), + # Observations (stele.identity) name the run by when it was recorded. + "created_at": row["created_at"], } if with_state: body["state"] = row["state"] @@ -401,9 +459,7 @@ def backfill(conn: sqlite3.Connection) -> int: }) count += 1 for row in conn.execute("SELECT * FROM replays ORDER BY replayed_at, replay_id").fetchall(): - append_event(conn, "replay.imported", row["replay_id"], { - k: row[k] for k in ("record_id", "outcome", "reason", "replay_artifact_hash") - }) + append_event(conn, "replay.imported", row["replay_id"], _replay_row(row)) count += 1 return count @@ -417,7 +473,7 @@ def _main(argv: list[str] | None = None) -> int: from pathlib import Path from ..archive import BlobStore - from .store import LedgerStore + from .store import LedgerSchemaError, LedgerStore ap = argparse.ArgumentParser( prog="python -m stele.ledger.events", @@ -432,7 +488,13 @@ def _main(argv: list[str] | None = None) -> int: if args.anchor: seq, _, digest = args.anchor.partition(":") anchor = (int(seq), digest) - ledger = LedgerStore(args.ledger, BlobStore(args.archive)) + try: + # Read-only: migrating a ledger older than the event log would chain + # its current tables and then report them as verified. + ledger = LedgerStore(args.ledger, BlobStore(args.archive), migrate=False) + except LedgerSchemaError as exc: + print(json.dumps({"ok": False, "length": 0, "head": None, "problems": [str(exc)]}, indent=2)) + return 1 report = verify_ledger(ledger, anchor=anchor) print(json.dumps({ "ok": report.ok, diff --git a/stele/ledger/store.py b/stele/ledger/store.py index 2aa0f17..a333fa4 100644 --- a/stele/ledger/store.py +++ b/stele/ledger/store.py @@ -387,15 +387,60 @@ class LedgerStore: (callers serialize writes externally or rely on SQLite's own locking). """ - def __init__(self, db_path: Path, archive: BlobStore) -> None: + def __init__(self, db_path: Path, archive: BlobStore, *, migrate: bool = True) -> None: + """Open (or create) the ledger at db_path. + + With migrate=False, as verifiers open it, the ledger must already + exist at the current schema: nothing is created or migrated, and + LedgerSchemaError says why a ledger cannot be opened. A verifier + must never rewrite what it verifies; migrating a ledger older than + the event log would chain its current tables as *.imported events + and then vouch for them. + """ if not isinstance(archive, BlobStore): raise TypeError("LedgerStore needs the BlobStore its records point into") self.archive = archive self.db_path = Path(db_path) + if not migrate: + if not self.db_path.is_file(): + raise LedgerSchemaError(f"no ledger at {self.db_path}") + self._conn = connect(self.db_path) + try: + self._check_schema() + except BaseException: + self._conn.close() + raise + return self.db_path.parent.mkdir(parents=True, exist_ok=True) self._conn = connect(self.db_path) self._open_schema() + def _check_schema(self) -> None: + version = self._conn.execute("PRAGMA user_version").fetchone()[0] + has_table = self._conn.execute( + "SELECT 1 FROM sqlite_master WHERE type='table' AND name='artifact_records'" + ).fetchone() is not None + if not has_table: + raise LedgerSchemaError(f"{self.db_path} is not a Stele ledger") + if version > SCHEMA_VERSION: + raise LedgerSchemaError( + f"ledger schema version {version} is newer than this Stele ({SCHEMA_VERSION})" + ) + if version < SCHEMA_VERSION: + has_events = self._conn.execute( + "SELECT 1 FROM sqlite_master WHERE type='table' AND name='events'" + ).fetchone() is not None + detail = ( + "" if has_events else + "; it predates the event log, so there is no chain to verify its " + "tables against" + ) + raise LedgerSchemaError( + f"ledger schema version {version} is older than this Stele ({SCHEMA_VERSION})" + f"{detail}. Verification never migrates a ledger; open it with " + "LedgerStore (e.g. by recording a run) to migrate it first" + ) + def _open_schema(self) -> None: from .migration import migrate_to_current diff --git a/stele/verification/__main__.py b/stele/verification/__main__.py index eb9b152..cba9f01 100644 --- a/stele/verification/__main__.py +++ b/stele/verification/__main__.py @@ -18,15 +18,20 @@ from pathlib import Path from ..archive import BlobStore -from ..ledger.store import LedgerStore +from ..ledger.store import LedgerSchemaError, LedgerStore from .campaign import Campaign, Journal, Observation from .corpus import Corpus from .report import Gates, VerificationReport, Waiver, build_report from .signoff import SignOffRefused, check_sign_off, sign_off, sign_offs -def _ledger(args: argparse.Namespace) -> LedgerStore: - return LedgerStore(args.ledger, BlobStore(args.archive)) +def _ledger(args: argparse.Namespace, *, migrate: bool = False) -> LedgerStore: + """The ledger; only `run`, which records runs, may migrate it. + + report, sign-off, check and list verify the ledger, and a verifier never + migrates what it verifies (LedgerSchemaError says why it cannot open it). + """ + return LedgerStore(args.ledger, BlobStore(args.archive), migrate=migrate) def _store_args(p: argparse.ArgumentParser) -> None: @@ -72,7 +77,7 @@ def _cmd_run(args: argparse.Namespace) -> int: config = {**defaults, **config} else: lanes = lane_mod.extractor_lanes(args.extractor, backends) - ledger = _ledger(args) + ledger = _ledger(args, migrate=True) try: campaign = Campaign( ledger, Corpus.load(args.corpus), args.root, lanes, baseline=args.baseline, @@ -220,7 +225,11 @@ def main(argv: list[str] | None = None) -> int: p.set_defaults(func=_cmd_list) args = ap.parse_args(argv) - return args.func(args) + try: + return args.func(args) + except LedgerSchemaError as exc: + print(f"cannot open the ledger: {exc}", file=sys.stderr) + return 1 if __name__ == "__main__": diff --git a/tests/test_chatgpt_adapter.py b/tests/test_chatgpt_adapter.py index b977044..5bc8351 100644 --- a/tests/test_chatgpt_adapter.py +++ b/tests/test_chatgpt_adapter.py @@ -222,6 +222,42 @@ def test_duplicate_conversation_ids_stay_apart(tmp_path) -> None: assert "conv-branched#2:a1r" in ids +def test_a_renamed_duplicate_never_collides_with_a_real_id(tmp_path) -> None: + # Two copies of "conv-branched" plus a real conversation already called + # "conv-branched#2": the copy's suffix must not land on the real id. + real = json.loads(json.dumps(BRANCHED)) + real["conversation_id"] = "conv-branched#2" + chunks = ChatGPTExportAdapter().transform( + _bundle(tmp_path, _export(BRANCHED, real, BRANCHED)) + ) + ids = [c.chunk_id for c in chunks] + assert len(ids) == len(set(ids)) == 15 + conversations = {c.metadata["conversation_id"] for c in chunks} + assert conversations == {"conv-branched", "conv-branched%232", "conv-branched#2"} + + +def test_colons_in_ids_cannot_make_two_chunks_one(tmp_path) -> None: + # Without encoding, conversation "a:b" node "c" and conversation "a" node + # "b:c" would both be the chunk "a:b:c". + first = {"conversation_id": "a:b", "mapping": { + "c": _node("c", None, [], _msg("c", "user", ["one"])), + }} + second = {"conversation_id": "a", "mapping": { + "b:c": _node("b:c", None, [], _msg("b:c", "user", ["two"])), + }} + chunks = ChatGPTExportAdapter().transform(_bundle(tmp_path, _export(first, second))) + assert sorted(c.chunk_id for c in chunks) == ["a%3Ab:c", "a:b%3Ac"] + # The node id in metadata stays raw: extraction anchors point into mapping by it. + assert {c.metadata["node_id"] for c in chunks} == {"c", "b:c"} + + +def test_uuid_ids_are_unchanged_by_encoding(tmp_path) -> None: + conversation = json.loads(json.dumps(BRANCHED)) + conversation["conversation_id"] = "6710a4c5-1b2e-8003-a4b9-2f6c1a7d9e10" + chunks = ChatGPTExportAdapter().transform(_bundle(tmp_path, _export(conversation))) + assert "6710a4c5-1b2e-8003-a4b9-2f6c1a7d9e10:a1r" in {c.chunk_id for c in chunks} + + def test_deep_conversation_does_not_recurse(tmp_path) -> None: depth = 5000 mapping = { @@ -244,6 +280,8 @@ def test_deep_conversation_does_not_recurse(tmp_path) -> None: (lambda c: c.update(mapping=[]), "mapping is not an object"), (lambda c: c["mapping"]["q1"].update(children="a1"), "children that are not a list"), (lambda c: c.update(current_node=["a1r"]), "current_node"), + (lambda c: c["mapping"]["a1"]["message"].update(author="assistant"), + "author that is not an object"), ]) def test_malformed_conversations_fail(tmp_path, mutate, message) -> None: conversation = json.loads(json.dumps(BRANCHED)) diff --git a/tests/test_event_log.py b/tests/test_event_log.py index 8459d8f..dec3598 100644 --- a/tests/test_event_log.py +++ b/tests/test_event_log.py @@ -234,7 +234,9 @@ def test_record_flipped_back_by_hand(self, ledger, tmp_path) -> None: ) report = verify_ledger(ledger) assert report.chain.ok and not report.ok + # Both halves of the edit are reported: the state and the cleared reason. assert report.problems == ( + f"record {ids['withdrawn']}: error is None, the chain says 'INVALIDATED: superseded'", f"record {ids['withdrawn']}: state is 'sealed', the chain says 'invalidated'", ) @@ -289,7 +291,9 @@ def test_delivery_and_replay_rows_changed(self, ledger, tmp_path) -> None: conn.execute("UPDATE replays SET outcome='reproduced' WHERE replay_id=?", (ids["replay"],)) problems = verify_ledger(ledger).problems assert any("its events differ from the chain" in p for p in problems) - assert f"replay {ids['replay']} differs from the chain" in problems + assert ( + f"replay {ids['replay']}: outcome is 'reproduced', the chain says 'unreplayable'" + ) in problems def test_truncation_is_caught_by_an_anchor(self, ledger, tmp_path) -> None: _lifecycle(ledger, tmp_path) @@ -379,3 +383,168 @@ def test_cli_reports_head_and_problems(ledger, tmp_path) -> None: code, report = _cli(ledger, "--anchor", f"{seq}:{digest}") assert code == 1 and not report["ok"] assert any("the chain says 'invalidated'" in p for p in report["problems"]) + + +# --------------------------------------------------------------------------- +# Every chained field is checked +# --------------------------------------------------------------------------- + +def _with_input(ledger: LedgerStore, tmp_path: Path): + """A SEALED record over an archived input Snapshot and Source, run by bubblewrap.""" + from stele.archive import Source + from stele.archive.records import Snapshot, SnapshotKind + + doc = b"%PDF-1.7 the input" + digest = ledger.archive.put_bytes(doc) + snapshot = ledger.archive.put_snapshot(Snapshot(SnapshotKind.FILE, digest, len(doc), 1)) + out = tmp_path / f"out-{uuid.uuid4().hex[:6]}" + out.mkdir() + (out / "doc.txt").write_text("hello") + record = ledger.create_pending( + **PROVENANCE, run_id=str(uuid.uuid4()), artifact_dir=out, + artifact_paths=[out / "doc.txt"], input_snapshot=snapshot, + source=Source(locator="/corpus/report.pdf"), backend="bubblewrap", + ) + return ledger.seal(record.record_id) + + +class TestEveryChainedField: + + @pytest.mark.parametrize(("column", "value", "problem"), [ + ("source_kind", "tree", "source_kind is 'tree', the chain says 'file'"), + ("source_id", "f" * 64, "source_id is 'ffff"), + ("backend", "wasmtime", "backend is 'wasmtime', the chain says 'bubblewrap'"), + ("created_at", "2001-01-01T00:00:00+00:00", "created_at is '2001-01-01T00:00:00+00:00'"), + ("source_path", "/elsewhere/report.pdf", + "source_path is '/elsewhere/report.pdf', its Source says '/corpus/report.pdf'"), + ]) + def test_record_field_edited(self, ledger, tmp_path, column, value, problem) -> None: + record = _with_input(ledger, tmp_path) + assert verify_ledger(ledger).ok + _raw(ledger).execute( + f"UPDATE artifact_records SET {column}=? WHERE record_id=?", (value, record.record_id) + ) + report = verify_ledger(ledger) + assert report.chain.ok and not report.ok + assert any(problem in p for p in report.problems), report.problems + + def test_invalidation_reason_edited(self, ledger, tmp_path) -> None: + ids = _lifecycle(ledger, tmp_path) + _raw(ledger).execute( + "UPDATE artifact_records SET error='INVALIDATED: routine cleanup' WHERE record_id=?", + (ids["withdrawn"],), + ) + assert verify_ledger(ledger).problems == ( + f"record {ids['withdrawn']}: error is 'INVALIDATED: routine cleanup', " + "the chain says 'INVALIDATED: superseded'", + ) + + def test_failure_reason_edited(self, ledger, tmp_path) -> None: + ids = _lifecycle(ledger, tmp_path) + _raw(ledger).execute( + "UPDATE artifact_records SET error='out of disk' WHERE record_id=?", (ids["failed"],) + ) + assert verify_ledger(ledger).problems == ( + f"record {ids['failed']}: error is 'out of disk', the chain says 'parser crashed'", + ) + + @pytest.mark.parametrize(("column", "value"), [ + ("reason", "looked fine"), + ("platform", "plan9-mips"), + ("policy", '{"name":"lenient"}'), + ]) + def test_replay_field_edited(self, ledger, tmp_path, column, value) -> None: + ids = _lifecycle(ledger, tmp_path) + _raw(ledger).execute(f"UPDATE replays SET {column}=? WHERE replay_id=?", (value, ids["replay"])) + report = verify_ledger(ledger) + assert any(p.startswith(f"replay {ids['replay']}: {column} is") for p in report.problems) + + def test_an_intact_lifecycle_still_verifies(self, ledger, tmp_path) -> None: + _lifecycle(ledger, tmp_path) + _with_input(ledger, tmp_path) + assert verify_ledger(ledger).ok + + +# --------------------------------------------------------------------------- +# Verifiers never migrate what they verify +# --------------------------------------------------------------------------- + +def _pre_event_log(db: Path, tmp_path: Path) -> str: + """A version-6 ledger (no event log) whose invalidated record was then un-invalidated.""" + ledger = open_ledger(db) + record = ledger.seal(_pending(ledger, tmp_path).record_id) + ledger.invalidate(record.record_id, "withdrawn") + ledger.close() + conn = sqlite3.connect(db) + conn.execute("DROP TABLE events") + conn.execute("ALTER TABLE delivery_events DROP COLUMN attempt_id") + conn.execute("PRAGMA user_version = 6") + conn.execute("UPDATE artifact_records SET state='sealed', error=NULL") + conn.commit() + conn.close() + return record.record_id + + +def _version(db: Path) -> int: + conn = sqlite3.connect(db) + try: + return conn.execute("PRAGMA user_version").fetchone()[0] + finally: + conn.close() + + +class TestVerifiersDoNotMigrate: + + def test_an_older_ledger_is_refused_and_left_alone(self, tmp_path) -> None: + from stele.archive import BlobStore + from stele.ledger.store import LedgerSchemaError + + db = tmp_path / "ledger.db" + _pre_event_log(db, tmp_path) + with pytest.raises(LedgerSchemaError, match="predates the event log"): + LedgerStore(db, BlobStore(tmp_path / "archive"), migrate=False) + assert _version(db) == 6 + + def test_a_missing_ledger_is_not_created(self, tmp_path) -> None: + from stele.archive import BlobStore + from stele.ledger.store import LedgerSchemaError + + with pytest.raises(LedgerSchemaError, match="no ledger"): + LedgerStore(tmp_path / "none" / "ledger.db", BlobStore(tmp_path / "archive"), + migrate=False) + assert not (tmp_path / "none").exists() + + def test_a_current_ledger_opens(self, ledger, tmp_path) -> None: + from stele.archive import BlobStore + + _lifecycle(ledger, tmp_path) + again = LedgerStore(ledger.db_path, BlobStore(tmp_path / "archive"), migrate=False) + assert verify_ledger(again).ok + + def test_the_cli_refuses_instead_of_certifying(self, tmp_path) -> None: + db = tmp_path / "ledger.db" + _pre_event_log(db, tmp_path) + out = subprocess.run( + [sys.executable, "-m", "stele.ledger.events", str(db), str(tmp_path / "archive")], + capture_output=True, text=True, + ) + assert out.returncode == 1 + report = json.loads(out.stdout) + assert report["ok"] is False and "predates the event log" in report["problems"][0] + assert _version(db) == 6 # untouched: nothing was imported into a new chain + + @pytest.mark.parametrize("argv", [ + lambda db, archive, rid: ["-m", "stele.identity", "refs", db, archive, rid], + lambda db, archive, rid: ["-m", "stele.extraction", "extract", db, archive, rid], + lambda db, archive, rid: ["-m", "stele.verification", "list", + "--ledger", db, "--archive", archive], + ], ids=["identity", "extraction", "verification"]) + def test_other_verifiers_refuse_too(self, tmp_path, argv) -> None: + db = tmp_path / "ledger.db" + record_id = _pre_event_log(db, tmp_path) + out = subprocess.run( + [sys.executable, *argv(str(db), str(tmp_path / "archive"), record_id)], + capture_output=True, text=True, + ) + assert out.returncode == 1 and "cannot open the ledger" in out.stderr + assert _version(db) == 6 diff --git a/tests/test_identity.py b/tests/test_identity.py index d17cdca..a285e15 100644 --- a/tests/test_identity.py +++ b/tests/test_identity.py @@ -221,6 +221,23 @@ def test_a_different_observation_is_refused(self, tmp_path) -> None: problems = IdentityResolver(ledger).check(other.ref) assert problems and "observation has digest" in problems[0] + @pytest.mark.parametrize("column", ["created_at", "source_id"]) + def test_an_observation_rebuilt_from_an_edited_row_is_refused(self, tmp_path, column) -> None: + # Edit the row, then present the observation rebuilt from it. Its + # digest matches the row, so only the record's creation event in the + # chain can show that the row is not what was recorded. + ledger, record = _sealed(tmp_path) + other_source = ledger.archive.put_source(Source(locator="/somewhere/else.pdf")) + value = {"created_at": "2001-01-01T00:00:00+00:00", "source_id": other_source}[column] + conn = sqlite3.connect(ledger.db_path, isolation_level=None) + conn.execute(f"UPDATE artifact_records SET {column}=? WHERE record_id=?", + (value, record.record_id)) + conn.close() + forged = observation_of(ledger.get(record.record_id)) + problems = IdentityResolver(ledger).check(forged.ref) + assert problems and f"the record's {column} is" in problems[0] + assert "creation event" in problems[0] + def test_record_without_input_has_no_observation(self, tmp_path) -> None: ledger = open_ledger(tmp_path / "ledger.db") out = tmp_path / "out"