Verify every chained field; verifiers never migrate; harden ChatGPT ids

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, `<conversation_id>:<node_id>`, stable across
-  re-exports.
+  content becomes one chunk, `<conversation key>:<node key>`, stable across
+  re-exports. Both parts are percent-encoded, so neither contains `:` or
+  `#`. The `#<n>` 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;

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`

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 "<conversation_id>:<node_id>", stable across re-exports, so a
-re-delivered conversation updates rather than duplicates.
+Chunk ids are "<conversation key>:<node key>", 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 "#<n>" 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,

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)

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:

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: