fix(storage): attest source mutations across train continuity

polylogue/cli/commands/maintenance/_blob_integrity.py

@@ -93,8 +93,10 @@ def blob_reference_liveness_command(
     }
     if output_format == "json":
         click.echo(json.dumps(payload, indent=2, sort_keys=True))
-        return
-    _render_blob_reference_liveness_plain(report, sample_limit=sample_limit)
+    else:
+        _render_blob_reference_liveness_plain(report, sample_limit=sample_limit)
+    if report.continuity_refresh_pending:
+        raise click.exceptions.Exit(1)
 
 
 def _render_blob_reference_liveness_plain(report: BlobRefLivenessReconciliationReport, *, sample_limit: int) -> None:
@@ -120,6 +122,10 @@ def _render_blob_reference_liveness_plain(report: BlobRefLivenessReconciliationR
         )
     if report.receipt_path is not None:
         click.echo(f"Receipt:      {report.receipt_path}")
+    if report.continuity_refresh_receipt is not None:
+        click.echo(f"Train proof:  {report.continuity_refresh_receipt}")
+    if report.continuity_refresh_error is not None:
+        click.echo(f"Train proof refresh failed: {report.continuity_refresh_error}")
     for candidate in report.classification.candidates[: max(0, sample_limit)]:
         click.echo(
             f"  orphan {candidate.ref_type} ref_id={candidate.ref_id} "

polylogue/storage/blob_ref_liveness.py

@@ -8,8 +8,10 @@
 
 from __future__ import annotations
 
+import hashlib
+import json
 import sqlite3
-from collections.abc import Iterator
+from collections.abc import Iterable, Iterator
 from dataclasses import dataclass
 from typing import cast
 
@@ -48,6 +50,33 @@ def to_dict(self) -> dict[str, object]:
         }
 
 
+class BlobRefLivenessCandidateDigest:
+    """Incrementally encode the canonical candidate stream used by receipts."""
+
+    def __init__(self) -> None:
+        self._digest = hashlib.sha256(b"[")
+        self._first = True
+
+    def update(self, candidate: BlobRefLivenessCandidate) -> None:
+        if not self._first:
+            self._digest.update(b",")
+        self._digest.update(json.dumps(candidate.to_dict(), sort_keys=True, separators=(",", ":")).encode("utf-8"))
+        self._first = False
+
+    def hexdigest(self) -> str:
+        digest = self._digest.copy()
+        digest.update(b"]")
+        return digest.hexdigest()
+
+
+def digest_blob_ref_liveness_candidates(candidates: Iterable[BlobRefLivenessCandidate]) -> str:
+    """Hash candidates with the receipt's canonical JSON array framing."""
+    digest = BlobRefLivenessCandidateDigest()
+    for candidate in candidates:
+        digest.update(candidate)
+    return digest.hexdigest()
+
+
 @dataclass(frozen=True, slots=True)
 class BlobRefLivenessClassification:
     """Complete read-only classification of source-tier blob references."""