Write the record to disk as a bundle a stranger can verify

apps/api/src/cora/infrastructure/record_export/__init__.py

@@ -24,6 +24,15 @@
   re-identify without the token map.
 """
 
+from cora.infrastructure.record_export._bundle import (
+    LOGBOOKS_DIR,
+    MANIFEST_NAME,
+    STREAMS_NAME,
+    BundleDestinationNotEmptyError,
+    MalformedBundleError,
+    read_bundle_body,
+    write_bundle,
+)
 from cora.infrastructure.record_export._export import (
     EmptyExportError,
     ExportedRecord,
@@ -32,11 +41,14 @@
 )
 from cora.infrastructure.record_export._hashing import (
     LOGBOOKS_PAYLOAD_TYPE,
+    PUBLISHED_RECORD_PAYLOAD_TYPE,
     RECORD_PAYLOAD_TYPE,
     REDACTION_PROFILE_PAYLOAD_TYPE,
     STREAMS_PAYLOAD_TYPE,
+    TwoTierRecord,
     hash_logbooks,
     hash_record,
+    hash_redacted_record,
     hash_redaction_profile,
     hash_streams,
 )
@@ -50,9 +62,8 @@
     TIER2_DISPOSITIONS,
     TIER2_JSONB_CLEARED_POINTERS,
     TIER2_JSONB_DROPPED_COLUMNS,
-    UnfiredClearanceError,
-    ensure_all_clearances_fired,
     redact_tier2_row,
+    unfired_clearances,
 )
 from cora.infrastructure.record_export._redaction import (
     RedactedRecord,
@@ -78,43 +89,52 @@
 
 __all__ = [
     "KNOWN_STREAM_TYPES",
+    "LOGBOOKS_DIR",
     "LOGBOOKS_PAYLOAD_TYPE",
+    "MANIFEST_NAME",
+    "PUBLISHED_RECORD_PAYLOAD_TYPE",
     "RECORD_PAYLOAD_TYPE",
     "REDACTION_PROFILE_PAYLOAD_TYPE",
+    "STREAMS_NAME",
     "STREAMS_PAYLOAD_TYPE",
     "TIER2_DISPOSITIONS",
     "TIER2_JSONB_CLEARED_POINTERS",
     "TIER2_JSONB_DROPPED_COLUMNS",
+    "BundleDestinationNotEmptyError",
     "EmptyExportError",
     "EntriesReader",
     "EntriesTableSpec",
     "ExportedRecord",
+    "MalformedBundleError",
     "Manifest",
     "RedactedRecord",
     "RedactionProfileMismatchError",
     "RedactionResult",
     "Tier1Redactor",
     "TokenMap",
-    "UnfiredClearanceError",
+    "TwoTierRecord",
     "UnknownEventTypeError",
     "UnknownLogbookKindError",
     "UnknownStreamTypeError",
     "all_specs",
     "build_manifest",
     "capture_git_commit",
     "capture_watermark",
-    "ensure_all_clearances_fired",
     "ensure_stream_type_known",
     "export_record",
     "hash_logbooks",
     "hash_record",
+    "hash_redacted_record",
     "hash_redaction_profile",
     "hash_streams",
+    "read_bundle_body",
     "redact_record",
     "redact_tier1_payload",
     "redact_tier2_row",
     "registered_envelope_classes",
     "render_row",
     "render_value",
     "resolve",
+    "unfired_clearances",
+    "write_bundle",
 ]

apps/api/src/cora/infrastructure/record_export/_bundle.py

@@ -0,0 +1,185 @@
+"""Write an exported record to disk as the bundle layout F5 names.
+
+`project_record_export_v3.md`'s "Naming" section fixes the layout:
+`manifest.json`, `streams.jsonl`, `logbooks/<kind>.jsonl`. Everything
+upstream of this module produces in-memory structures; this is the step
+that makes an artifact somebody can archive, cite, or email.
+
+## Why JSON Lines, and the trap it carries
+
+One row per line means a reader can stream a large export, `grep` it,
+and see a per-row diff in review. The trap is that the bundle's line
+format is NOT what the hashes cover. `compute_content_hash` hashes a
+DSSE-PAE-wrapped, NFC-normalized, sorted-key serialization of the whole
+BODY, so the on-disk file is a transport for the body, not the hashed
+bytes themselves. Two consequences, both load-bearing:
+
+- A verifier must REASSEMBLE the body (`{"streams": [...],
+  "logbooks": {...}}`) from the files before hashing. `read_bundle_body`
+  is that reassembly, and `scripts/verify_record_hash.py`'s
+  `verify-bundle` subcommand reimplements it in stdlib-only form.
+- Reordering lines, or losing one, changes the hash. That is the point:
+  order is part of the record (F2's per-kind order key), not a
+  presentation detail.
+
+## Refusals
+
+Writing into a non-empty directory refuses. A bundle is an atomic claim
+about one database at one watermark; a directory holding two exports'
+`logbooks/` files, or one export's `streams.jsonl` beside another's
+`manifest.json`, would hash-verify per file and be a lie as a whole.
+"""
+
+import json
+from dataclasses import asdict
+from pathlib import Path
+from typing import cast
+
+from cora.infrastructure.record_export._hashing import TwoTierRecord
+from cora.infrastructure.record_export._manifest import Manifest
+
+MANIFEST_NAME = "manifest.json"
+STREAMS_NAME = "streams.jsonl"
+LOGBOOKS_DIR = "logbooks"
+
+__all__ = [
+    "LOGBOOKS_DIR",
+    "MANIFEST_NAME",
+    "STREAMS_NAME",
+    "BundleDestinationNotEmptyError",
+    "MalformedBundleError",
+    "read_bundle_body",
+    "write_bundle",
+]
+
+
+class BundleDestinationNotEmptyError(RuntimeError):
+    """`write_bundle` was pointed at a directory that already holds files.
+
+    Refuses rather than merging or overwriting: a bundle is one export's
+    whole claim, and a directory mixing two exports would verify
+    file-by-file while being incoherent as a record.
+    """
+
+    def __init__(self, destination: Path) -> None:
+        super().__init__(
+            f"refusing to write a bundle into non-empty directory {destination}: "
+            "a bundle is one export at one watermark. Write to a fresh directory."
+        )
+        self.destination = destination
+
+
+class MalformedBundleError(RuntimeError):
+    """A directory does not hold a readable bundle.
+
+    Raised for a missing `streams.jsonl`, a missing `manifest.json`, or a
+    line that is not a JSON object. Never falls back to a partial read:
+    a body reassembled from half a bundle would hash to something that
+    matches nothing, which reads as tampering rather than as the file
+    error it is.
+    """
+
+
+def _kind_filename(kind: str) -> str:
+    """`logbooks/<kind>.jsonl`, with `kind` rejected if it could escape.
+
+    Registry kinds are code-defined identifiers today, so this cannot
+    currently fire. It is here because the alternative failure is
+    writing outside the destination directory, and a kind reaches this
+    function from a registry entry that a future edit could widen.
+    """
+    if not kind or "/" in kind or "\\" in kind or kind.startswith("."):
+        message = f"logbook kind {kind!r} is not a safe filename component"
+        raise MalformedBundleError(message)
+    return f"{kind}.jsonl"
+
+
+def _write_jsonl(path: Path, rows: tuple[dict[str, object], ...]) -> None:
+    """One compact JSON object per line, in the order given.
+
+    `sort_keys=True` is for reviewability of the file itself, not for
+    the hash: the canonicalizer sorts independently, so a differently
+    ordered file would still hash the same. Written together in one
+    `write_text` so a partial line cannot survive a crash mid-export.
+    """
+    body = "".join(
+        json.dumps(row, sort_keys=True, ensure_ascii=False, separators=(",", ":")) + "\n"
+        for row in rows
+    )
+    path.write_text(body, encoding="utf-8")
+
+
+def _read_jsonl(path: Path) -> tuple[dict[str, object], ...]:
+    rows: list[dict[str, object]] = []
+    for number, line in enumerate(path.read_text(encoding="utf-8").splitlines(), start=1):
+        if not line.strip():
+            continue
+        try:
+            row = json.loads(line)
+        except json.JSONDecodeError as exc:
+            message = f"{path.name} line {number} is not valid JSON: {exc}"
+            raise MalformedBundleError(message) from exc
+        if not isinstance(row, dict):
+            message = f"{path.name} line {number} is a {type(row).__name__}, not a JSON object"
+            raise MalformedBundleError(message)
+        rows.append(cast("dict[str, object]", row))
+    return tuple(rows)
+
+
+def write_bundle(record: TwoTierRecord, manifest: Manifest, destination: Path) -> Path:
+    """Write `record` and `manifest` as a bundle under `destination`.
+
+    Creates the directory if absent; refuses if it exists with anything
+    in it. Returns `destination` so a caller can chain.
+
+    The manifest is written LAST. An interrupted export therefore leaves
+    a directory with no `manifest.json`, which `read_bundle_body`
+    refuses, rather than a complete-looking bundle whose row files are
+    truncated.
+    """
+    destination.mkdir(parents=True, exist_ok=True)
+    existing = sorted(p.name for p in destination.iterdir())
+    if existing:
+        raise BundleDestinationNotEmptyError(destination)
+
+    _write_jsonl(destination / STREAMS_NAME, record.streams)
+
+    logbooks_dir = destination / LOGBOOKS_DIR
+    logbooks_dir.mkdir()
+    for kind, rows in record.logbooks.items():
+        _write_jsonl(logbooks_dir / _kind_filename(kind), rows)
+
+    manifest_path = destination / MANIFEST_NAME
+    manifest_path.write_text(
+        json.dumps(asdict(manifest), sort_keys=True, indent=2, ensure_ascii=False) + "\n",
+        encoding="utf-8",
+    )
+    return destination
+
+
+def read_bundle_body(destination: Path) -> dict[str, object]:
+    """Reassemble the hashed body from a bundle on disk.
+
+    Returns exactly the structure `hash_record` / `hash_redacted_record`
+    hash, so a caller can recompute either and compare. An empty
+    `logbooks/` directory yields `{"logbooks": {}}`, which is a real
+    export shape (a record with no entries rows at all), not an error.
+    """
+    streams_path = destination / STREAMS_NAME
+    if not streams_path.is_file():
+        message = f"{destination} has no {STREAMS_NAME}; not a bundle"
+        raise MalformedBundleError(message)
+    if not (destination / MANIFEST_NAME).is_file():
+        message = (
+            f"{destination} has no {MANIFEST_NAME}. The manifest is written last, "
+            "so this is what an interrupted export leaves behind."
+        )
+        raise MalformedBundleError(message)
+
+    logbooks_dir = destination / LOGBOOKS_DIR
+    logbooks: dict[str, object] = {}
+    if logbooks_dir.is_dir():
+        for path in sorted(logbooks_dir.glob("*.jsonl")):
+            logbooks[path.stem] = list(_read_jsonl(path))
+
+    return {"streams": list(_read_jsonl(streams_path)), "logbooks": logbooks}

apps/api/src/cora/infrastructure/record_export/_hashing.py

@@ -27,6 +27,8 @@
 manifest body in hand rather than guessed at here.
 """
 
+from typing import Protocol
+
 from cora.infrastructure.record_export._dispositions import DISPOSITIONS
 from cora.infrastructure.record_export._export import ExportedRecord
 from cora.infrastructure.record_export._redact_tier2 import (
@@ -40,6 +42,31 @@
 STREAMS_PAYLOAD_TYPE = "application/vnd.cora.record-streams+json"
 LOGBOOKS_PAYLOAD_TYPE = "application/vnd.cora.record-logbooks+json"
 REDACTION_PROFILE_PAYLOAD_TYPE = "application/vnd.cora.record-redaction-profile+json"
+PUBLISHED_RECORD_PAYLOAD_TYPE = "application/vnd.cora.record-published+json"
+
+
+class TwoTierRecord(Protocol):
+    """The shape both `ExportedRecord` and `RedactedRecord` share.
+
+    Structural, not a base class, so `_hashing` can hash a redacted
+    record without importing `_redaction` (which imports this module:
+    the dependency runs one way only).
+    """
+
+    @property
+    def streams(self) -> tuple[dict[str, object], ...]: ...
+
+    @property
+    def logbooks(self) -> dict[str, tuple[dict[str, object], ...]]: ...
+
+
+def _two_tier_body(record: TwoTierRecord) -> dict[str, object]:
+    """The hashed body shape shared by H1 and H3, so the two can never
+    disagree about what "the whole bundle" means."""
+    return {
+        "streams": list(record.streams),
+        "logbooks": {kind: list(rows) for kind, rows in record.logbooks.items()},
+    }
 
 
 def hash_streams(streams: tuple[dict[str, object], ...]) -> str:
@@ -61,16 +88,39 @@ def hash_logbooks(logbooks: dict[str, tuple[dict[str, object], ...]]) -> str:
 def hash_record(record: ExportedRecord) -> str:
     """SHA-256 content hash over the whole bundle, both tiers, no exclusions.
 
-    This is THE record hash: per F2, it covers everything `export_record`
-    produced. Re-running the export against an unchanged database
-    reproduces this value exactly; any single differing byte anywhere in
-    either tier changes it.
+    This is THE record hash (H1): per F2, it covers everything
+    `export_record` produced. Re-running the export against an unchanged
+    database reproduces this value exactly; any single differing byte
+    anywhere in either tier changes it.
     """
-    body = {
-        "streams": list(record.streams),
-        "logbooks": {kind: list(rows) for kind, rows in record.logbooks.items()},
-    }
-    return compute_content_hash(RECORD_PAYLOAD_TYPE, body)
+    return compute_content_hash(RECORD_PAYLOAD_TYPE, _two_tier_body(record))
+
+
+def hash_redacted_record(record: TwoTierRecord) -> str:
+    """SHA-256 content hash over the PUBLISHED record: H3.
+
+    The third of `project_record_export_v3.md` F5's three hashes (full
+    record H1, redaction profile H2, published record H3), and the one
+    the paper actually prints beside its locator, because it is the only
+    one covering the bytes a reader can hold.
+
+    The payload type is deliberately distinct from
+    `RECORD_PAYLOAD_TYPE`. `compute_content_hash` binds the payload type
+    into the DSSE-PAE preamble, so an unredacted and a redacted record
+    that happened to reduce to identical bodies still hash differently.
+    That matters: without it, an export whose redaction dropped nothing
+    (every field already publishable) would produce H1 == H3, and a
+    reader could not tell a published record from a full one by its hash
+    alone. They must never collide.
+
+    Takes the structural `TwoTierRecord` rather than `RedactedRecord`
+    only because importing `_redaction` here would invert this module's
+    dependency; callers pass `RedactionResult.redacted_record`. Passing
+    an unredacted `ExportedRecord` is a caller error this signature
+    cannot catch, and the reason `write_bundle` derives H3 itself from
+    the record it is handed rather than accepting one as a parameter.
+    """
+    return compute_content_hash(PUBLISHED_RECORD_PAYLOAD_TYPE, _two_tier_body(record))
 
 
 def hash_redaction_profile() -> str:
@@ -119,11 +169,14 @@ def hash_redaction_profile() -> str:
 
 __all__ = [
     "LOGBOOKS_PAYLOAD_TYPE",
+    "PUBLISHED_RECORD_PAYLOAD_TYPE",
     "RECORD_PAYLOAD_TYPE",
     "REDACTION_PROFILE_PAYLOAD_TYPE",
     "STREAMS_PAYLOAD_TYPE",
+    "TwoTierRecord",
     "hash_logbooks",
     "hash_record",
+    "hash_redacted_record",
     "hash_redaction_profile",
     "hash_streams",
 ]