diff --git a/README.md b/README.md index 3875d26..c23edf4 100644 --- a/README.md +++ b/README.md @@ -156,6 +156,7 @@ Details and known limits: [docs/containment.md](docs/containment.md). | [Replay](docs/replay.md) | Validation, replay outcomes, invalidation, ledger views | | [Adapter contract](docs/adapter.md) | `SteleAdapter`, `TargetWriter`, the Dispatcher and delivery log, the ChatGPT export adapter | | [Extraction contract](docs/extraction.md) | `stele.extraction` v1, normalizers, the trusted resolver, `ExtractionAdapter` | +| [Identity contract](docs/identity.md) | `stele:` references to sources, snapshots, records, observations, anchors and events, and how they resolve | | [Evidence store](docs/archive.md) | Content-addressed Snapshot and artifact archive | | [Cloud sandboxes](docs/cloud-sandboxes.md) | Design note for hosted sandbox backends (not implemented) | | [Parser images](parsers/README.md) | Building and running MinerU, Marker, Docling | diff --git a/docs/extraction.md b/docs/extraction.md index 1a15900..51efdea 100644 --- a/docs/extraction.md +++ b/docs/extraction.md @@ -127,8 +127,10 @@ A `SteleAdapter` for any bundle that has a normalizer. fails the transform, and nothing reaches a target. - It emits one chunk per unit. The chunk id is `:`. - The chunk metadata holds `kind`, `order`, `page`, `bbox`, `level`, - `parent` (as a chunk id), `anchor`, `attributes`, `schema` and - `normalizer`. + `parent` (as a chunk id), `anchor`, `attributes`, `schema`, + `normalizer`, and `stele_anchor`: the unit's anchor as a + [Stele reference](identity.md) (`stele:anchor:...`) that a consumer can + resolve to check a citation against the evidence. ## CLI diff --git a/docs/identity.md b/docs/identity.md new file mode 100644 index 0000000..13431c1 --- /dev/null +++ b/docs/identity.md @@ -0,0 +1,117 @@ +# Identity Contract + +Each layer of Stele names its evidence in its own way: + +- a Snapshot by `(kind, digest)` +- a Source by `source_id` +- a ledger record by `record_id` and `artifact_hash` +- an extraction unit by an anchor into a sealed artifact +- an event by `(seq, hash)` in the hash-chained log + +`stele.identity` gives each of these one typed reference with one canonical +string. A downstream system can store, pass on and cite that string, and check +it later. + +Every reference **commits to content**: it contains the digest of what it +names. It therefore resolves only while that evidence is unchanged. + +## References (`stele/identity/refs.py`) + +| Type | Canonical string | Commits to | +|------|------------------|------------| +| `SourceRef` | `stele:source:` | the Source record (a descriptive locator) | +| `SnapshotRef` | `stele:snapshot::` | the exact bytes a parser saw | +| `RecordRef` | `stele:record:@` | one run's sealed output tree | +| `ObservationRef` | `stele:observation:@` | one run's Observation (below) | +| `AnchorRef` | `stele:anchor:@/@#bytes=-` | a byte range of one sealed artifact | +| | `stele:anchor:@/@#pointer=&render=` | a JSON value of one sealed artifact, rendered by a named trusted renderer | +| `EventRef` | `stele:event:@` | one event, and through the chain, all history before it | + +**An Observation** records that one run saw its input: + +``` +{schema: "stele.observation", schema_version: 1, + record_id, source_id (or null), snapshot: {kind, digest}, observed_at} +``` + +- `observed_at` is the record's `created_at`, in ISO 8601 (UTC). +- Its reference names the record plus the SHA-256 of that canonical JSON. Each + field is covered, so a different source, snapshot or time gives a different + reference. + +**Canonical strings.** `parse_ref()` accepts a string only if formatting the +parsed reference gives the same string back, so each reference has exactly one +spelling. + +- In anchors, the artifact path and the JSON pointer are percent-encoded. + Unreserved characters and `/` stay literal. Everything else becomes `%XX` + with upper-case hex, as `urllib.parse.quote` writes it. +- A renderer name is encoded the same way, but with `/` escaped too. +- Numbers have no leading zeros. Digests are lower-case hex. + +**Constructors:** + +| Function | Returns | +|---|---| +| `record_ref(record)` | the `RecordRef` of an `ArtifactRecord` or `SealedBundle` | +| `observation_of(record)` | the record's `Observation`; its `.ref` is the `ObservationRef` | +| `anchor_ref(extraction, unit)` | the `AnchorRef` of one extraction unit | +| `event_ref(event)` | the `EventRef` of an event from `EventLog` | + +## Resolution (`stele/identity/resolver.py`) + +`IdentityResolver(ledger).resolve(ref)` accepts a reference or its string. It +returns what the reference names or raises `ResolveError`. `check(ref)` returns +the problems as a list instead. Each type is checked against the evidence, +never against the caller's word: + +| Reference | Resolves when | Returns | +|---|---|---| +| `SourceRef` | The Source record is in the archive and hashes to `source_id`. | `Source` | +| `SnapshotRef` | The Snapshot record is in the archive, and every blob it covers re-hashes: the file, or the tree object and each file it lists. | `Snapshot` | +| `RecordRef` | The record exists and is `SEALED` (or `INVALIDATED` with `allow_invalidated=True`). It must have that `artifact_hash`, and the archived tree object must list exactly the record's manifest. | `ArtifactRecord` | +| `ObservationRef` | The Observation rebuilt from the record has that digest, and its Snapshot and Source resolve. An observation stays true whatever happened to the run later, so it resolves in any record state. | `Observation` | +| `AnchorRef` | The record resolves as a `RecordRef`. The [extraction resolver](extraction.md) then reads the span from bytes the archive has just re-hashed. | the anchored text | +| `EventRef` | The chain verifies with this event as its anchor: the event is at `seq` with this hash, and nothing before or after it has been altered, removed or reordered. | `Event` | + +**What resolution does not check:** +- A `RecordRef` checks the tree object but not every artifact blob. An + `AnchorRef` re-hashes the one blob it reads. +- An `EventRef` depends on the whole chain being intact. A chain broken + anywhere resolves no event, because none of its hashes can be trusted until + the break is explained. +- As with the event log, anyone with write access to the ledger and archive + can rewrite them consistently. A reference held outside that access, such + as a published `EventRef`, is what detects that. + +## Citations + +`ExtractionAdapter` chunks carry their unit's anchor reference in +`metadata["stele_anchor"]`. A system that quotes a chunk can keep that string +next to the quote. Resolving it later checks that the quoted text is still +exactly what the sealed evidence says, and that the record still stands. + +## CLI + +```bash +python -m stele.identity refs ledger.db archive/ + # {"record": "stele:record:...", "observation": "...", "snapshot": "...", "events": [...]} +python -m stele.identity resolve ledger.db archive/ ... [--allow-invalidated] + # one JSON line per reference; exit 1 if any does not resolve +``` + +## Tests + +`tests/test_identity.py`: + +| Property | Tests | +|---|---| +| Every type round-trips through one canonical string, and other spellings are refused | `TestFormat` | +| Every type resolves to what it names | `TestResolve::test_every_type_resolves_to_what_it_names` | +| Wrong digests, kinds, records, artifacts and ranges are refused | `TestResolve::test_wrong_digests_are_refused` | +| A different observation is refused | `TestResolve::test_a_different_observation_is_refused` | +| Invalidated records resolve only on request; their observations still hold | `TestResolve::test_invalidated_record_needs_permission` | +| Tampered output and input blobs are refused | `TestResolve::test_tampered_evidence_is_refused` | +| A rewritten event is refused | `TestResolve::test_rewritten_event_is_refused` | +| Extraction chunks cite resolvable anchors | `test_extraction_chunks_carry_resolvable_anchor_refs` | +| CLI | `test_cli_refs_and_resolve` | diff --git a/stele/adapters/extraction.py b/stele/adapters/extraction.py index 54acbde..4c08db5 100644 --- a/stele/adapters/extraction.py +++ b/stele/adapters/extraction.py @@ -6,8 +6,10 @@ verified bytes by the resolver before anything is returned, so a normalizer bug that misquotes the evidence fails the transform instead of reaching a target. Chunk ids are ":"; metadata carries the unit's -kind, order, page, bbox, level, parent, anchor and attributes, plus the -extraction's schema and normalizer. +kind, order, page, bbox, level, parent, anchor and attributes, the +extraction's schema and normalizer, and stele_anchor: the unit's anchor as a +Stele reference (stele.identity), which a consumer can resolve to check a +citation against the evidence. """ from __future__ import annotations @@ -22,6 +24,7 @@ def transform(self, bundle: SealedBundle) -> list[SteleChunk]: # Imported here: the normalizers build on the other adapters in this package. from ..extraction.normalizers import normalize from ..extraction.resolver import BundleResolver, ResolutionError + from ..identity.refs import anchor_ref extraction = normalize(bundle) problems = BundleResolver(bundle).verify(extraction) @@ -45,6 +48,7 @@ def transform(self, bundle: SealedBundle) -> list[SteleChunk]: level=unit.level, parent=f"{bundle.record_id}:{unit.parent}" if unit.parent else None, anchor=unit.anchor.to_json(), + stele_anchor=str(anchor_ref(extraction, unit)), attributes=dict(unit.attributes), ) for unit in extraction.units diff --git a/stele/archive/store.py b/stele/archive/store.py index 27daa7c..c372cc6 100644 --- a/stele/archive/store.py +++ b/stele/archive/store.py @@ -428,6 +428,10 @@ def has_snapshot(self, digest: str, kind: SnapshotKind | str) -> bool: kind = SnapshotKind(kind) return self._sharded(self._snapshots / kind.value, _check_digest(digest)).is_file() + def verify_snapshot(self, snapshot: Snapshot) -> None: + """Re-hash every blob a recorded snapshot covers (IntegrityError if any changed).""" + self._check_snapshot_content(snapshot) + def _check_snapshot_content(self, snapshot: Snapshot) -> None: """Re-hash every blob the snapshot covers; a size match is not enough.""" if snapshot.kind is SnapshotKind.FILE: diff --git a/stele/identity/__init__.py b/stele/identity/__init__.py new file mode 100644 index 0000000..e4ecdb1 --- /dev/null +++ b/stele/identity/__init__.py @@ -0,0 +1,23 @@ +"""The identity contract (roadmap #42): typed, self-verifying references to evidence. + +refs + SourceRef, SnapshotRef, RecordRef, ObservationRef, AnchorRef, EventRef: + one canonical string each ("stele::..."), strict parse_ref(), and + constructors from ledger records, extraction units and events. +resolver + IdentityResolver(ledger).resolve(ref) checks a reference against the + ledger, the evidence archive and the event chain, and returns what it names. + +CLI: python -m stele.identity resolve ... +""" +from .refs import ( + PREFIX, AnchorRef, EventRef, Observation, ObservationRef, RecordRef, Ref, RefFormatError, + SnapshotRef, SourceRef, anchor_ref, event_ref, observation_of, parse_ref, record_ref, +) +from .resolver import IdentityResolver, ResolveError + +__all__ = [ + "PREFIX", "AnchorRef", "EventRef", "Observation", "ObservationRef", "RecordRef", "Ref", + "RefFormatError", "SnapshotRef", "SourceRef", "anchor_ref", "event_ref", "observation_of", + "parse_ref", "record_ref", "IdentityResolver", "ResolveError", +] diff --git a/stele/identity/__main__.py b/stele/identity/__main__.py new file mode 100644 index 0000000..19408eb --- /dev/null +++ b/stele/identity/__main__.py @@ -0,0 +1,93 @@ +""" +python -m stele.identity resolve ... [--allow-invalidated] + Resolve each reference against the evidence and print one JSON result per + reference; exit 1 if any does not resolve. +python -m stele.identity refs + Print the record, observation and event references of one record. +""" +from __future__ import annotations + +import argparse +import dataclasses +import json +import sys +from pathlib import Path +from typing import Any + +from ..archive import BlobStore, Snapshot, Source +from ..ledger.events import EventLog +from ..ledger.models import ArtifactRecord +from ..ledger.store import LedgerStore, RecordNotFoundError +from .refs import Observation, event_ref, observation_of, record_ref +from .resolver import IdentityResolver, ResolveError + + +def _summary(value: Any) -> Any: + """A small JSON description of what a reference resolved to.""" + if isinstance(value, str): + return {"text": value} + if isinstance(value, ArtifactRecord): + return {"record_id": value.record_id, "state": value.state.value, + "artifact_hash": value.artifact_hash, "files": len(value.artifact_manifest)} + if isinstance(value, Observation): + return json.loads(value.to_canonical()) + if isinstance(value, (Source, Snapshot)): + return json.loads(value.to_canonical()) + if dataclasses.is_dataclass(value): + return {k: v for k, v in dataclasses.asdict(value).items()} + return repr(value) + + +def main(argv: list[str] | None = None) -> int: + ap = argparse.ArgumentParser(prog="python -m stele.identity", + description="Stele references: resolve and list them.") + sub = ap.add_subparsers(dest="command", required=True) + re_ = sub.add_parser("resolve", help="resolve references against the evidence") + re_.add_argument("ledger", type=Path) + re_.add_argument("archive", type=Path) + re_.add_argument("refs", nargs="+") + re_.add_argument("--allow-invalidated", action="store_true") + rf = sub.add_parser("refs", help="the references of one record") + rf.add_argument("ledger", type=Path) + rf.add_argument("archive", type=Path) + rf.add_argument("record_id") + args = ap.parse_args(argv) + + ledger = LedgerStore(args.ledger, BlobStore(args.archive)) + try: + if args.command == "refs": + try: + record = ledger.get(args.record_id) + except RecordNotFoundError: + print(f"no record {args.record_id}", file=sys.stderr) + return 1 + out: dict[str, Any] = {"record": str(record_ref(record))} + if record.source_hash is not None and record.source_kind is not None: + observation = observation_of(record) + out["observation"] = str(observation.ref) + out["snapshot"] = str(observation.snapshot) + log = EventLog(ledger) + try: + out["events"] = [str(event_ref(e)) for e in log.for_subject(record.record_id)] + finally: + log.close() + print(json.dumps(out, indent=2)) + return 0 + resolver = IdentityResolver(ledger, allow_invalidated=args.allow_invalidated) + failed = False + for ref in args.refs: + try: + value = resolver.resolve(ref) + except ResolveError as exc: + failed = True + print(json.dumps({"ref": ref, "ok": False, "problem": str(exc)})) + else: + print(json.dumps({"ref": ref, "ok": True, "value": _summary(value)}, + ensure_ascii=False, default=str)) + return 1 if failed else 0 + finally: + ledger.close() + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/stele/identity/refs.py b/stele/identity/refs.py new file mode 100644 index 0000000..7202bc7 --- /dev/null +++ b/stele/identity/refs.py @@ -0,0 +1,316 @@ +""" +Typed references to Stele evidence, each with exactly one canonical string. + + stele:source: + stele:snapshot:: + stele:record:@ + stele:observation:@ + stele:anchor:@/@#bytes=- + stele:anchor:@/@#pointer=&render= + stele:event:@ + +Every reference commits to content: a digest of the bytes, tree, record or +chain position it names. It therefore resolves (resolver.py) only while that +evidence is unchanged, and it can be passed around as a plain string. + +parse_ref() is strict: a string is accepted only if formatting the parsed +reference gives it back byte for byte, so one reference has one spelling. +In anchors, the artifact path and the JSON pointer are percent-encoded +(RFC 3986: unreserved characters and "/" stay literal, everything else is +%XX with upper-case hex, as urllib.parse.quote writes it). +""" +from __future__ import annotations + +import hashlib +import re +from dataclasses import dataclass +from datetime import datetime +from typing import Any, Union +from urllib.parse import quote, unquote + +from ..archive.records import SnapshotKind, canonical_json, is_digest +from ..extraction.contract import Anchor, Extraction, ExtractionFormatError, Unit + +PREFIX = "stele:" + +# Record ids are UUIDs today; any id made of these characters can be named. +_RECORD_ID = re.compile(r"[A-Za-z0-9][A-Za-z0-9._-]*") +_UINT = re.compile(r"0|[1-9][0-9]*") + + +class RefFormatError(ValueError): + """A string that is not the canonical form of any Stele reference.""" + + +def _check_record_id(record_id: object) -> str: + if not isinstance(record_id, str) or not _RECORD_ID.fullmatch(record_id): + raise RefFormatError(f"not a record id: {record_id!r}") + return record_id + + +def _check_digest(value: object, what: str) -> str: + if not is_digest(value): + raise RefFormatError(f"{what} is not a SHA-256 hex digest: {value!r}") + return value # type: ignore[return-value] + + +def _split_at(text: str, what: str) -> tuple[str, str]: + """'@' -> (left, digest); the digest never contains '@'.""" + left, sep, digest = text.rpartition("@") + if not sep: + raise RefFormatError(f"{what}: expected <...>@, got {text!r}") + return left, digest + + +def _uint(text: str, what: str) -> int: + if not _UINT.fullmatch(text): + raise RefFormatError(f"{what} is not a non-negative integer: {text!r}") + return int(text) + + +@dataclass(frozen=True) +class SourceRef: + """A Source record (a descriptive locator) by its source_id.""" + + source_id: str + + def __post_init__(self) -> None: + _check_digest(self.source_id, "source_id") + + def __str__(self) -> str: + return f"{PREFIX}source:{self.source_id}" + + @classmethod + def _parse(cls, rest: str) -> "SourceRef": + return cls(rest) + + +@dataclass(frozen=True) +class SnapshotRef: + """The exact staged bytes a parser saw: (kind, digest).""" + + kind: SnapshotKind + digest: str + + def __post_init__(self) -> None: + try: + object.__setattr__(self, "kind", SnapshotKind(self.kind)) + except ValueError: + raise RefFormatError(f"not a snapshot kind: {self.kind!r}") from None + _check_digest(self.digest, "snapshot digest") + + def __str__(self) -> str: + return f"{PREFIX}snapshot:{self.kind.value}:{self.digest}" + + @classmethod + def _parse(cls, rest: str) -> "SnapshotRef": + kind, sep, digest = rest.partition(":") + if not sep: + raise RefFormatError(f"snapshot: expected :, got {rest!r}") + return cls(kind, digest) # type: ignore[arg-type] + + +@dataclass(frozen=True) +class RecordRef: + """A ledger record together with the output tree it sealed.""" + + record_id: str + artifact_hash: str + + def __post_init__(self) -> None: + _check_record_id(self.record_id) + _check_digest(self.artifact_hash, "artifact_hash") + + def __str__(self) -> str: + return f"{PREFIX}record:{self.record_id}@{self.artifact_hash}" + + @classmethod + def _parse(cls, rest: str) -> "RecordRef": + return cls(*_split_at(rest, "record")) + + +@dataclass(frozen=True) +class Observation: + """One run's observation: this Source was seen as this Snapshot at this time. + + observed_at is the record's created_at in ISO 8601 (UTC). source_id is None when the run was given no Source. + """ + + record_id: str + source_id: str | None + snapshot: SnapshotRef + observed_at: str + + SCHEMA = "stele.observation" + SCHEMA_VERSION = 1 + + def __post_init__(self) -> None: + _check_record_id(self.record_id) + if self.source_id is not None: + _check_digest(self.source_id, "source_id") + if not isinstance(self.snapshot, SnapshotRef): + raise RefFormatError("observation.snapshot must be a SnapshotRef") + try: + datetime.fromisoformat(self.observed_at) + except (TypeError, ValueError): + raise RefFormatError(f"observed_at is not ISO 8601: {self.observed_at!r}") from None + + def to_canonical(self) -> bytes: + return canonical_json({ + "schema": self.SCHEMA, "schema_version": self.SCHEMA_VERSION, + "record_id": self.record_id, "source_id": self.source_id, + "snapshot": {"kind": self.snapshot.kind.value, "digest": self.snapshot.digest}, + "observed_at": self.observed_at, + }) + + @property + def digest(self) -> str: + return hashlib.sha256(self.to_canonical()).hexdigest() + + @property + def ref(self) -> "ObservationRef": + return ObservationRef(self.record_id, self.digest) + + +@dataclass(frozen=True) +class ObservationRef: + """An Observation, named by its record and the digest of its canonical form.""" + + record_id: str + digest: str + + def __post_init__(self) -> None: + _check_record_id(self.record_id) + _check_digest(self.digest, "observation digest") + + def __str__(self) -> str: + return f"{PREFIX}observation:{self.record_id}@{self.digest}" + + @classmethod + def _parse(cls, rest: str) -> "ObservationRef": + return cls(*_split_at(rest, "observation")) + + +@dataclass(frozen=True) +class AnchorRef: + """One span of one sealed artifact: an extraction Anchor, made addressable.""" + + record: RecordRef + anchor: Anchor + + def __post_init__(self) -> None: + if not isinstance(self.record, RecordRef) or not isinstance(self.anchor, Anchor): + raise RefFormatError("an AnchorRef is a RecordRef and an extraction Anchor") + + def __str__(self) -> str: + a = self.anchor + if a.range is not None: + fragment = f"bytes={a.range[0]}-{a.range[1]}" + else: + fragment = f"pointer={quote(a.pointer)}&render={quote(a.render, safe='')}" + return ( + f"{PREFIX}anchor:{self.record.record_id}@{self.record.artifact_hash}" + f"/{quote(a.artifact)}@{a.digest}#{fragment}" + ) + + @classmethod + def _parse(cls, rest: str) -> "AnchorRef": + location, sep, fragment = rest.partition("#") + if not sep: + raise RefFormatError(f"anchor: missing #bytes= or #pointer= in {rest!r}") + record_part, slash, artifact_part = location.partition("/") + if not slash: + raise RefFormatError(f"anchor: expected /, got {location!r}") + record = RecordRef(*_split_at(record_part, "anchor record")) + artifact, digest = _split_at(artifact_part, "anchor artifact") + try: + if fragment.startswith("bytes="): + start, dash, end = fragment[len("bytes="):].partition("-") + if not dash: + raise RefFormatError(f"anchor: bad byte range {fragment!r}") + anchor = Anchor(artifact=unquote(artifact), digest=digest, + range=(_uint(start, "range start"), _uint(end, "range end"))) + elif fragment.startswith("pointer="): + pointer, amp, render = fragment[len("pointer="):].partition("&render=") + if not amp: + raise RefFormatError(f"anchor: pointer without &render= in {fragment!r}") + anchor = Anchor(artifact=unquote(artifact), digest=digest, + pointer=unquote(pointer), render=unquote(render)) + else: + raise RefFormatError(f"anchor: unknown fragment {fragment!r}") + except ExtractionFormatError as exc: + raise RefFormatError(f"anchor: {exc}") from exc + return cls(record, anchor) + + +@dataclass(frozen=True) +class EventRef: + """An event in the ledger's hash chain: (seq, hash).""" + + seq: int + hash: str + + def __post_init__(self) -> None: + if type(self.seq) is not int or self.seq < 1: + raise RefFormatError(f"event seq must be a positive int, got {self.seq!r}") + _check_digest(self.hash, "event hash") + + def __str__(self) -> str: + return f"{PREFIX}event:{self.seq}@{self.hash}" + + @classmethod + def _parse(cls, rest: str) -> "EventRef": + seq, digest = _split_at(rest, "event") + return cls(_uint(seq, "event seq"), digest) + + +Ref = Union[SourceRef, SnapshotRef, RecordRef, ObservationRef, AnchorRef, EventRef] + +_TYPES: dict[str, Any] = { + "source": SourceRef, "snapshot": SnapshotRef, "record": RecordRef, + "observation": ObservationRef, "anchor": AnchorRef, "event": EventRef, +} + + +def parse_ref(text: str) -> Ref: + """The reference a canonical string names (RefFormatError otherwise).""" + if not isinstance(text, str) or not text.startswith(PREFIX): + raise RefFormatError(f"not a Stele reference: {text!r}") + kind, sep, rest = text[len(PREFIX):].partition(":") + if not sep or kind not in _TYPES: + raise RefFormatError(f"unknown reference type in {text!r}") + ref = _TYPES[kind]._parse(rest) + if str(ref) != text: + raise RefFormatError(f"not the canonical form (that is {str(ref)!r}): {text!r}") + return ref + + +# --------------------------------------------------------------------------- +# From existing objects +# --------------------------------------------------------------------------- + +def record_ref(record: Any) -> RecordRef: + """The reference of a ledger ArtifactRecord (or SealedBundle).""" + return RecordRef(record.record_id, record.artifact_hash) + + +def observation_of(record: Any) -> Observation: + """What a record observed: its Source (if any) as its input Snapshot.""" + if record.source_hash is None or record.source_kind is None: + raise ValueError(f"record {record.record_id} has no input Snapshot to have observed") + return Observation( + record_id=record.record_id, + source_id=record.source_id, + snapshot=SnapshotRef(record.source_kind, record.source_hash), + observed_at=record.created_at.isoformat(), + ) + + +def anchor_ref(extraction: Extraction, unit: Unit) -> AnchorRef: + """The reference of one unit's anchor in an Extraction.""" + return AnchorRef(RecordRef(extraction.record_id, extraction.artifact_hash), unit.anchor) + + +def event_ref(event: Any) -> EventRef: + """The reference of an event from the hash-chained log.""" + return EventRef(event.seq, event.hash) diff --git a/stele/identity/resolver.py b/stele/identity/resolver.py new file mode 100644 index 0000000..b5c984f --- /dev/null +++ b/stele/identity/resolver.py @@ -0,0 +1,147 @@ +""" +Resolve a reference against the evidence, never against the caller's word. + + SourceRef the Source record is in the archive and hashes to source_id + SnapshotRef the Snapshot record is in the archive, and every blob it + covers re-hashes (the file, or the tree and each member) + RecordRef the record exists, is SEALED (INVALIDATED only when + allowed), has that artifact_hash, and the archived tree + object for it lists exactly the record's manifest + ObservationRef the Observation rebuilt from the record (its Source, + input Snapshot and creation time) has that digest, and its + Snapshot resolves. An observation stays true whatever + happened to the run afterwards, so any record state will do. + AnchorRef the record resolves as above, then the extraction + resolver reads the anchored span from archive-verified + bytes (stele.extraction.resolver) + EventRef the event is in the chain at that seq with that hash, and + the whole chain verifies (EventLog.verify with the event + as anchor), so the event still commits to all history + before it + +resolve() returns what the reference names (a Source, Snapshot, +ArtifactRecord, Observation, the anchored text, or an Event) or raises +ResolveError; check() returns the problems as a list instead. +""" +from __future__ import annotations + +from typing import Any + +from ..archive.store import ArchiveError +from ..contracts.adapter import SealedBundle +from ..extraction.resolver import BundleResolver, ResolutionError +from ..ledger.events import EventLog +from ..ledger.models import ArtifactRecord, ArtifactState +from ..ledger.store import LedgerStore, RecordNotFoundError +from .refs import ( + AnchorRef, EventRef, ObservationRef, RecordRef, Ref, RefFormatError, SnapshotRef, SourceRef, + observation_of, parse_ref, +) + + +class ResolveError(ValueError): + """A reference that does not resolve against the evidence.""" + + +class IdentityResolver: + """Resolves Stele references against one ledger and its evidence archive.""" + + def __init__(self, ledger: LedgerStore, *, allow_invalidated: bool = False) -> None: + self.ledger = ledger + self.archive = ledger.archive + self.allow_invalidated = allow_invalidated + + def resolve(self, ref: Ref | str) -> Any: + if isinstance(ref, str): + try: + ref = parse_ref(ref) + except RefFormatError as exc: + raise ResolveError(str(exc)) from exc + method = getattr(self, f"_resolve_{type(ref).__name__}", None) + if method is None: + raise ResolveError(f"not a Stele reference: {ref!r}") + return method(ref) + + def check(self, ref: Ref | str) -> list[str]: + """[] when ref resolves, else the reason it does not.""" + try: + self.resolve(ref) + except ResolveError as exc: + return [str(exc)] + return [] + + # -- one per reference type ---------------------------------------------- + + def _resolve_SourceRef(self, ref: SourceRef) -> Any: + try: + return self.archive.get_source(ref.source_id) + except ArchiveError as exc: + raise ResolveError(f"{ref}: {exc}") from exc + + def _resolve_SnapshotRef(self, ref: SnapshotRef) -> Any: + try: + snapshot = self.archive.get_snapshot(ref.digest, ref.kind) + self.archive.verify_snapshot(snapshot) + except ArchiveError as exc: + raise ResolveError(f"{ref}: {exc}") from exc + return snapshot + + def _record(self, record_id: str, ref: Any) -> ArtifactRecord: + try: + return self.ledger.get(record_id) + except RecordNotFoundError: + raise ResolveError(f"{ref}: no record {record_id} in the ledger") from None + + def _resolve_RecordRef(self, ref: RecordRef) -> ArtifactRecord: + record = self._record(ref.record_id, ref) + allowed = {ArtifactState.SEALED} + if self.allow_invalidated: + allowed.add(ArtifactState.INVALIDATED) + if record.state not in allowed: + raise ResolveError(f"{ref}: record is {record.state.value}; only sealed evidence resolves") + if record.artifact_hash != ref.artifact_hash: + raise ResolveError( + f"{ref}: the record sealed artifact_hash {record.artifact_hash}, not {ref.artifact_hash}" + ) + try: + tree = self.archive.read_tree(record.artifact_hash) + except ArchiveError as exc: + raise ResolveError(f"{ref}: sealed output tree unreadable: {exc}") from exc + if tree != dict(record.artifact_manifest): + raise ResolveError(f"{ref}: the archived output tree does not list the record's manifest") + return record + + def _resolve_ObservationRef(self, ref: ObservationRef) -> Any: + record = self._record(ref.record_id, ref) + try: + observation = observation_of(record) + except ValueError as exc: + raise ResolveError(f"{ref}: {exc}") from exc + if observation.digest != ref.digest: + raise ResolveError( + f"{ref}: the record's observation has digest {observation.digest}, not {ref.digest}" + ) + self._resolve_SnapshotRef(observation.snapshot) + if observation.source_id is not None: + self._resolve_SourceRef(SourceRef(observation.source_id)) + return observation + + def _resolve_AnchorRef(self, ref: AnchorRef) -> str: + record = self._resolve_RecordRef(ref.record) + try: + return BundleResolver(SealedBundle.from_record(record, self.archive)).resolve(ref.anchor) + except ResolutionError as exc: + raise ResolveError(f"{ref}: {exc}") from exc + + def _resolve_EventRef(self, ref: EventRef) -> Any: + log = EventLog(self.ledger) + try: + report = log.verify(anchor=(ref.seq, ref.hash)) + if not report.ok: + raise ResolveError(f"{ref}: {'; '.join(report.problems[:3])}") + event = next(log.events(after=ref.seq - 1), None) + finally: + log.close() + if event is None or (event.seq, event.hash) != (ref.seq, ref.hash): + raise ResolveError(f"{ref}: no such event in the chain") + return event diff --git a/tests/test_identity.py b/tests/test_identity.py new file mode 100644 index 0000000..d17cdca --- /dev/null +++ b/tests/test_identity.py @@ -0,0 +1,342 @@ +""" +Identity contract (roadmap #42): typed, self-verifying references. + + 1. Every reference type round-trips through exactly one canonical string; + every other spelling is refused. + 2. Each type resolves against honest evidence to what it names. + 3. Each type stops resolving once the evidence it committed to changes: + a wrong digest, a tampered blob, an invalidated record, a different + observation, a misquoted anchor, a rewritten event. + 4. Extraction chunks carry anchor references that resolve to their text. + 5. The CLI lists and resolves references. +""" +from __future__ import annotations + +import json +import sqlite3 +import uuid +from pathlib import Path + +import pytest + +from stele.adapters import ExtractionAdapter +from stele.archive import Snapshot, SnapshotKind, Source +from stele.contracts.adapter import SealedBundle +from stele.extraction import Anchor, normalize +from stele.identity import ( + AnchorRef, EventRef, IdentityResolver, Observation, ObservationRef, RecordRef, RefFormatError, + ResolveError, SnapshotRef, SourceRef, anchor_ref, event_ref, observation_of, parse_ref, + record_ref, +) +from stele.identity.__main__ import main as cli +from stele.ledger.events import EventLog +from stele.ledger.models import ParserIdentity +from tests.ledger_helpers import open_ledger + +D1, D2 = "a" * 64, "b" * 64 +RID = "3f2c9a1e-0000-4000-8000-000000000001" + +MARKDOWN = "# Title\n\nFirst paragraph, naïve.\n\n## Part\n\n- item\n".encode("utf-8") + + +def _sealed(tmp_path: Path, *, with_source: bool = True): + """A SEALED Marker-shaped record over an archived input Snapshot.""" + ledger = open_ledger(tmp_path / "ledger.db") + archive = ledger.archive + doc = b"%PDF-1.7 the input document" + digest = archive.put_bytes(doc) + snapshot = archive.put_snapshot(Snapshot(SnapshotKind.FILE, digest, len(doc), 1)) + source = Source(locator="/corpus/report.pdf") if with_source else None + if source is not None: + archive.put_source(source) + out = tmp_path / f"out-{uuid.uuid4().hex[:6]}" + out.mkdir() + (out / "document.md").write_bytes(MARKDOWN) + (out / "stele-parser.json").write_text(json.dumps( + {"parser": "marker", "version": "1.0", "outputs": {"markdown": "document.md"}} + )) + record = ledger.create_pending( + run_id=str(uuid.uuid4()), artifact_dir=out, artifact_paths=sorted(out.iterdir()), + parser=ParserIdentity("marker", "1.0"), parser_config={}, + input_snapshot=snapshot, source=source, + ) + return ledger, ledger.seal(record.record_id) + + +def _extraction(ledger, record): + return normalize(SealedBundle.from_record(ledger.get(record.record_id), ledger.archive)) + + +# --------------------------------------------------------------------------- +# 1 — canonical strings +# --------------------------------------------------------------------------- + +class TestFormat: + + @pytest.mark.parametrize("ref", [ + SourceRef(D1), + SnapshotRef(SnapshotKind.FILE, D1), + SnapshotRef(SnapshotKind.TREE, D2), + RecordRef(RID, D1), + ObservationRef(RID, D2), + EventRef(1, D1), + EventRef(12345, D2), + AnchorRef(RecordRef(RID, D1), Anchor(artifact="document.md", digest=D2, range=(0, 12))), + AnchorRef(RecordRef(RID, D1), Anchor(artifact="dir/a b@#%.md", digest=D2, range=(3, 3))), + AnchorRef(RecordRef(RID, D1), Anchor( + artifact="conversations/c1.json", digest=D2, + pointer="/mapping/a~1b c/message", render="chatgpt-message")), + AnchorRef(RecordRef(RID, D1), Anchor( + artifact="x.json", digest=D2, pointer="", render="json-string")), + AnchorRef(RecordRef(RID, D1), Anchor(artifact="文書.md", digest=D2, range=(0, 1))), + ], ids=str) + def test_round_trip(self, ref) -> None: + text = str(ref) + assert text.startswith("stele:") + assert parse_ref(text) == ref + assert str(parse_ref(text)) == text + + def test_expected_spellings(self) -> None: + assert str(SnapshotRef("file", D1)) == f"stele:snapshot:file:{D1}" + assert str(RecordRef(RID, D1)) == f"stele:record:{RID}@{D1}" + assert str(EventRef(7, D1)) == f"stele:event:7@{D1}" + ref = AnchorRef(RecordRef(RID, D1), Anchor(artifact="a b/c.md", digest=D2, range=(4, 9))) + assert str(ref) == f"stele:anchor:{RID}@{D1}/a%20b/c.md@{D2}#bytes=4-9" + ref = AnchorRef(RecordRef(RID, D1), Anchor( + artifact="c.json", digest=D2, pointer="/mapping/n 1/message", render="chatgpt-message")) + assert str(ref).endswith("#pointer=/mapping/n%201/message&render=chatgpt-message") + + @pytest.mark.parametrize("text", [ + "", + f"stele:thing:{D1}", + f"source:{D1}", + f"stele:source:{D1.upper()}", + f"stele:source:{D1[:-1]}", + f"stele:snapshot:blob:{D1}", + f"stele:snapshot:FILE:{D1}", + f"stele:record:{RID}", + f"stele:record:{RID}@{D1}@{D2}", + f"stele:record:bad/id@{D1}", + f"stele:event:0@{D1}", + f"stele:event:07@{D1}", + f"stele:event:-1@{D1}", + f"stele:anchor:{RID}@{D1}/doc.md@{D2}", # no fragment + f"stele:anchor:{RID}@{D1}/doc.md@{D2}#bytes=9-4", # backwards range + f"stele:anchor:{RID}@{D1}/doc.md@{D2}#bytes=01-4", # leading zero + f"stele:anchor:{RID}@{D1}/doc.md@{D2}#bytes=4", + f"stele:anchor:{RID}@{D1}/doc.md@{D2}#lines=1-2", + f"stele:anchor:{RID}@{D1}/doc.md@{D2}#pointer=/x", # no renderer + f"stele:anchor:{RID}@{D1}/a%2fb.md@{D2}#bytes=0-1", # lower-case / needless escape + f"stele:anchor:{RID}@{D1}/a%2Fb.md@{D2}#bytes=0-1", # "/" must stay literal + f"stele:anchor:{RID}@{D1}/a b.md@{D2}#bytes=0-1", # space must be escaped + f"stele:anchor:{RID}@{D1}/@{D2}#bytes=0-1", # empty artifact + ]) + def test_non_canonical_is_refused(self, text) -> None: + with pytest.raises(RefFormatError): + parse_ref(text) + + def test_anchor_ref_needs_typed_parts(self) -> None: + with pytest.raises(RefFormatError): + AnchorRef(RecordRef(RID, D1), {"artifact": "x", "digest": D2, "range": [0, 1]}) + + def test_observation_digest_covers_every_field(self) -> None: + base = Observation(RID, D1, SnapshotRef("file", D2), "2026-09-25T12:00:00+00:00") + variants = [ + Observation(RID.replace("1", "2"), D1, base.snapshot, base.observed_at), + Observation(RID, None, base.snapshot, base.observed_at), + Observation(RID, D1, SnapshotRef("tree", D2), base.observed_at), + Observation(RID, D1, base.snapshot, "2026-09-25T12:00:01+00:00"), + ] + assert len({base.digest, *(v.digest for v in variants)}) == 5 + assert base.ref == ObservationRef(RID, base.digest) + with pytest.raises(RefFormatError, match="ISO 8601"): + Observation(RID, D1, base.snapshot, "yesterday") + + +# --------------------------------------------------------------------------- +# 2, 3 — resolution against the evidence +# --------------------------------------------------------------------------- + +class TestResolve: + + def test_every_type_resolves_to_what_it_names(self, tmp_path) -> None: + ledger, record = _sealed(tmp_path) + resolver = IdentityResolver(ledger) + observation = observation_of(record) + + assert resolver.resolve(SourceRef(record.source_id)).locator == "/corpus/report.pdf" + assert resolver.resolve(observation.snapshot).digest == record.source_hash + assert resolver.resolve(record_ref(record)).record_id == record.record_id + assert resolver.resolve(str(observation.ref)) == observation + + extraction = _extraction(ledger, record) + for unit in extraction.units: + assert resolver.resolve(str(anchor_ref(extraction, unit))) == unit.text + + log = EventLog(ledger) + events = log.for_subject(record.record_id) + log.close() + assert [e.kind for e in events] == ["record.created", "record.sealed"] + for event in events: + assert resolver.resolve(str(event_ref(event))) == event + + def test_malformed_string_is_a_resolve_error(self, tmp_path) -> None: + ledger, _ = _sealed(tmp_path) + with pytest.raises(ResolveError, match="not a Stele reference"): + IdentityResolver(ledger).resolve("https://example.com") + + def test_wrong_digests_are_refused(self, tmp_path) -> None: + ledger, record = _sealed(tmp_path) + resolver = IdentityResolver(ledger) + extraction = _extraction(ledger, record) + unit = extraction.units[1] + forged = [ + SourceRef(D1), + SnapshotRef("file", D1), + SnapshotRef("tree", record.source_hash), # right digest, wrong kind + RecordRef(record.record_id, D1), + RecordRef(str(uuid.uuid4()), record.artifact_hash), + ObservationRef(record.record_id, D1), + AnchorRef(RecordRef(record.record_id, D1), unit.anchor), + AnchorRef(record_ref(record), Anchor( + artifact=unit.anchor.artifact, digest=D1, range=unit.anchor.range)), + AnchorRef(record_ref(record), Anchor( + artifact="elsewhere.md", digest=unit.anchor.digest, range=unit.anchor.range)), + AnchorRef(record_ref(record), Anchor( + artifact=unit.anchor.artifact, digest=unit.anchor.digest, range=(0, 10_000))), + EventRef(1, D1), + EventRef(999, D1), + ] + for ref in forged: + assert resolver.check(ref), f"{ref} resolved" + + def test_a_different_observation_is_refused(self, tmp_path) -> None: + ledger, record = _sealed(tmp_path) + honest = observation_of(record) + for other in ( + Observation(record.record_id, None, honest.snapshot, honest.observed_at), + Observation(record.record_id, honest.source_id, honest.snapshot, + "2020-01-01T00:00:00+00:00"), + ): + problems = IdentityResolver(ledger).check(other.ref) + assert problems and "observation has digest" 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" + out.mkdir() + (out / "x.txt").write_text("x") + record = ledger.seal(ledger.create_pending( + run_id=str(uuid.uuid4()), artifact_dir=out, artifact_paths=[out / "x.txt"], + parser=ParserIdentity("p", "1"), parser_config={}, + ).record_id) + with pytest.raises(ValueError, match="no input Snapshot"): + observation_of(record) + assert "no input Snapshot" in IdentityResolver(ledger).check( + ObservationRef(record.record_id, D1))[0] + + def test_observation_without_source(self, tmp_path) -> None: + ledger, record = _sealed(tmp_path, with_source=False) + observation = observation_of(record) + assert observation.source_id is None + assert IdentityResolver(ledger).resolve(observation.ref) == observation + + def test_invalidated_record_needs_permission(self, tmp_path) -> None: + ledger, record = _sealed(tmp_path) + extraction = _extraction(ledger, record) + anchor = anchor_ref(extraction, extraction.units[0]) + observation = observation_of(record).ref + ledger.invalidate(record.record_id, "withdrawn") + + strict = IdentityResolver(ledger) + for ref in (record_ref(record), anchor): + assert "invalidated" in strict.check(ref)[0] + # What was observed stays observed. + assert strict.check(observation) == [] + + lenient = IdentityResolver(ledger, allow_invalidated=True) + assert lenient.check(record_ref(record)) == [] + assert lenient.resolve(anchor) == extraction.units[0].text + + def test_tampered_evidence_is_refused(self, tmp_path) -> None: + ledger, record = _sealed(tmp_path) + resolver = IdentityResolver(ledger) + extraction = _extraction(ledger, record) + anchor = anchor_ref(extraction, extraction.units[0]) + observation = observation_of(record) + + def blob(digest: str) -> Path: + path = ledger.archive._blob_path(digest) + path.chmod(0o644) + return path + + # Output blob: the anchor reads verified bytes, so it is refused. + md = blob(record.artifact_manifest["document.md"]) + md.write_bytes(md.read_bytes().replace(b"Title", b"Tit1e")) + assert "unreadable" in resolver.check(anchor)[0] + + # Input blob: the Snapshot, and so the Observation, no longer resolve. + src = blob(record.source_hash) + src.write_bytes(b"%PDF-1.7 another document!!") + assert resolver.check(observation.snapshot) + assert resolver.check(observation.ref) + + def test_rewritten_event_is_refused(self, tmp_path) -> None: + ledger, record = _sealed(tmp_path) + log = EventLog(ledger) + first = log.for_subject(record.record_id)[0] + log.close() + ref = event_ref(first) + assert IdentityResolver(ledger).check(ref) == [] + + conn = sqlite3.connect(ledger.db_path) + conn.execute("DROP TRIGGER events_append_only_u") + conn.execute("UPDATE events SET body=? WHERE seq=?", ('{"forged":true}', first.seq)) + conn.commit() + conn.close() + problems = IdentityResolver(ledger).check(ref) + assert problems and "altered" in problems[0] + + +# --------------------------------------------------------------------------- +# 4 — extraction chunks cite their evidence +# --------------------------------------------------------------------------- + +def test_extraction_chunks_carry_resolvable_anchor_refs(tmp_path) -> None: + ledger, record = _sealed(tmp_path) + bundle = SealedBundle.from_record(ledger.get(record.record_id), ledger.archive) + chunks = ExtractionAdapter().transform(bundle) + resolver = IdentityResolver(ledger) + assert chunks + for chunk in chunks: + ref = parse_ref(chunk.metadata["stele_anchor"]) + assert isinstance(ref, AnchorRef) + assert ref.record == record_ref(record) + assert resolver.resolve(ref) == chunk.content + + +# --------------------------------------------------------------------------- +# 5 — CLI +# --------------------------------------------------------------------------- + +def test_cli_refs_and_resolve(tmp_path, capsys) -> None: + ledger, record = _sealed(tmp_path) + db, archive = str(ledger.db_path), str(ledger.archive.root) + ledger.close() + + assert cli(["refs", db, archive, record.record_id]) == 0 + refs = json.loads(capsys.readouterr().out) + assert refs["record"] == str(record_ref(record)) + assert refs["observation"].startswith(f"stele:observation:{record.record_id}@") + assert len(refs["events"]) == 2 + + good = [refs["record"], refs["observation"], refs["snapshot"], *refs["events"]] + assert cli(["resolve", db, archive, *good]) == 0 + lines = [json.loads(line) for line in capsys.readouterr().out.splitlines()] + assert [line["ok"] for line in lines] == [True] * len(good) + + assert cli(["resolve", db, archive, refs["record"], f"stele:record:{record.record_id}@{D1}"]) == 1 + lines = [json.loads(line) for line in capsys.readouterr().out.splitlines()] + assert [line["ok"] for line in lines] == [True, False] + + assert cli(["refs", db, archive, "no-such-record"]) == 1