v0.8.0: full-codebase audit remediation

Custody: rows+chain commit atomically per batch; audit gains reverse check
(unchained content fails), opens read-only, refuses missing stores. Gate:
tie-averaged Spearman (order-invariant; old impl could flip the gate by
alphabetization), both raters required, peeks labeled + unservable + never
pass. Manifests: corpus_hash covers full records (manifest_version 2).
Embedding cache self-invalidating (text-hash + model@revision key, chunked
reads, one implementation). serve: --config, health survives bad timestamps,
routes generated from artifacts.KINDS. cycle command owns window policy;
compose ships the collector. artifacts.py pins byte-determinism (utf-8,
sorted keys) for every writer. congress.py (dead custody-bypassing draft)
deleted. stats.py + hardening regression suite; 60 tests.

CHANGELOG.md

@@ -8,6 +8,62 @@ requires a version bump and, if it changes methodology, a decision record in
 
 ## [Unreleased]
 
+## [0.8.0] - 2026-07-10
+
+Full-codebase audit (8 finder angles, 42 candidates, 19 verified findings)
+and the refactor to zero. The fingerprinted store is untouched; the
+embeddings cache (derived data) rebuilds itself once.
+
+### Fixed
+
+- Custody integrity: collected rows and their chain entry now commit in ONE
+  transaction per outlet/day batch — no crash or malformed feed entry can
+  leave content outside the chain; malformed entries are skipped, batches
+  roll back whole. `audit` gained the reverse check (content outside the
+  chain fails) and now opens the store strictly read-only, refusing missing
+  paths instead of creating an empty store and passing.
+- Gate math: Spearman now uses tie-averaged ranks (was order-dependent on
+  tied data — could flip the 0.7 gate by alphabetization; verified
+  numerically). Gate requires BOTH raters present and passing; peeking with
+  --allow-unverified is recorded in the artifact, written to an unservable
+  validation-peek-* file, and can never pass.
+- Manifest integrity: corpus_hash now covers every field of every article
+  record (outlet attribution, URLs, byline, timestamps), not just content
+  fingerprints — metadata edits in published manifests are detected
+  (manifest_version 2; v1 manifests refused).
+- Embedding cache is self-invalidating: keyed by hash of the exact embedded
+  text plus model@revision, so pin or recipe changes can never serve stale
+  vectors; one shared implementation for articles and speeches; lookups are
+  chunked (no more whole-table scans that grow with the corpus).
+- /health survives bad timestamps (marks the outlet stale instead of dying);
+  serve gained --config so health scoping and /outlets work from any cwd.
+- Evidence pages: pole story lists are disjoint by construction.
+- Deleted src/tiltmeter/congress.py — an abandoned pre-reference.py draft
+  (swept in by git add -A at v0.2.0) that bypassed the content store and
+  custody chain with unverifiable fingerprints. Never imported; now gone.
+
+### Added
+
+- `tiltmeter cycle`: the deployment unit — ingest, reference top-up, rolling
+  14-day snapshot + run, audit — so window policy and orchestration live in
+  tested Python; compose now ships the collector service with only a sleep
+  loop in shell.
+- artifacts.py: one deterministic writer (UTF-8, sorted keys, no ASCII
+  escaping) and one naming table for every release artifact; the API's
+  routes are generated from it. All writers/readers migrated (three had
+  drifted; several were locale-dependent).
+- stats.py: the tie-averaged Spearman, tiny and textbook-checkable.
+- tests/test_hardening.py: one regression test per audit finding family,
+  plus Dockerfile↔code model-pin consistency and byte-determinism checks.
+
+### Changed
+
+- Dockerfile bakes the model before copying source (code changes no longer
+  re-download it); pins declared as ARGs, drift-gated by test.
+- score.compute returns the full pipeline result; ratings, stories artifact,
+  and evidence pages are one computation by identity, and the sweep shares
+  the same orientation-proxy helper.
+
 ## [0.7.0] - 2026-07-10
 
 Early-development reset (ADR-0005): with a one-day corpus, compatibility debt
@@ -217,3 +273,14 @@ human attention and produce trustworthy corpus + fresh dry-run ratings.
   WSJ politics feed → WSJ world news feed (politics feed dead; Opinion feed
   deliberately not used — news and opinion are rated separately by every
   incumbent rater).
+
+[Unreleased]: https://github.com/sudolulo/tiltmeter/compare/v0.8.0...HEAD
+[0.8.0]: https://github.com/sudolulo/tiltmeter/compare/v0.7.0...v0.8.0
+[0.7.0]: https://github.com/sudolulo/tiltmeter/compare/v0.6.0...v0.7.0
+[0.6.0]: https://github.com/sudolulo/tiltmeter/compare/v0.5.0...v0.6.0
+[0.5.0]: https://github.com/sudolulo/tiltmeter/compare/v0.4.1...v0.5.0
+[0.4.1]: https://github.com/sudolulo/tiltmeter/compare/v0.4.0...v0.4.1
+[0.4.0]: https://github.com/sudolulo/tiltmeter/compare/v0.3.0...v0.4.0
+[0.3.0]: https://github.com/sudolulo/tiltmeter/compare/v0.2.0...v0.3.0
+[0.2.0]: https://github.com/sudolulo/tiltmeter/compare/v0.1.0...v0.2.0
+[0.1.0]: https://github.com/sudolulo/tiltmeter/releases/tag/v0.1.0

Dockerfile

@@ -3,10 +3,9 @@
 # recompute any release fully offline — part of the reproducibility story,
 # not just packaging convenience.
 #
-# Default command serves the read-only API over /app/releases; every other
-# pipeline stage is available as a one-shot command, e.g.:
-#   docker compose run --rm tiltmeter ingest
-#   docker compose run --rm tiltmeter run --manifest releases/manifest-<id>.json
+# Default command serves the read-only API over /app/releases; the collector
+# runs `tiltmeter cycle` on a sleep loop (see compose.yaml) so all
+# orchestration logic lives in tested Python, not deployment shell.
 
 FROM python:3.13-slim
 
@@ -19,16 +18,20 @@ ENV UV_LINK_MODE=copy UV_COMPILE_BYTECODE=1
 COPY pyproject.toml uv.lock README.md ./
 RUN --mount=type=cache,target=/root/.cache/uv uv sync --frozen --no-dev --no-install-project
 
+# bake the pinned model BEFORE copying source, so day-to-day code changes
+# never re-download it. The ARGs must match src/tiltmeter/embed.py — a test
+# (tests/test_docs.py) fails the build pipeline if they drift.
+ARG EMBED_MODEL=sentence-transformers/all-MiniLM-L6-v2
+ARG EMBED_REVISION=c9745ed1d9f207416be6d2e6f8de32d1f16199bf
+ENV HF_HOME=/opt/hf-cache
+RUN /app/.venv/bin/python -c "from sentence_transformers import SentenceTransformer; \
+    SentenceTransformer('${EMBED_MODEL}', revision='${EMBED_REVISION}', device='cpu')" \
+    && chmod -R a+rX /opt/hf-cache
+
 COPY src ./src
 COPY config ./config
 RUN --mount=type=cache,target=/root/.cache/uv uv sync --frozen --no-dev
 
-# bake the pinned model so runs need no network and no HF account; cache must
-# be world-readable because deployments run the pipeline as a non-root uid
-ENV HF_HOME=/opt/hf-cache
-RUN /app/.venv/bin/python -c "from tiltmeter import embed; embed._load_model()" \
-    && chmod -R a+rX /opt/hf-cache
-
 ENV PATH="/app/.venv/bin:$PATH"
 VOLUME ["/app/data", "/app/releases"]
 EXPOSE 8477

METHODOLOGY.md

@@ -160,8 +160,12 @@ label — mitigated by the evidence pages (D10) and the plain-language docs laye
 
 **Decision**: `config/reference_ratings.yaml` holds published AllSides (5-point,
 mapped to −2..+2) and Ad Fontes (numeric) ratings with retrieval dates, used for
-validation only. The v1 gate: Spearman rank correlation ρ ≥ 0.7 against both raters
-over the 20-outlet sample. Every release also publishes a parameter sensitivity
+validation only. The v1 gate: Spearman rank correlation (tie-averaged ranks —
+order-invariant on the 5-point scale's guaranteed ties) ρ ≥ 0.7 against both
+raters over the 20-outlet sample; a rater with no verified values is a missing
+rater and fails the gate outright. Unverified reference values are refused;
+peeking past that is labeled `peek: true`, written to a `validation-peek-*` file
+the public API never serves, and can never pass the gate. Every release also publishes a parameter sensitivity
 sweep: how much ratings move under alternative tunables (clustering threshold,
 embedding model, window length).
 

compose.yaml

@@ -1,15 +1,38 @@
-# API serving + one-shot pipeline commands.
-#   docker compose up -d                                # serve ratings API :8477
-#   docker compose run --rm tiltmeter ingest            # poll feeds once
-#   docker compose run --rm tiltmeter snapshot --start ... --end ...
-#   docker compose run --rm tiltmeter run --manifest releases/manifest-<id>.json
+# The full deployment: read-only API plus the 6-hourly collector.
+#   docker compose up -d                                # api :8477 + collector
+#   docker compose run --rm api ingest                  # any one-shot command
+#   docker compose run --rm api validate --ratings ...  # gate day
+#
+# The collector's only shell is a sleep loop; everything with logic in it —
+# window policy, artifact naming, audit — is `tiltmeter cycle`, tested Python.
+x-common: &common
+  image: tiltmeter:latest
+  build: .
+  user: "568:568"
+  environment:
+    HOME: /tmp
+  volumes:
+    - ./data:/app/data
+    - ./releases:/app/releases
+  restart: unless-stopped
+  logging:
+    driver: json-file
+    options:
+      max-size: 10m
+      max-file: "3"
+
 services:
-  tiltmeter:
-    build: .
-    image: tiltmeter:latest
+  api:
+    <<: *common
+    command: ["serve", "--releases", "releases"]
     ports:
       - "8477:8477"
-    volumes:
-      - ./data:/app/data
-      - ./releases:/app/releases
-    restart: unless-stopped
+  collector:
+    <<: *common
+    entrypoint: ["/bin/sh", "-c"]
+    command:
+      - |
+        while true; do
+          tiltmeter cycle || echo "cycle failed; retrying next interval"
+          sleep 21600
+        done

docs/feasibility-factuality.md

@@ -38,7 +38,7 @@ benchmark for numerical claim verification (arXiv 2403.17169), Full Fact's
 prototype Stats Checker verifying claims against official statistics, and the
 CLEF CheckThat! lab series.
 
-The costs, honestly: (1) **claim extraction requires an NLP model**, which
+The costs, honestly: (1) **finding the checkable claims inside articles requires a language model** (software that reads sentences and pulls out statements like “unemployment fell to 3.9%”), which
 collides with METHODOLOGY D8 — resolvable only via a documented carve-out where
 models may *extract and align* claims but never *evaluate* them, the mechanical
 record-comparison doing all evaluation, with published human-audited extraction

docs/glossary.md

@@ -132,20 +132,3 @@ rebuilt.
 ### Byline
 The author credit on an article, stored exactly as the outlet published it.
 Collected because it cannot be backfilled later; not yet used by any signal.
-
-### Content-addressed storage
-Storing each piece of text under its own fingerprint. The fingerprint is both
-the address and the integrity check: if the text changes, it no longer
-matches its address. Identical texts are stored exactly once.
-
-### Custody chain
-The dataset's tamper-evident logbook. Every batch of newly collected items
-adds a line, and each line is locked to the line before it by a fingerprint.
-Changing or removing any old line breaks every line after it — so the
-collection's history cannot be rewritten quietly. `tiltmeter audit` checks
-the whole chain.
-
-### Append-only
-Data is added, never edited or deleted. Collected articles and speeches are
-append-only in tiltmeter; only derived caches (like embeddings) may be
-rebuilt.

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "tiltmeter"
-version = "0.7.0"
+version = "0.8.0"
 description = "Auditable, reproducible political-lean ratings for news outlets"
 readme = "README.md"
 requires-python = ">=3.13"

releases/custody-head.json

@@ -0,0 +1,10 @@
+{
+ "custody_head": {
+  "entry_hash": "cb604de99fe1c210ebdde49108168d8571f08f4cf456eebf9fd7352700e8462f",
+  "seq": 20,
+  "ts": "2026-07-10T16:57:50.515431+00:00"
+ },
+ "intact": true,
+ "n_contents": 572,
+ "problems": []
+}

src/tiltmeter/__init__.py

@@ -1,7 +1,10 @@
-"""tiltmeter: auditable, reproducible political-lean ratings for news outlets.
+"""What is tiltmeter, and what does it promise?
+
+tiltmeter computes auditable, reproducible political-lean ratings for news
+outlets: open code over an open corpus, every number re-derivable.
 
 Every module in this package opens with a plain-language docstring stating the
 question it answers. See docs/how-it-works.md for the full plain-language tour.
 """
 
-__version__ = "0.7.0"
+__version__ = "0.8.0"

src/tiltmeter/artifacts.py

@@ -0,0 +1,46 @@
+"""How does anything become a published file?
+
+One writer, one reader, one naming rule for everything under releases/.
+Artifacts must be byte-identical wherever they are produced, so the writer
+pins everything a platform could vary: UTF-8 encoding (never the locale),
+sorted keys (never dict order), no ASCII escaping, one trailing newline.
+
+The KINDS table is the single home of the artifact naming convention; the
+API's route table is generated from it, so a new artifact kind becomes
+servable by adding one line here.
+"""
+
+import json
+from pathlib import Path
+
+# kind -> filename prefix; API route name == kind
+KINDS = {
+    "manifests": "manifest",
+    "ratings": "ratings",
+    "stories": "stories",
+    "validation": "validation",
+    "sweeps": "sweep",
+}
+
+
+def artifact_path(out_dir: str | Path, kind: str, snapshot_id: str) -> Path:
+    return Path(out_dir) / f"{KINDS[kind]}-{snapshot_id}.json"
+
+
+def write_json(path: str | Path, payload) -> Path:
+    """Deterministic serialization: same payload, same bytes, any machine."""
+    path = Path(path)
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(
+        json.dumps(payload, indent=1, sort_keys=True, ensure_ascii=False, default=str) + "\n",
+        encoding="utf-8",
+    )
+    return path
+
+
+def write(out_dir: str | Path, kind: str, snapshot_id: str, payload) -> Path:
+    return write_json(artifact_path(out_dir, kind, snapshot_id), payload)
+
+
+def read_json(path: str | Path):
+    return json.loads(Path(path).read_text(encoding="utf-8"))

src/tiltmeter/cli.py

@@ -6,9 +6,13 @@ The command-line interface:
   tiltmeter status     — show how many articles we hold per outlet
   tiltmeter snapshot   — freeze a window of the corpus into a manifest
   tiltmeter reference  — fetch congressional floor speeches (the D5 anchor)
-  tiltmeter run        — manifest → ratings.json + evidence pages
+  tiltmeter run        — manifest → ratings.json + stories + evidence pages
+  tiltmeter cycle      — one full collection cycle: ingest, reference top-up,
+                         rolling snapshot + run, audit (the deployment loop,
+                         so window policy lives in tested code, not in shell)
   tiltmeter validate   — the M3 gate: rank-correlate a release vs the raters
   tiltmeter sweep      — sensitivity: rescore across the threshold grid
+  tiltmeter audit      — verify every content fingerprint + custody chain
   tiltmeter serve      — read-only HTTP API over computed releases
 """
 
@@ -20,6 +24,7 @@ from tiltmeter import db, ingest
 
 DEFAULT_CONFIG = "config/outlets.yaml"
 DEFAULT_DB = "data/tiltmeter.db"
+WINDOW_DAYS = 14  # rolling snapshot window (METHODOLOGY D6/D7 corpus policy)
 
 
 def cmd_ingest(args: argparse.Namespace) -> int:
@@ -55,33 +60,38 @@ def cmd_reference(args: argparse.Namespace) -> int:
         f"  {totals['days']} session days, {totals['speeches']} speeches stored, "
         f"{totals['unmatched']} unmatched speakers dropped, {totals['skipped']} recess days"
     )
-    row = conn.execute(
+    for party, count in conn.execute(
         "SELECT party, COUNT(*) FROM reference_speeches GROUP BY party"
-    ).fetchall()
-    for party, count in row:
+    ).fetchall():
         print(f"  {party}: {count} speeches total")
     return 0
 
 
 def cmd_run(args: argparse.Namespace) -> int:
-    from tiltmeter import __version__, report, score, snapshot
+    from tiltmeter import __version__, artifacts, report, score, snapshot
 
     manifest = snapshot.load(args.manifest)
     conn = db.connect(args.db)
-    ratings = score.compute(conn, manifest, __version__)
-    ratings_path = score.write(ratings, args.out)
-    stories, matrix, articles = score.story_details(conn, manifest)
-    stories_path = score.write_stories(score.stories_json(stories, articles, manifest), args.out)
-    report_dir = report.write(report.render(ratings, stories, matrix, articles), ratings, args.out)
+    result = score.compute(conn, manifest, __version__)
+    ratings_path = artifacts.write(args.out, "ratings", manifest["snapshot_id"], result.ratings)
+    stories_path = artifacts.write(
+        args.out, "stories", manifest["snapshot_id"], score.stories_json(result, manifest)
+    )
+    report_dir = report.write(
+        report.render(result.ratings, result.stories, result.matrix, result.articles),
+        result.ratings,
+        args.out,
+    )
 
     print(f"ratings: {ratings_path}\nstories: {stories_path}\nevidence: {report_dir}/")
-    o = ratings["orientation"]
+    o = result.ratings["orientation"]
     flag = "" if o["reliable"] else "  [UNRELIABLE — do not interpret]"
     print(
-        f"stories: {ratings['n_stories']}, axis inertia {ratings['axis_inertia_share']:.0%}, "
+        f"stories: {result.ratings['n_stories']}, "
+        f"axis inertia {result.ratings['axis_inertia_share']:.0%}, "
         f"orientation rho {o['correlation']:+.2f}{flag}"
     )
-    for entry in ratings["outlets"]:
+    for entry in result.ratings["outlets"]:
         print(
             f"  {entry['score']:+.3f}  [{entry['ci_low']:+.3f} {entry['ci_high']:+.3f}]"
             f"  {entry['outlet']}"
@@ -89,22 +99,74 @@ def cmd_run(args: argparse.Namespace) -> int:
     return 0
 
 
+def cmd_cycle(args: argparse.Namespace) -> int:
+    """One full collection cycle — the unit deployments repeat on a schedule.
+
+    Window policy lives here, in code with tests: a rolling WINDOW_DAYS-day
+    window ending today (exclusive), keyed on observed_at, so each day's
+    re-runs are byte-identical and yesterday's window is complete.
+    """
+    from datetime import date, timedelta
+
+    from tiltmeter import __version__, artifacts, report, score, snapshot
+
+    rc = cmd_ingest(args)
+
+    from tiltmeter import reference
+
+    conn = db.connect(args.db)
+    try:
+        end = date.today()
+        reference.fetch_range(conn, end.isoformat(), args.reference_days, args.congress)
+    except Exception as exc:  # noqa: BLE001 - anchor top-up must not kill collection
+        print(f"  reference top-up failed (non-fatal): {exc}")
+
+    start = (date.today() - timedelta(days=WINDOW_DAYS)).isoformat()
+    end = date.today().isoformat()
+    try:
+        manifest = snapshot.create(conn, start, end, __version__)
+        snapshot.write(manifest, args.out)
+        result = score.compute(conn, manifest, __version__)
+        artifacts.write(args.out, "ratings", manifest["snapshot_id"], result.ratings)
+        artifacts.write(
+            args.out, "stories", manifest["snapshot_id"], score.stories_json(result, manifest)
+        )
+        report.write(
+            report.render(result.ratings, result.stories, result.matrix, result.articles),
+            result.ratings,
+            args.out,
+        )
+        print(f"  rolling release {manifest['snapshot_id']} written")
+    except ValueError as exc:
+        print(f"  no rolling release: {exc}")
+
+    audit_rc = cmd_audit(argparse.Namespace(db=args.db, emit=args.audit_emit))
+    return rc or audit_rc
+
+
 def cmd_validate(args: argparse.Namespace) -> int:
-    import json
     from pathlib import Path
 
-    from tiltmeter import validate
+    from tiltmeter import artifacts, validate
 
-    ratings = json.loads(Path(args.ratings).read_text())
+    ratings = artifacts.read_json(args.ratings)
     reference = validate.load_reference(args.reference, allow_unverified=args.allow_unverified)
     result = validate.report(ratings, reference)
 
-    out = Path(args.ratings).parent / f"validation-{result['snapshot_id']}.json"
-    out.write_text(json.dumps(result, indent=1, sort_keys=True) + "\n")
+    # peeks are labeled inside AND outside: a different filename that the
+    # public API never serves, so a peek cannot masquerade as the gate
+    prefix = "validation-peek" if result["peek"] else "validation"
+    out = Path(args.ratings).parent / f"{prefix}-{result['snapshot_id']}.json"
+    artifacts.write_json(out, result)
 
+    if result["peek"]:
+        print("  PEEK RUN — unverified reference values used; can never pass the gate")
+        print(f"  unverified used: {len(result['unverified_used'])}")
     if result["skipped_unverified"]:
         print(f"  SKIPPED {len(result['skipped_unverified'])} unverified reference entries"
-              " (verify at source or pass --allow-unverified to peek)")
+              " (verify at source to include them)")
+    for rater in result["raters_missing"]:
+        print(f"  {rater:10} MISSING — no verified values; gate cannot pass")
     for rater, r in result["raters"].items():
         mark = "PASS" if r["passes_gate"] else "fail"
         print(f"  {rater:10} rho={r['rho']:+.3f}  n={r['n']}  p={r['permutation_p']}  [{mark}]")
@@ -129,18 +191,32 @@ def cmd_sweep(args: argparse.Namespace) -> int:
 
 
 def cmd_audit(args: argparse.Namespace) -> int:
-    """Full dataset integrity check: content hashes + custody chain."""
-    import json
-    from pathlib import Path
+    """Full dataset integrity check — strictly read-only.
 
-    conn = db.connect(args.db)
-    problems = db.custody_verify(conn) + db.verify_contents(conn)
-    head = db.custody_head(conn)
-    n_contents = conn.execute("SELECT COUNT(*) FROM contents").fetchone()[0]
+    Opens the store in read-only mode and refuses a missing file: an audit
+    that could create an empty store and pass would be worse than no audit.
+    """
+    from tiltmeter import artifacts
+
+    try:
+        conn = db.connect_readonly(args.db)
+    except FileNotFoundError as exc:
+        print(f"  AUDIT FAILED: {exc}")
+        return 1
+    try:
+        problems = db.custody_verify(conn) + db.verify_contents(conn)
+        head = db.custody_head(conn)
+        n_contents = conn.execute("SELECT COUNT(*) FROM contents").fetchone()[0]
+    except db.sqlite3.OperationalError as exc:
+        print(f"  AUDIT FAILED: store unreadable or pre-custody schema ({exc})")
+        return 1
+    finally:
+        conn.close()
     if args.emit:
-        Path(args.emit).write_text(json.dumps(
-            {"custody_head": head, "n_contents": n_contents,
-             "intact": not problems, "problems": problems}, indent=1) + "\n")
+        artifacts.write_json(args.emit, {
+            "custody_head": head, "n_contents": n_contents,
+            "intact": not problems, "problems": problems,
+        })
     print(f"  contents: {n_contents} items, chain head seq {head['seq']}")
     if problems:
         for p in problems[:20]:
@@ -154,7 +230,8 @@ def cmd_audit(args: argparse.Namespace) -> int:
 def cmd_serve(args: argparse.Namespace) -> int:
     from tiltmeter import serve
 
-    serve.run(args.releases, args.host, args.port, db_path=args.db)
+    serve.run(args.releases, args.host, args.port,
+              outlets_config=args.config, db_path=args.db)
     return 0
 
 
@@ -205,12 +282,21 @@ def main(argv: list[str] | None = None) -> int:
     p_run.add_argument("--out", default="releases", help="output directory")
     p_run.set_defaults(func=cmd_run)
 
+    p_cycle = sub.add_parser("cycle", help="one full collection cycle (the deployment unit)")
+    p_cycle.add_argument("--config", default=DEFAULT_CONFIG)
+    p_cycle.add_argument("--out", default="releases")
+    p_cycle.add_argument("--no-text", action="store_true")
+    p_cycle.add_argument("--reference-days", type=int, default=15)
+    p_cycle.add_argument("--congress", type=int, default=119)
+    p_cycle.add_argument("--audit-emit", default="releases/custody-head.json")
+    p_cycle.set_defaults(func=cmd_cycle)
+
     p_validate = sub.add_parser("validate", help="M3 gate: rank-correlate a release vs raters")
     p_validate.add_argument("--ratings", required=True, help="path to a ratings-*.json release")
     p_validate.add_argument("--reference", default="config/reference_ratings.yaml")
     p_validate.add_argument(
         "--allow-unverified", action="store_true",
-        help="include reference values not verified at source (peeking only, never for the gate)",
+        help="peek at unverified reference values; labeled, unservable, can never pass the gate",
     )
     p_validate.set_defaults(func=cmd_validate)
 
@@ -225,6 +311,8 @@ def main(argv: list[str] | None = None) -> int:
 
     p_serve = sub.add_parser("serve", help="read-only HTTP API over computed releases")
     p_serve.add_argument("--releases", default="releases")
+    p_serve.add_argument("--config", default=DEFAULT_CONFIG,
+                         help="outlets config for /outlets and health scoping")
     p_serve.add_argument("--host", default="0.0.0.0")
     p_serve.add_argument("--port", type=int, default=8477)
     p_serve.set_defaults(func=cmd_serve)

src/tiltmeter/congress.py

@@ -1,186 +0,0 @@
-"""What does each party's language actually sound like?
-
-The axis orientation anchor (METHODOLOGY.md D5): floor speeches from the
-Congressional Record, tagged by the speaker's party. Party membership comes
-from voteview.com member data — public records, not ratings.
-
-Sources, both fetchable without keys or accounts:
-- govinfo.gov daily-issue zips: CREC-YYYY-MM-DD.zip (HTM granules inside).
-  Days Congress wasn't in session return an HTML page, not a zip — skipped.
-- voteview HSall members CSV: bioname, state, chamber, party per congress.
-
-Speaker attribution is heuristic (surname headers like "Mr. THUNE." or
-"Ms. DELBENE of Washington.") and deliberately conservative: a speech whose
-speaker can't be matched to exactly one party is dropped. We need bulk party
-language, not a perfect transcript.
-"""
-
-import csv
-import io
-import logging
-import re
-import sqlite3
-import zipfile
-from datetime import date, timedelta
-
-import requests
-
-from tiltmeter import db
-
-log = logging.getLogger("tiltmeter.congress")
-
-CREC_URL = "https://www.govinfo.gov/content/pkg/CREC-{day}.zip"
-MEMBERS_URL = "https://voteview.com/static/data/out/members/HSall_members.csv"
-PARTY_CODES = {"100": "D", "200": "R"}  # others (independents etc.) dropped
-MIN_SPEECH_WORDS = 50  # ignore procedural one-liners
-
-SCHEMA = """
-CREATE TABLE IF NOT EXISTS ref_speeches (
-    id INTEGER PRIMARY KEY,
-    day TEXT NOT NULL,
-    granule TEXT NOT NULL,
-    speaker TEXT NOT NULL,
-    party TEXT NOT NULL,
-    text TEXT NOT NULL,
-    content_hash TEXT NOT NULL UNIQUE
-);
-"""
-
-STATE_ABBREV = {
-    "Alabama": "AL", "Alaska": "AK", "Arizona": "AZ", "Arkansas": "AR",
-    "California": "CA", "Colorado": "CO", "Connecticut": "CT", "Delaware": "DE",
-    "Florida": "FL", "Georgia": "GA", "Hawaii": "HI", "Idaho": "ID",
-    "Illinois": "IL", "Indiana": "IN", "Iowa": "IA", "Kansas": "KS",
-    "Kentucky": "KY", "Louisiana": "LA", "Maine": "ME", "Maryland": "MD",
-    "Massachusetts": "MA", "Michigan": "MI", "Minnesota": "MN", "Mississippi": "MS",
-    "Missouri": "MO", "Montana": "MT", "Nebraska": "NE", "Nevada": "NV",
-    "New Hampshire": "NH", "New Jersey": "NJ", "New Mexico": "NM", "New York": "NY",
-    "North Carolina": "NC", "North Dakota": "ND", "Ohio": "OH", "Oklahoma": "OK",
-    "Oregon": "OR", "Pennsylvania": "PA", "Rhode Island": "RI",
-    "South Carolina": "SC", "South Dakota": "SD", "Tennessee": "TN", "Texas": "TX",
-    "Utah": "UT", "Vermont": "VT", "Virginia": "VA", "Washington": "WA",
-    "West Virginia": "WV", "Wisconsin": "WI", "Wyoming": "WY",
-}
-
-# "  Mr. THUNE." / "  Ms. DELBENE of Washington." / "  Mr. VAN HOLLEN. Mr. President,"
-SPEAKER_RE = re.compile(
-    r"^\s{1,4}(?:Mr|Mrs|Ms|Miss)\.\s+([A-Z][A-Z'\- ]{1,30}?)"
-    r"(?:\s+of\s+([A-Z][a-z]+(?:\s[A-Z][a-z]+)?))?\.\s",
-    re.MULTILINE,
-)
-
-
-def fetch_members(congress: int, session=None) -> dict:
-    """Surname → chamber → set of (party, state) for one congress, from voteview."""
-    http = session or requests
-    text = http.get(MEMBERS_URL, timeout=120).text
-    members: dict[str, dict[str, set[tuple[str, str]]]] = {}
-    for row in csv.DictReader(io.StringIO(text)):
-        if int(row["congress"]) != congress:
-            continue
-        party = PARTY_CODES.get(row["party_code"])
-        if party is None:
-            continue
-        chamber = {"House": "H", "Senate": "S"}.get(row["chamber"])
-        if chamber is None:
-            continue
-        surname = row["bioname"].split(",")[0].strip().upper()
-        members.setdefault(surname, {}).setdefault(chamber, set()).add(
-            (party, row["state_abbrev"])
-        )
-    return members
-
-
-def resolve_party(
-    members: dict, surname: str, chamber: str, state_name: str | None
-) -> str | None:
-    """One unambiguous party for this speaker, or None (dropped)."""
-    candidates = members.get(surname, {}).get(chamber, set())
-    if state_name:
-        abbrev = STATE_ABBREV.get(state_name)
-        candidates = {(p, s) for p, s in candidates if s == abbrev}
-    parties = {p for p, _ in candidates}
-    return parties.pop() if len(parties) == 1 else None
-
-
-def _granule_chamber(name: str) -> str | None:
-    """Floor granules only: PgH = House, PgS = Senate, PgE = Extensions (House)."""
-    m = re.search(r"-Pg([HSE])", name)
-    if m is None:
-        return None
-    return {"H": "H", "S": "S", "E": "H"}[m.group(1)]
-
-
-def split_speeches(granule_text: str) -> list[tuple[str, str | None, str]]:
-    """(surname, state or None, speech text) for each speaker turn in a granule."""
-    text = re.sub(r"<[^>]+>", "", granule_text)  # granule HTM is text in <pre>
-    hits = list(SPEAKER_RE.finditer(text))
-    speeches = []
-    for i, m in enumerate(hits):
-        end = hits[i + 1].start() if i + 1 < len(hits) else len(text)
-        body = " ".join(text[m.end(): end].split())
-        if len(body.split()) >= MIN_SPEECH_WORDS:
-            speeches.append((m.group(1).strip(), m.group(2), body))
-    return speeches
-
-
-def fetch_day(day: str, session=None) -> bytes | None:
-    """One day's CREC zip, or None if Congress wasn't in session."""
-    http = session or requests
-    resp = http.get(CREC_URL.format(day=day), timeout=300, allow_redirects=True)
-    if resp.status_code != 200 or not resp.content.startswith(b"PK"):
-        return None
-    return resp.content
-
-
-def ingest_range(
-    conn: sqlite3.Connection, start: str, end: str, congress: int, session=None
-) -> dict:
-    """Fetch and store party-tagged speeches for [start, end]. Returns counts."""
-    conn.executescript(SCHEMA)
-    members = fetch_members(congress, session=session)
-    counts = {"days_in_session": 0, "speeches": 0, "dropped_ambiguous": 0}
-    day = date.fromisoformat(start)
-    last = date.fromisoformat(end)
-    while day <= last:
-        blob = fetch_day(day.isoformat(), session=session)
-        if blob is not None:
-            counts["days_in_session"] += 1
-            with zipfile.ZipFile(io.BytesIO(blob)) as zf:
-                for name in sorted(zf.namelist()):
-                    chamber = _granule_chamber(name)
-                    if chamber is None or not name.endswith(".htm"):
-                        continue
-                    for surname, state, body in split_speeches(
-                        zf.read(name).decode("utf-8", errors="replace")
-                    ):
-                        party = resolve_party(members, surname, chamber, state)
-                        if party is None:
-                            counts["dropped_ambiguous"] += 1
-                            continue
-                        conn.execute(
-                            "INSERT OR IGNORE INTO ref_speeches"
-                            " (day, granule, speaker, party, text, content_hash)"
-                            " VALUES (?, ?, ?, ?, ?, ?)",
-                            (
-                                day.isoformat(),
-                                name.rsplit("/", 1)[-1],
-                                surname,
-                                party,
-                                body,
-                                db.content_hash(f"{day}|{surname}", body),
-                            ),
-                        )
-                        counts["speeches"] += 1
-            conn.commit()
-            log.info("%s: in session, %d speeches so far", day, counts["speeches"])
-        day += timedelta(days=1)
-    return counts
-
-
-def party_counts(conn: sqlite3.Connection) -> list[tuple[str, int]]:
-    """How much D vs R language do we hold? (Health check.)"""
-    conn.executescript(SCHEMA)
-    return conn.execute(
-        "SELECT party, COUNT(*) FROM ref_speeches GROUP BY party ORDER BY party"
-    ).fetchall()

src/tiltmeter/db.py

@@ -92,14 +92,17 @@ CREATE INDEX IF NOT EXISTS idx_custody_items_seq ON custody_items (seq);
 
 
 def connect(db_path: str | Path) -> sqlite3.Connection:
-    """Open the database, creating the schema if needed. Pre-v3 stores are
-    refused: this is early development and old stores are recollected, not
-    migrated (the migration machinery was deleted with them)."""
-    if isinstance(db_path, str) and db_path != ":memory:" and not db_path.startswith("file:"):
-        Path(db_path).parent.mkdir(parents=True, exist_ok=True)
-    elif isinstance(db_path, Path):
-        db_path.parent.mkdir(parents=True, exist_ok=True)
-    conn = sqlite3.connect(db_path)
+    """Open the store for collection, creating the schema if needed. Pre-v3
+    stores are refused: this is early development and old stores are
+    recollected, not migrated (the migration machinery was deleted with them).
+
+    This call WRITES (schema creation, WAL). Anything that must not alter the
+    store — auditing above all — uses connect_readonly instead.
+    """
+    s = str(db_path)
+    if s != ":memory:":
+        Path(s).parent.mkdir(parents=True, exist_ok=True)
+    conn = sqlite3.connect(s)
     conn.execute("PRAGMA journal_mode=WAL")
     conn.execute("PRAGMA foreign_keys=ON")
     version = conn.execute("PRAGMA user_version").fetchone()[0]
@@ -115,6 +118,18 @@ def connect(db_path: str | Path) -> sqlite3.Connection:
     return conn
 
 
+def connect_readonly(db_path: str | Path) -> sqlite3.Connection:
+    """Open an existing store without the ability to change a single byte.
+
+    Refuses missing files loudly — an audit that silently creates an empty
+    store and passes would be worse than no audit at all.
+    """
+    path = Path(db_path)
+    if not path.is_file():
+        raise FileNotFoundError(f"no store at {path} — refusing to audit nothing")
+    return sqlite3.connect(f"file:{path}?mode=ro", uri=True)
+
+
 def _article_payload(title: str, text: str, summary: str) -> str:
     """The exact byte content an article fingerprint covers: everything we
     captured about the piece — headline, body, and feed summary."""
@@ -140,15 +155,35 @@ def get_content(conn: sqlite3.Connection, chash: str) -> str | None:
     return zlib.decompress(row[0]).decode("utf-8") if row else None
 
 
+def get_contents(conn: sqlite3.Connection, hashes: list[str]) -> dict[str, str]:
+    """Bulk fetch: fingerprint → decompressed payload, chunked queries."""
+    out: dict[str, str] = {}
+    for i in range(0, len(hashes), 400):
+        chunk = hashes[i : i + 400]
+        placeholders = ",".join("?" * len(chunk))
+        for chash, blob in conn.execute(
+            f"SELECT content_hash, text_z FROM contents"
+            f" WHERE content_hash IN ({placeholders})",
+            chunk,
+        ):
+            out[chash] = zlib.decompress(blob).decode("utf-8")
+    return out
+
+
+def split_article_payload(payload: str) -> tuple[str, str, str]:
+    """(title, body, summary) from a fingerprinted article payload."""
+    parts = payload.split("\x1f", 2)
+    while len(parts) < 3:
+        parts.append("")
+    return parts[0], parts[1], parts[2]
+
+
 def get_article_content(conn: sqlite3.Connection, chash: str) -> tuple[str, str, str] | None:
     """(title, body, summary) for an article fingerprint, or None."""
     payload = get_content(conn, chash)
     if payload is None:
         return None
-    parts = payload.split("\x1f", 2)
-    while len(parts) < 3:
-        parts.append("")
-    return parts[0], parts[1], parts[2]
+    return split_article_payload(payload)
 
 
 def outlet_id(conn: sqlite3.Connection, name: str, first_seen: str | None = None) -> int:
@@ -256,7 +291,12 @@ def custody_head(conn: sqlite3.Connection) -> dict:
 
 def custody_append(conn: sqlite3.Connection, kind: str, new_hashes: list[str]) -> dict | None:
     """Chain one batch of newly collected fingerprints. Empty batches are not
-    recorded — the chain logs data arrival, not polling."""
+    recorded — the chain logs data arrival, not polling.
+
+    Does NOT commit: the caller commits chain entry and collected rows in ONE
+    transaction, so no crash window exists in which rows are durable but
+    unchained. Collectors must never commit collected rows separately.
+    """
     if not new_hashes:
         return None
     head = custody_head(conn)
@@ -273,7 +313,6 @@ def custody_append(conn: sqlite3.Connection, kind: str, new_hashes: list[str]) -
         "INSERT INTO custody_items (seq, content_hash) VALUES (?, ?)",
         [(seq, h) for h in new_hashes],
     )
-    conn.commit()
     return {"seq": seq, "entry_hash": entry, "n_items": len(new_hashes)}
 
 
@@ -301,6 +340,13 @@ def verify_contents(conn: sqlite3.Connection) -> list[str]:
         " WHERE content_hash NOT IN (SELECT content_hash FROM contents)"
     ):
         problems.append(f"custody-chained content deleted: {chash[:12]}…")
+    # the reverse direction: collected content that no chain entry covers —
+    # either an interrupted collector (a bug) or rows inserted around the chain
+    for (chash,) in conn.execute(
+        "SELECT content_hash FROM contents"
+        " WHERE content_hash NOT IN (SELECT content_hash FROM custody_items)"
+    ):
+        problems.append(f"content outside the custody chain: {chash[:12]}…")
     return problems
 
 

src/tiltmeter/embed.py

@@ -1,15 +1,18 @@
-"""How does the pipeline see that two articles are about the same story?
+"""How does the pipeline see that two texts are alike?
 
-Each article's headline + opening text is turned into an embedding — a vector
-of numbers where similar texts land close together. Embeddings are used ONLY
-to group similar articles (and to compare poles of the axis to congressional
-language); they never judge anything (METHODOLOGY.md D8).
+Texts are turned into embeddings — vectors where similar texts land close
+together. Embeddings only ever *group* text (story clustering, axis
+orientation); they never judge anything (METHODOLOGY.md D8).
 
-The model is pinned by name and revision and runs on CPU so that anyone,
-anywhere, reproduces the same vectors. Vectors are cached in SQLite keyed by
-content fingerprint: an article's embedding is computed once, ever.
+One cache, one rule: a vector is stored under the SHA-256 of the exact text
+that was embedded, plus the exact model name@revision that embedded it.
+Change the passage recipe, the model, or the pinned revision and every key
+changes with it — the cache cannot serve a stale or mismatched vector, by
+construction. Lookups are chunked queries proportional to the request, never
+scans of the whole cache.
 """
 
+import hashlib
 import sqlite3
 
 import numpy as np
@@ -19,25 +22,26 @@ MODEL_NAME = "sentence-transformers/all-MiniLM-L6-v2"
 MODEL_REVISION = "c9745ed1d9f207416be6d2e6f8de32d1f16199bf"
 LEDE_WORDS = 40  # headline + this many words of body ≈ headline + lede
 
+# The cache key's model column carries the full pin: bumping the revision can
+# never silently reuse old vectors.
+CACHE_MODEL_KEY = f"{MODEL_NAME}@{MODEL_REVISION}"
+_CHUNK = 400  # SQLite bound-parameter comfort zone
+
 _model = None  # loaded lazily: importing torch takes seconds and tests may not need it
 
 CACHE_SCHEMA = """
 CREATE TABLE IF NOT EXISTS embeddings (
-    content_hash TEXT NOT NULL,
+    text_hash TEXT NOT NULL,
     model TEXT NOT NULL,
     vector BLOB NOT NULL,
-    PRIMARY KEY (content_hash, model)
+    PRIMARY KEY (text_hash, model)
 );
 """
 
 
 def passage(title: str, text: str | None, summary: str | None) -> str:
-    """The text we embed: headline plus lede (or feed summary as fallback).
-
-    Everything read here is inside the fingerprinted payload, so two articles
-    with identical passages always share a fingerprint and vice versa — the
-    embedding cache can never serve the wrong vector.
-    """
+    """The text we embed for an article: headline plus lede (or feed summary
+    as fallback). Everything read here comes from the fingerprinted payload."""
     body = text or summary or ""
     lede = " ".join(body.split()[:LEDE_WORDS])
     return f"{title}. {lede}".strip()
@@ -60,41 +64,65 @@ def embed_texts(texts: list[str]) -> np.ndarray:
     ).astype(np.float32)
 
 
-def embed_hashes(conn: sqlite3.Connection, content_hashes: list[str]) -> np.ndarray:
-    """Embeddings for articles named by fingerprint, using/filling the cache.
+def cached_embed(conn: sqlite3.Connection, texts: list[str]) -> np.ndarray:
+    """Vectors for exact texts, using/filling the cache. Row order preserved.
 
-    Passage text (headline + lede) comes from the local articles table — the
-    manifest deliberately carries no text (METHODOLOGY.md D9), so embedding a
-    published manifest requires the locally collected corpus behind it.
-    Row order follows content_hashes.
+    The single entry point for every cached embedding in the pipeline —
+    articles and speeches alike — so the cache invariants live in one place.
+    The cache is derived data (rebuildable, custody-exempt): a legacy-shaped
+    table is dropped and rebuilt rather than migrated.
     """
+    legacy = conn.execute(
+        "SELECT 1 FROM pragma_table_info('embeddings') WHERE name = 'content_hash'"
+    ).fetchone()
+    if legacy:
+        conn.execute("DROP TABLE embeddings")
+        conn.commit()
     conn.executescript(CACHE_SCHEMA)
-    cached: dict[str, np.ndarray] = {}
-    for row in conn.execute(
-        "SELECT content_hash, vector FROM embeddings WHERE model = ?", (MODEL_NAME,)
-    ):
-        cached[row[0]] = np.frombuffer(row[1], dtype=np.float32)
-
-    missing = [h for h in content_hashes if h not in cached]
+    keys = [hashlib.sha256(t.encode("utf-8")).hexdigest() for t in texts]
+    found: dict[str, np.ndarray] = {}
+    unique = list(dict.fromkeys(keys))
+    for i in range(0, len(unique), _CHUNK):
+        chunk = unique[i : i + _CHUNK]
+        placeholders = ",".join("?" * len(chunk))
+        for row in conn.execute(
+            f"SELECT text_hash, vector FROM embeddings"
+            f" WHERE model = ? AND text_hash IN ({placeholders})",
+            [CACHE_MODEL_KEY, *chunk],
+        ):
+            found[row[0]] = np.frombuffer(row[1], dtype=np.float32)
+    by_key = dict(zip(keys, texts))
+    missing = [k for k in unique if k not in found]
     if missing:
-        from tiltmeter import db as tdb
-
-        found = {}
-        for chash in missing:
-            content = tdb.get_article_content(conn, chash)
-            if content is not None:
-                found[chash] = passage(content[0], content[1], content[2])
-        absent = [h for h in missing if h not in found]
-        if absent:
-            raise ValueError(
-                f"{len(absent)} manifest articles not in local corpus (first: {absent[0]})"
-            )
-        order = [h for h in missing]
-        vectors = embed_texts([found[h] for h in order])
+        vectors = embed_texts([by_key[k] for k in missing])
         conn.executemany(
-            "INSERT OR IGNORE INTO embeddings (content_hash, model, vector) VALUES (?, ?, ?)",
-            [(h, MODEL_NAME, v.tobytes()) for h, v in zip(order, vectors)],
+            "INSERT OR IGNORE INTO embeddings (text_hash, model, vector) VALUES (?, ?, ?)",
+            [(k, CACHE_MODEL_KEY, v.tobytes()) for k, v in zip(missing, vectors)],
         )
         conn.commit()
-        cached.update(dict(zip(order, vectors)))
-    return np.stack([cached[h] for h in content_hashes])
+        found.update(dict(zip(missing, vectors)))
+    return np.stack([found[k] for k in keys])
+
+
+def embed_hashes(conn: sqlite3.Connection, content_hashes: list[str]) -> np.ndarray:
+    """Embeddings for articles named by content fingerprint, in order.
+
+    Passage text is rebuilt from the fingerprinted payload in the local
+    contents table — the manifest deliberately carries no text (D9) — then
+    embedded via the shared cache.
+    """
+    from tiltmeter import db
+
+    payloads = db.get_contents(conn, list(dict.fromkeys(content_hashes)))
+    absent = [h for h in dict.fromkeys(content_hashes) if h not in payloads]
+    if absent:
+        raise ValueError(
+            f"{len(absent)} manifest articles not in local corpus (first: {absent[0]})"
+        )
+    passages = {}
+    for chash, payload in payloads.items():
+        title, text, summary = db.split_article_payload(payload)
+        passages[chash] = passage(title, text, summary)
+    vectors = cached_embed(conn, [passages[h] for h in dict.fromkeys(content_hashes)])
+    by_hash = dict(zip(dict.fromkeys(content_hashes), vectors))
+    return np.stack([by_hash[h] for h in content_hashes])

src/tiltmeter/ingest.py

@@ -82,61 +82,71 @@ def fetch_article_text(url: str) -> str | None:
 
 
 def ingest_outlet(conn, outlet: dict, *, fetch_text: bool = True) -> tuple[int, list[str]]:
-    """Poll one outlet's feed; store unseen articles. Returns (seen, new hashes)."""
+    """Poll one outlet's feed; store unseen articles and chain them, in one
+    transaction. Returns (seen, new hashes).
+
+    Rows and their custody entry become durable together or not at all: no
+    commit happens between insert and chain, so no crash or bad feed entry
+    can leave collected content outside the chain. A malformed entry is
+    skipped, never allowed to abort the batch.
+    """
     parsed = feedparser.parse(outlet["feed"], agent=USER_AGENT)
     now = datetime.now(timezone.utc).isoformat()
     seen, new_hashes = 0, []
     for entry in parsed.entries:
-        raw_url = entry.get("link")
-        title = (entry.get("title") or "").strip()
-        if not raw_url or not title:
-            continue
-        seen += 1
-        url = canonical_url(raw_url)
-        if db.have_url(conn, url):
-            continue
-        text = None
-        if fetch_text:
-            try:
-                text = fetch_article_text(raw_url)
-            except Exception as exc:  # noqa: BLE001 - one bad page must not stop the run
-                log.warning("text fetch failed for %s: %s", raw_url, exc)
-            time.sleep(FETCH_DELAY_SECONDS)
-        chash = db.insert_article(
-            conn,
-            outlet=outlet["name"],
-            url=url,
-            url_original=raw_url if raw_url != url else None,
-            title=title,
-            byline=(entry.get("author") or "").strip() or None,
-            published=entry.get("published") or entry.get("updated"),
-            fetched_at=now,
-            summary=strip_html(entry.get("summary")),
-            text=text,
-        )
-        if chash:
-            new_hashes.append(chash)
+        try:
+            raw_url = entry.get("link")
+            title = (entry.get("title") or "").strip()
+            if not raw_url or not title:
+                continue
+            seen += 1
+            url = canonical_url(raw_url)
+            if db.have_url(conn, url):
+                continue
+            text = None
+            if fetch_text:
+                try:
+                    text = fetch_article_text(raw_url)
+                except Exception as exc:  # noqa: BLE001 - one bad page must not stop the run
+                    log.warning("text fetch failed for %s: %s", raw_url, exc)
+                time.sleep(FETCH_DELAY_SECONDS)
+            chash = db.insert_article(
+                conn,
+                outlet=outlet["name"],
+                url=url,
+                url_original=raw_url if raw_url != url else None,
+                title=title,
+                byline=(entry.get("author") or "").strip() or None,
+                published=entry.get("published") or entry.get("updated"),
+                fetched_at=now,
+                summary=strip_html(entry.get("summary")),
+                text=text,
+            )
+            if chash:
+                new_hashes.append(chash)
+        except Exception as exc:  # noqa: BLE001 - one malformed entry must not orphan a batch
+            log.warning("%s: skipping malformed feed entry: %s", outlet["name"], exc)
+    entry = db.custody_append(conn, "ingest", new_hashes)
     conn.commit()
+    if entry:
+        log.info("%s: custody seq %d chains %d items", outlet["name"], entry["seq"],
+                 entry["n_items"])
     return seen, new_hashes
 
 
 def ingest_all(config_path: str, db_path: str, *, fetch_text: bool = True) -> list[dict]:
-    """Poll every configured outlet once; chain the batch into the custody
-    log. Returns a per-outlet result report."""
+    """Poll every configured outlet once; each outlet's articles are chained
+    and committed atomically. Returns a per-outlet result report."""
     conn = db.connect(db_path)
     results = []
-    run_hashes: list[str] = []
     for outlet in load_outlets(config_path):
         try:
             seen, new_hashes = ingest_outlet(conn, outlet, fetch_text=fetch_text)
-            run_hashes.extend(new_hashes)
             results.append({"outlet": outlet["name"], "seen": seen, "new": len(new_hashes)})
             log.info("%s: %d entries in feed, %d new", outlet["name"], seen, len(new_hashes))
         except Exception as exc:  # noqa: BLE001 - one bad feed must not stop the run
+            conn.rollback()  # nothing half-collected may leak into the next batch
             results.append({"outlet": outlet["name"], "error": str(exc)})
             log.error("%s: feed failed: %s", outlet["name"], exc)
-    entry = db.custody_append(conn, "ingest", run_hashes)
-    if entry:
-        log.info("custody: seq %d chains %d new items", entry["seq"], entry["n_items"])
     conn.close()
     return results

src/tiltmeter/orient.py

@@ -13,12 +13,13 @@ diagnostic: a weak agreement means orientation (and possibly the axis
 itself) shouldn't be trusted, and the output says so rather than hiding it.
 """
 
-import hashlib
 import sqlite3
 from dataclasses import dataclass
 
 import numpy as np
 
+from tiltmeter.stats import spearman
+
 SPEECH_EMBED_WORDS = 200  # MiniLM reads ~256 tokens; the opening covers the topic
 MIN_ABS_CORRELATION = 0.3  # below this, orientation is flagged unreliable
 
@@ -31,52 +32,16 @@ class Orientation:
     proxy_by_outlet: tuple[float, ...]  # cos-to-R minus cos-to-D per outlet
 
 
-def _spearman(a: np.ndarray, b: np.ndarray) -> float:
-    ra = np.argsort(np.argsort(a)).astype(np.float64)
-    rb = np.argsort(np.argsort(b)).astype(np.float64)
-    ra -= ra.mean()
-    rb -= rb.mean()
-    denom = np.sqrt((ra**2).sum() * (rb**2).sum())
-    return float((ra * rb).sum() / denom) if denom > 0 else 0.0
-
-
-def _cached_embed_texts(conn: sqlite3.Connection, texts: list[str]) -> np.ndarray:
-    """Embed with the same fingerprint-keyed cache articles use."""
-    from tiltmeter import embed
-
-    hashes = [hashlib.sha256(t.encode()).hexdigest() for t in texts]
-    conn.executescript(embed.CACHE_SCHEMA)
-    cached = {}
-    for h in hashes:
-        row = conn.execute(
-            "SELECT vector FROM embeddings WHERE content_hash = ? AND model = ?",
-            (h, embed.MODEL_NAME),
-        ).fetchone()
-        if row:
-            cached[h] = np.frombuffer(row[0], dtype=np.float32)
-    missing = [(h, t) for h, t in zip(hashes, texts) if h not in cached]
-    if missing:
-        vectors = embed.embed_texts([t for _, t in missing])
-        conn.executemany(
-            "INSERT OR IGNORE INTO embeddings (content_hash, model, vector) VALUES (?, ?, ?)",
-            [(h, embed.MODEL_NAME, v.tobytes()) for (h, _), v in zip(missing, vectors)],
-        )
-        conn.commit()
-        cached.update({h: v for (h, _), v in zip(missing, vectors)})
-    return np.stack([cached[h] for h in hashes])
-
-
 def party_means(conn: sqlite3.Connection) -> dict[str, np.ndarray]:
     """Average embedding of each party's floor speeches (unit-normalized)."""
     from tiltmeter import db as tdb
+    from tiltmeter.embed import cached_embed
 
-    rows = [
-        (party, tdb.get_content(conn, chash))
-        for party, chash in conn.execute(
-            "SELECT party, content_hash FROM reference_speeches"
-        ).fetchall()
-    ]
-    rows = [(p, t) for p, t in rows if t]
+    pairs = conn.execute(
+        "SELECT party, content_hash FROM reference_speeches"
+    ).fetchall()
+    payloads = tdb.get_contents(conn, [chash for _, chash in pairs])
+    rows = [(p, payloads[c]) for p, c in pairs if c in payloads]
     if not rows:
         raise ValueError("no reference speeches; run: tiltmeter reference")
     means = {}
@@ -88,7 +53,7 @@ def party_means(conn: sqlite3.Connection) -> dict[str, np.ndarray]:
         ]
         if len(texts) < 20:
             raise ValueError(f"only {len(texts)} {party} speeches; reference corpus too thin")
-        mean = _cached_embed_texts(conn, texts).mean(axis=0)
+        mean = cached_embed(conn, texts).mean(axis=0)
         means[party] = mean / np.linalg.norm(mean)
     return means
 
@@ -106,7 +71,7 @@ def outlet_proxy(
 
 def orient_sign(axis_positions: list[float], proxy_values: list[float]) -> Orientation:
     """Pure decision: flip the axis if it anti-correlates with party language."""
-    rho = _spearman(np.asarray(axis_positions), np.asarray(proxy_values))
+    rho = spearman(np.asarray(axis_positions), np.asarray(proxy_values))
     return Orientation(
         sign=-1 if rho < 0 else 1,
         correlation=rho,

src/tiltmeter/reference.py

@@ -161,7 +161,11 @@ def ingest_day(conn: sqlite3.Connection, day: str, zip_bytes: bytes, members: di
                 if chash:
                     counts["speeches"] += 1
                     counts["hashes"].append(chash)
+    # rows + chain entry become durable together: same rule as ingest
+    entry = db.custody_append(conn, "reference", counts["hashes"])
     conn.commit()
+    if entry:
+        log.info("%s: custody seq %d chains %d speeches", day, entry["seq"], entry["n_items"])
     return counts
 
 
@@ -170,7 +174,6 @@ def fetch_range(conn: sqlite3.Connection, end_day: str, session_days: int, congr
     members = load_members(congress)
     http = requests.Session()
     totals = {"days": 0, "speeches": 0, "unmatched": 0, "skipped": 0}
-    run_hashes: list[str] = []
     cursor = date.fromisoformat(end_day)
     attempts = 0
     while totals["days"] < session_days and attempts < session_days * 5:
@@ -188,13 +191,14 @@ def fetch_range(conn: sqlite3.Connection, end_day: str, session_days: int, congr
             totals["skipped"] += 1
             log.info("%s: no Record (recess/weekend)", day)
             continue
-        counts = ingest_day(conn, day, zip_bytes, members)
+        try:
+            counts = ingest_day(conn, day, zip_bytes, members)
+        except Exception as exc:  # noqa: BLE001 - one bad day must not poison the next batch
+            conn.rollback()
+            log.error("%s: day failed, rolled back: %s", day, exc)
+            continue
         totals["days"] += 1
         totals["speeches"] += counts["speeches"]
         totals["unmatched"] += counts["unmatched"]
-        run_hashes.extend(counts["hashes"])
         log.info("%s: %d speeches (%d unmatched)", day, counts["speeches"], counts["unmatched"])
-    entry = db.custody_append(conn, "reference", run_hashes)
-    if entry:
-        log.info("custody: seq %d chains %d new items", entry["seq"], entry["n_items"])
     return totals

src/tiltmeter/report.py

@@ -11,14 +11,14 @@ from pathlib import Path
 
 import numpy as np
 
-from tiltmeter.signals.selection import _first_axis  # same axis, story side
+from tiltmeter.signals.selection import first_axis  # same axis, story side
 
 TOP_STORIES = 8
 
 
 def _story_axis_coords(matrix: np.ndarray) -> np.ndarray:
     """Story (column) positions on the same axis the outlets were scaled on."""
-    return _first_axis(matrix.T)
+    return first_axis(matrix.T)
 
 
 def render(ratings: dict, stories: list, matrix: np.ndarray, articles: list) -> dict[str, str]:
@@ -65,9 +65,11 @@ def render(ratings: dict, stories: list, matrix: np.ndarray, articles: list) ->
             "and skipped among them is what placed it where it is.",
             "",
         ]
+        # disjoint by construction: never let one story argue for both poles
+        k = min(TOP_STORIES, len(ranked_stories) // 2)
         for label, indices in (
-            ("Left-pole stories", ranked_stories[:TOP_STORIES]),
-            ("Right-pole stories", ranked_stories[::-1][:TOP_STORIES]),
+            ("Left-pole stories", ranked_stories[:k]),
+            ("Right-pole stories", ranked_stories[::-1][:k]),
         ):
             lines.append(f"## {label}")
             lines.append("")
@@ -101,5 +103,5 @@ def write(pages: dict[str, str], ratings: dict, out_dir: str | Path) -> Path:
     directory = Path(out_dir) / f"report-{ratings['snapshot_id']}"
     directory.mkdir(parents=True, exist_ok=True)
     for filename, content in pages.items():
-        (directory / filename).write_text(content)
+        (directory / filename).write_text(content, encoding="utf-8")
     return directory

src/tiltmeter/score.py

@@ -2,19 +2,22 @@
 
 The assembly line, end to end: manifest → embeddings → story clusters →
 coverage matrix → selection axis with confidence intervals → orientation by
-congressional language → one deterministic JSON file. No timestamps, no
-randomness outside the fixed bootstrap seed: rerunning on the same snapshot
-must produce the same bytes (METHODOLOGY.md D1, D10).
+congressional language → one deterministic JSON file. Everything is computed
+exactly once per run — the scores, the evidence pages, and the stories
+artifact all describe the same clustering because they are handed the same
+objects, never a recomputation.
+
+No timestamps, no randomness outside the fixed bootstrap seed: rerunning on
+the same snapshot must produce the same bytes (METHODOLOGY.md D1, D10).
 """
 
-import json
 import sqlite3
-from pathlib import Path
+from dataclasses import dataclass
 
 import numpy as np
 
 from tiltmeter import embed, orient
-from tiltmeter.cluster import cluster_articles, coverage_matrix
+from tiltmeter.cluster import Story, cluster_articles, coverage_matrix
 from tiltmeter.signals import selection
 
 RATINGS_SCHEMA_VERSION = 1
@@ -24,8 +27,29 @@ REFERENCE_FRAME = (
 )
 
 
-def compute(conn: sqlite3.Connection, manifest: dict, pipeline_version: str) -> dict:
-    """Run the full pipeline on a loaded manifest; return the ratings dict."""
+@dataclass(frozen=True)
+class PipelineResult:
+    """One run's complete output: the ratings and the objects behind them."""
+
+    ratings: dict
+    stories: list[Story]
+    matrix: np.ndarray
+    articles: list[dict]
+
+
+def outlet_mean_vectors(
+    articles: list[dict], vectors: np.ndarray, outlet_order: list[str]
+) -> dict[str, np.ndarray]:
+    """Each outlet's average article embedding — the orientation proxy input,
+    shared by scoring and the sensitivity sweep so they can never diverge."""
+    return {
+        name: vectors[[i for i, a in enumerate(articles) if a["outlet"] == name]].mean(axis=0)
+        for name in outlet_order
+    }
+
+
+def compute(conn: sqlite3.Connection, manifest: dict, pipeline_version: str) -> PipelineResult:
+    """Run the full pipeline on a loaded manifest, once."""
     articles = manifest["articles"]
     outlet_order = manifest["outlets"]
 
@@ -34,12 +58,8 @@ def compute(conn: sqlite3.Connection, manifest: dict, pipeline_version: str) ->
     matrix = coverage_matrix(stories, outlet_order)
     axis = selection.compute(matrix, outlet_order)
 
-    outlet_vectors = {}
-    for name in outlet_order:
-        rows = [i for i, a in enumerate(articles) if a["outlet"] == name]
-        outlet_vectors[name] = vectors[rows].mean(axis=0)
     party = orient.party_means(conn)
-    proxy = orient.outlet_proxy(outlet_vectors, party)
+    proxy = orient.outlet_proxy(outlet_mean_vectors(articles, vectors, outlet_order), party)
     orientation = orient.orient_sign(
         list(axis.positions), [proxy[name] for name in axis.outlets]
     )
@@ -60,7 +80,7 @@ def compute(conn: sqlite3.Connection, manifest: dict, pipeline_version: str) ->
     ]
     outlets_out.sort(key=lambda o: o["score"])
 
-    return {
+    ratings = {
         "schema_version": RATINGS_SCHEMA_VERSION,
         "pipeline_version": pipeline_version,
         "snapshot_id": manifest["snapshot_id"],
@@ -76,28 +96,12 @@ def compute(conn: sqlite3.Connection, manifest: dict, pipeline_version: str) ->
         },
         "outlets": outlets_out,
     }
+    return PipelineResult(ratings=ratings, stories=stories, matrix=matrix, articles=articles)
 
 
-def write(ratings: dict, out_dir: str | Path) -> Path:
-    """Deterministic serialization: same ratings dict, same bytes."""
-    path = Path(out_dir) / f"ratings-{ratings['snapshot_id']}.json"
-    path.parent.mkdir(parents=True, exist_ok=True)
-    path.write_text(json.dumps(ratings, indent=1, sort_keys=True, ensure_ascii=False) + "\n")
-    return path
-
-
-def story_details(conn: sqlite3.Connection, manifest: dict) -> tuple[list, np.ndarray, list]:
-    """Recompute stories + matrix + story axis coords for evidence pages."""
-    articles = manifest["articles"]
-    vectors = embed.embed_hashes(conn, [a["content_hash"] for a in articles])
-    stories = cluster_articles(vectors, [a["outlet"] for a in articles])
-    matrix = coverage_matrix(stories, manifest["outlets"])
-    return stories, matrix, articles
-
-
-def stories_json(stories: list, articles: list, manifest: dict) -> dict:
+def stories_json(result: PipelineResult, manifest: dict) -> dict:
     """The side-by-side primitive for consumers: who covered each story, how
-    each headlined it. Deterministic; same clusters the scores were built on."""
+    each headlined it. Same clusters the scores were built on, by identity."""
     return {
         "schema_version": RATINGS_SCHEMA_VERSION,
         "snapshot_id": manifest["snapshot_id"],
@@ -108,21 +112,14 @@ def stories_json(stories: list, articles: list, manifest: dict) -> dict:
                 "n_outlets": len(s.outlets),
                 "articles": [
                     {
-                        "outlet": articles[i]["outlet"],
-                        "title": articles[i]["title"],
-                        "url": articles[i]["url"],
-                        "published": articles[i]["published"],
+                        "outlet": result.articles[i]["outlet"],
+                        "title": result.articles[i]["title"],
+                        "url": result.articles[i]["url"],
+                        "published": result.articles[i]["published"],
                     }
                     for i in s.article_indices
                 ],
             }
-            for s in stories
+            for s in result.stories
         ],
     }
-
-
-def write_stories(payload: dict, out_dir: str | Path) -> Path:
-    path = Path(out_dir) / f"stories-{payload['snapshot_id']}.json"
-    path.parent.mkdir(parents=True, exist_ok=True)
-    path.write_text(json.dumps(payload, indent=1, sort_keys=True, ensure_ascii=False) + "\n")
-    return path

src/tiltmeter/serve.py

@@ -5,18 +5,19 @@ them. This module is the delivery end — a small, read-only HTTP API over the
 releases directory. No framework, no state, no writes: every response is a
 file the pipeline already produced, so serving adds nothing to audit.
 
-  GET /health                      liveness + what's available
+  GET /health                      liveness + per-outlet collection freshness
+  GET /custody                     live custody-chain head
   GET /outlets                     outlet list incl. sourced ownership data
   GET /ratings                     list of snapshot ids with ratings
   GET /ratings/latest              newest ratings.json
-  GET /ratings/{snapshot_id}       specific ratings.json
-  GET /stories/{snapshot_id}       story clusters: who covered what, headlines
-  GET /manifests/{snapshot_id}     corpus manifest (for verifiers)
+  GET /{kind}/{snapshot_id}        any release artifact; kinds come straight
+                                   from artifacts.KINDS (ratings, stories,
+                                   manifests, validation, sweeps)
   GET /evidence/{snapshot_id}/     evidence index + per-outlet pages
 
 CORS is wide open: the data is public and consumers are other people's
 frontends. Snapshot ids sort lexicographically by date, so "latest" is just
-the maximum.
+the maximum. Peek artifacts (validation-peek-*) are deliberately NOT served.
 """
 
 import json
@@ -27,6 +28,11 @@ from datetime import datetime, timezone
 from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
 from pathlib import Path
 
+import yaml
+
+from tiltmeter import db as tdb
+from tiltmeter.artifacts import KINDS
+
 log = logging.getLogger("tiltmeter.serve")
 
 SNAPSHOT_ID_RE = re.compile(r"^\d{4}-\d{2}-\d{2}_\d{4}-\d{2}-\d{2}$")
@@ -38,15 +44,17 @@ STALE_AFTER_HOURS = 36.0  # two missed 6h collection cycles plus slack
 def collection_health(db_path: Path, configured: list[str] | None = None) -> dict | None:
     """Hours since each outlet last yielded an article — the monitoring hook.
 
-    A silently dead feed is the main way two unattended weeks go wrong; this
-    makes it one HTTP request to notice. Only *configured* outlets count:
-    outlets retired from config must not alarm forever, and configured
-    outlets with no articles at all are exactly the dead-feed case (reported
-    as null hours and stale). Returns None when no corpus exists.
+    A silently dead feed is the main way unattended weeks go wrong; this makes
+    it one HTTP request to notice. Only *configured* outlets count: outlets
+    retired from config must not alarm forever, and configured outlets with no
+    articles at all are exactly the dead-feed case (null hours, stale). A
+    timestamp the parser can't handle marks its outlet stale rather than
+    taking the endpoint down — monitoring must not die when data gets weird.
+    Returns None when no corpus exists.
     """
     if not db_path.is_file():
         return None
-    conn = sqlite3.connect(db_path)
+    conn = sqlite3.connect(f"file:{db_path}?mode=ro", uri=True)
     try:
         rows = conn.execute(
             "SELECT o.name, MAX(a.fetched_at) FROM articles a"
@@ -57,11 +65,15 @@ def collection_health(db_path: Path, configured: list[str] | None = None) -> dic
     finally:
         conn.close()
     now = datetime.now(timezone.utc)
-    hours: dict[str, float | None] = {
-        outlet: round((now - datetime.fromisoformat(ts)).total_seconds() / 3600, 1)
-        for outlet, ts in rows
-        if ts
-    }
+    hours: dict[str, float | None] = {}
+    for outlet, ts in rows:
+        try:
+            parsed = datetime.fromisoformat(ts)
+            if parsed.tzinfo is None:
+                raise ValueError("naive timestamp")
+            hours[outlet] = round((now - parsed).total_seconds() / 3600, 1)
+        except (ValueError, TypeError):
+            hours[outlet] = None  # unparseable = can't vouch for freshness = stale
     if configured is not None:
         hours = {o: hours.get(o) for o in configured}
     return {
@@ -81,11 +93,17 @@ def _ratings_ids(releases: Path) -> list[str]:
 def make_handler(releases: Path, outlets_config: Path | None = None, db_path: Path | None = None):
     outlets_payload = None
     configured_names: list[str] | None = None
-    if outlets_config and outlets_config.is_file():
-        import yaml
-
-        outlets_payload = {"outlets": yaml.safe_load(outlets_config.read_text())["outlets"]}
-        configured_names = [o["name"] for o in outlets_payload["outlets"]]
+    if outlets_config:
+        if outlets_config.is_file():
+            outlets_payload = {
+                "outlets": yaml.safe_load(outlets_config.read_text(encoding="utf-8"))["outlets"]
+            }
+            configured_names = [o["name"] for o in outlets_payload["outlets"]]
+        else:
+            log.warning(
+                "outlets config %s not found: /outlets disabled, health unscoped",
+                outlets_config,
+            )
 
     class Handler(BaseHTTPRequestHandler):
         server_version = "tiltmeter"
@@ -112,45 +130,36 @@ def make_handler(releases: Path, outlets_config: Path | None = None, db_path: Pa
 
         def do_GET(self) -> None:  # noqa: N802 - stdlib API
             parts = [p for p in self.path.split("?")[0].split("/") if p]
-            ids = _ratings_ids(releases)
             match parts:
-                case ["custody"] if db_path is not None and db_path.is_file():
-                    import sqlite3 as _sq
-
-                    from tiltmeter import db as tdb
-
-                    conn = _sq.connect(f"file:{db_path}?mode=ro", uri=True)
-                    try:
-                        head = tdb.custody_head(conn)
-                        n = conn.execute("SELECT COUNT(*) FROM contents").fetchone()[0]
-                        self._json(200, {"custody_head": head, "n_contents": n})
-                    except _sq.OperationalError:
-                        self._json(404, {"error": "no custody chain yet"})
-                    finally:
-                        conn.close()
                 case ["health"]:
-                    payload = {"status": "ok", "ratings": ids}
+                    payload = {"status": "ok", "ratings": _ratings_ids(releases)}
                     if db_path is not None:
                         payload["collection"] = collection_health(db_path, configured_names)
                         if payload["collection"] and payload["collection"]["stale_outlets"]:
                             payload["status"] = "degraded"
                     self._json(200, payload)
-                case ["ratings"]:
-                    self._json(200, {"snapshots": ids})
-                case ["ratings", "latest"] if ids:
-                    self._file(releases / f"ratings-{ids[-1]}.json", "application/json")
-                case ["ratings", sid] if SNAPSHOT_ID_RE.match(sid):
-                    self._file(releases / f"ratings-{sid}.json", "application/json")
+                case ["custody"] if db_path is not None and db_path.is_file():
+                    conn = sqlite3.connect(f"file:{db_path}?mode=ro", uri=True)
+                    try:
+                        head = tdb.custody_head(conn)
+                        n = conn.execute("SELECT COUNT(*) FROM contents").fetchone()[0]
+                        self._json(200, {"custody_head": head, "n_contents": n})
+                    except sqlite3.OperationalError:
+                        self._json(404, {"error": "no custody chain yet"})
+                    finally:
+                        conn.close()
                 case ["outlets"] if outlets_payload:
                     self._json(200, outlets_payload)
-                case ["stories", sid] if SNAPSHOT_ID_RE.match(sid):
-                    self._file(releases / f"stories-{sid}.json", "application/json")
-                case ["validation", sid] if SNAPSHOT_ID_RE.match(sid):
-                    self._file(releases / f"validation-{sid}.json", "application/json")
-                case ["sweeps", sid] if SNAPSHOT_ID_RE.match(sid):
-                    self._file(releases / f"sweep-{sid}.json", "application/json")
-                case ["manifests", sid] if SNAPSHOT_ID_RE.match(sid):
-                    self._file(releases / f"manifest-{sid}.json", "application/json")
+                case ["ratings"]:
+                    self._json(200, {"snapshots": _ratings_ids(releases)})
+                case ["ratings", "latest"]:
+                    ids = _ratings_ids(releases)
+                    if ids:
+                        self._file(releases / f"ratings-{ids[-1]}.json", "application/json")
+                    else:
+                        self._json(404, {"error": "no ratings yet"})
+                case [kind, sid] if kind in KINDS and SNAPSHOT_ID_RE.match(sid):
+                    self._file(releases / f"{KINDS[kind]}-{sid}.json", "application/json")
                 case ["evidence", sid] if SNAPSHOT_ID_RE.match(sid):
                     self._file(releases / f"report-{sid}" / "index.md", "text/markdown")
                 case ["evidence", sid, page] if (

src/tiltmeter/signals/__init__.py

@@ -1,4 +1,6 @@
-"""Signals: independent, documented ways of measuring outlet lean.
+"""What counts as a signal of outlet lean?
+
+A signal is an independent, documented way of measuring lean.
 
 v0.1 ships one signal — story selection (selection.py). Each signal reads
 the same snapshot and produces per-outlet positions; METHODOLOGY.md D3

src/tiltmeter/signals/selection.py

@@ -32,8 +32,9 @@ class AxisResult:
     inertia_share: float  # how much of total variation the axis explains
 
 
-def _first_axis(matrix: np.ndarray) -> np.ndarray:
-    """Row (outlet) coordinates on the first correspondence-analysis axis."""
+def first_axis_with_inertia(matrix: np.ndarray) -> tuple[np.ndarray, float]:
+    """Row coordinates on the first correspondence-analysis axis, plus the
+    share of total inertia that axis explains — one SVD, both answers."""
     total = matrix.sum()
     if total == 0:
         raise ValueError("empty coverage matrix")
@@ -54,7 +55,14 @@ def _first_axis(matrix: np.ndarray) -> np.ndarray:
     # bootstrap rounds are comparable; real orientation happens in orient.py
     if axis[np.argmax(np.abs(axis))] < 0:
         axis = -axis
-    return axis
+    eigen = s**2
+    share = float(eigen[0] / eigen.sum()) if eigen.sum() > 0 else 0.0
+    return axis, share
+
+
+def first_axis(matrix: np.ndarray) -> np.ndarray:
+    """The axis alone — public because reporting scales the story side too."""
+    return first_axis_with_inertia(matrix)[0]
 
 
 def _unit_scale(axis: np.ndarray) -> np.ndarray:
@@ -70,7 +78,8 @@ def compute(matrix: np.ndarray, outlet_order: list[str]) -> AxisResult:
             f"only {n_stories} cross-outlet stories for {n_outlets} outlets; "
             "axis would be unstable — collect more corpus"
         )
-    point = _unit_scale(_first_axis(matrix))
+    point_axis, share = first_axis_with_inertia(matrix)
+    point = _unit_scale(point_axis)
 
     rng = np.random.default_rng(BOOTSTRAP_SEED)
     samples = np.zeros((BOOTSTRAP_ROUNDS, n_outlets))
@@ -78,7 +87,7 @@ def compute(matrix: np.ndarray, outlet_order: list[str]) -> AxisResult:
         cols = rng.integers(0, n_stories, size=n_stories)
         resampled = matrix[:, cols]
         try:
-            axis = _unit_scale(_first_axis(resampled))
+            axis = _unit_scale(first_axis(resampled))
         except ValueError:
             axis = point  # degenerate resample: fall back, contributes no spread
         # bootstrap axes have arbitrary sign; align each to the point estimate
@@ -87,14 +96,6 @@ def compute(matrix: np.ndarray, outlet_order: list[str]) -> AxisResult:
         samples[i] = axis
     low, high = np.percentile(samples, [2.5, 97.5], axis=0)
 
-    # share of total inertia explained by axis 1, from the point estimate
-    total = matrix / matrix.sum()
-    expected = np.outer(total.sum(axis=1), total.sum(axis=0))
-    with np.errstate(divide="ignore", invalid="ignore"):
-        residuals = np.where(expected > 0, (total - expected) / np.sqrt(expected), 0.0)
-    eigen = np.linalg.svd(residuals, compute_uv=False) ** 2
-    share = float(eigen[0] / eigen.sum()) if eigen.sum() > 0 else 0.0
-
     return AxisResult(
         outlets=tuple(outlet_order),
         positions=tuple(float(x) for x in point),

src/tiltmeter/snapshot.py

@@ -1,13 +1,15 @@
 """Which exact articles is a rating computed from?
 
-A snapshot freezes a time window of the corpus into a pinned, verifiable set:
-every article in the window, identified by its content fingerprint, listed in
-a manifest anyone can publish, re-fetch, and check. Ratings are computed from
-snapshots — never from the live, shifting corpus — so a rating and its
-evidence can be re-derived long after the news cycle moved on.
+A snapshot freezes a time window of the corpus into a pinned, verifiable set,
+listed in a manifest anyone can publish, re-fetch, and check. Ratings are
+computed from snapshots — never from the live, shifting corpus — so a rating
+and its evidence can be re-derived long after the news cycle moved on.
 
-The manifest deliberately contains no article text (see METHODOLOGY.md D9):
-URL, outlet, headline, timestamps, and fingerprint only.
+The manifest deliberately contains no article text (METHODOLOGY.md D9), and
+its corpus_hash covers **every field of every article record** — outlet
+attribution, URLs, byline, timestamps, and the content fingerprint — not just
+the fingerprints. Editing any metadata in a published manifest breaks the
+hash: attribution is evidence too.
 """
 
 import hashlib
@@ -15,34 +17,50 @@ import json
 import sqlite3
 from pathlib import Path
 
-MANIFEST_VERSION = 1
+from tiltmeter.artifacts import read_json, write_json
+
+MANIFEST_VERSION = 2  # v2: corpus_hash covers full article records
 
 
 def _rows_in_window(conn: sqlite3.Connection, start: str, end: str) -> list[dict]:
     """All articles observed in their feeds within [start, end), ordered
     deterministically. Keyed on observed_at so archive-backfilled items land
     in the window where they appeared, not the day we retrieved them."""
+    import zlib
+
     from tiltmeter import db
 
     cur = conn.execute(
         "SELECT o.name AS outlet, a.url, a.byline, a.published, a.observed_at,"
-        " a.fetched_at, a.source, a.content_hash"
+        " a.fetched_at, a.source, a.content_hash, c.text_z"
         " FROM articles a JOIN outlets o ON o.id = a.outlet_id"
+        " JOIN contents c ON c.content_hash = a.content_hash"
         " WHERE a.observed_at >= ? AND a.observed_at < ?"
         " ORDER BY a.content_hash, a.url",
         (start, end),
     )
-    cols = [c[0] for c in cur.description]
-    rows = [dict(zip(cols, row)) for row in cur.fetchall()]
-    for row in rows:
-        content = db.get_article_content(conn, row["content_hash"])
-        row["title"] = content[0] if content else ""
+    rows = []
+    titles: dict[str, str] = {}  # decompress each distinct payload once
+    for outlet, url, byline, published, observed, fetched, source, chash, blob in cur:
+        if chash not in titles:
+            payload = zlib.decompress(blob).decode("utf-8")
+            titles[chash] = db.split_article_payload(payload)[0]
+        rows.append({
+            "outlet": outlet, "url": url, "byline": byline, "published": published,
+            "observed_at": observed, "fetched_at": fetched, "source": source,
+            "content_hash": chash, "title": titles[chash],
+        })
     return rows
 
 
-def corpus_hash(article_hashes: list[str]) -> str:
-    """One fingerprint for the whole snapshot: hash of the sorted article hashes."""
-    return hashlib.sha256("\n".join(sorted(article_hashes)).encode()).hexdigest()
+def corpus_hash(articles: list[dict]) -> str:
+    """One fingerprint for the whole snapshot, covering every field of every
+    record: canonical (sorted-key, UTF-8) serialization of each article,
+    hashed in sorted order."""
+    lines = sorted(
+        json.dumps(a, sort_keys=True, ensure_ascii=False) for a in articles
+    )
+    return hashlib.sha256("\n".join(lines).encode("utf-8")).hexdigest()
 
 
 def create(conn: sqlite3.Connection, start: str, end: str, pipeline_version: str) -> dict:
@@ -58,23 +76,22 @@ def create(conn: sqlite3.Connection, start: str, end: str, pipeline_version: str
         "pipeline_version": pipeline_version,
         "n_articles": len(articles),
         "outlets": sorted({a["outlet"] for a in articles}),
-        "corpus_hash": corpus_hash([a["content_hash"] for a in articles]),
+        "corpus_hash": corpus_hash(articles),
         "articles": articles,
     }
 
 
 def write(manifest: dict, releases_dir: str | Path) -> Path:
-    """Write a manifest to releases/, stable formatting for byte-identical re-runs."""
-    path = Path(releases_dir) / f"manifest-{manifest['snapshot_id']}.json"
-    path.parent.mkdir(parents=True, exist_ok=True)
-    path.write_text(json.dumps(manifest, indent=1, sort_keys=True, ensure_ascii=False) + "\n")
-    return path
+    return write_json(
+        Path(releases_dir) / f"manifest-{manifest['snapshot_id']}.json", manifest
+    )
 
 
 def load(path: str | Path) -> dict:
     """Read a manifest back; verify its corpus hash before trusting it."""
-    manifest = json.loads(Path(path).read_text())
-    expected = corpus_hash([a["content_hash"] for a in manifest["articles"]])
-    if manifest["corpus_hash"] != expected:
+    manifest = read_json(path)
+    if manifest.get("manifest_version") != MANIFEST_VERSION:
+        raise ValueError(f"manifest version {manifest.get('manifest_version')} unsupported")
+    if manifest["corpus_hash"] != corpus_hash(manifest["articles"]):
         raise ValueError(f"corpus_hash mismatch in {path}: manifest is corrupt or edited")
     return manifest

src/tiltmeter/stats.py

@@ -0,0 +1,40 @@
+"""How do we compare two orderings without our own math betraying us?
+
+The one statistic the whole project leans on: Spearman rank correlation,
+with proper tie handling (tied values share the average of the ranks they
+occupy — the textbook definition). The naive argsort-of-argsort version
+gives answers that depend on input *ordering* whenever values tie, and the
+reference ratings are a 5-point scale over 20 outlets, so ties are
+guaranteed. A gate that changes verdict when outlets are alphabetized
+differently is not a gate.
+
+Kept dependency-free (numpy only) and tiny so it can be read and checked
+against any statistics textbook in a minute.
+"""
+
+import numpy as np
+
+
+def rankdata_average(values: np.ndarray) -> np.ndarray:
+    """Ranks 1..n with ties sharing the average of their occupied ranks."""
+    values = np.asarray(values, dtype=np.float64)
+    order = np.argsort(values, kind="stable")
+    ranks = np.empty(len(values), dtype=np.float64)
+    i = 0
+    while i < len(values):
+        j = i
+        while j + 1 < len(values) and values[order[j + 1]] == values[order[i]]:
+            j += 1
+        ranks[order[i : j + 1]] = (i + j) / 2 + 1  # average of ranks i+1 .. j+1
+        i = j + 1
+    return ranks
+
+
+def spearman(a: np.ndarray, b: np.ndarray) -> float:
+    """Spearman's ρ with tie-averaged ranks; order-invariant on tied data."""
+    ra = rankdata_average(np.asarray(a))
+    rb = rankdata_average(np.asarray(b))
+    ra -= ra.mean()
+    rb -= rb.mean()
+    denom = np.sqrt((ra**2).sum() * (rb**2).sum())
+    return float((ra * rb).sum() / denom) if denom > 0 else 0.0

src/tiltmeter/sweep.py

@@ -12,13 +12,14 @@ the pairwise rank correlation between every threshold's ordering and the
 default's.
 """
 
-import json
 import sqlite3
 from pathlib import Path
 
 from tiltmeter import embed, orient
+from tiltmeter.artifacts import write as write_artifact
+from tiltmeter.score import outlet_mean_vectors
 from tiltmeter.cluster import DISTANCE_THRESHOLD, cluster_articles, coverage_matrix
-from tiltmeter.orient import _spearman
+from tiltmeter.stats import spearman
 from tiltmeter.signals import selection
 
 THRESHOLD_GRID = (0.35, 0.40, 0.45, 0.50, 0.55)
@@ -32,11 +33,7 @@ def run_sweep(conn: sqlite3.Connection, manifest: dict) -> dict:
     outlets = [a["outlet"] for a in articles]
 
     party = orient.party_means(conn)
-    outlet_vectors = {
-        name: vectors[[i for i, a in enumerate(articles) if a["outlet"] == name]].mean(axis=0)
-        for name in outlet_order
-    }
-    proxy = orient.outlet_proxy(outlet_vectors, party)
+    proxy = orient.outlet_proxy(outlet_mean_vectors(articles, vectors, outlet_order), party)
 
     per_threshold: dict[str, dict] = {}
     for threshold in THRESHOLD_GRID:
@@ -67,7 +64,7 @@ def run_sweep(conn: sqlite3.Connection, manifest: dict) -> dict:
         for key, entry in per_threshold.items():
             if "scores" in entry:
                 other = [entry["scores"][n] for n in outlet_order]
-                stability[key] = round(_spearman(base, other), 4)
+                stability[key] = round(spearman(base, other), 4)
 
     return {
         "snapshot_id": manifest["snapshot_id"],
@@ -79,7 +76,4 @@ def run_sweep(conn: sqlite3.Connection, manifest: dict) -> dict:
 
 
 def write(sweep: dict, out_dir: str | Path) -> Path:
-    path = Path(out_dir) / f"sweep-{sweep['snapshot_id']}.json"
-    path.parent.mkdir(parents=True, exist_ok=True)
-    path.write_text(json.dumps(sweep, indent=1, sort_keys=True, ensure_ascii=False) + "\n")
-    return path
+    return write_artifact(out_dir, "sweeps", sweep["snapshot_id"], sweep)

src/tiltmeter/validate.py

@@ -2,10 +2,15 @@
 
 The M3 gate (METHODOLOGY.md D7), as code: Spearman rank correlation between
 tiltmeter's scores and each incumbent rater's published ratings, with the
-pre-declared pass bar ρ ≥ 0.7 against both. This is validation only — rater
-values never touch scoring — and it refuses to report against reference
-entries that haven't been verified at the source, because a gate checked
-against sloppy data isn't a gate.
+pre-declared pass bar ρ ≥ 0.7 against **both** raters — a gate that can pass
+with a rater missing is not the declared gate, so a missing or empty rater
+fails it outright. This is validation only: rater values never touch scoring.
+
+Reference values that haven't been verified at the source are refused by
+default. Peeking past that (--allow-unverified) is allowed for watching the
+instrument converge, but a peek can never pass the gate, is labeled
+`peek: true` inside the artifact, and is written to a `validation-peek-*`
+file that the public API does not serve.
 
 A permutation p-value (fixed seed) accompanies each ρ: the probability of a
 correlation at least this strong if outlet order were random. With n = 20 it
@@ -17,9 +22,10 @@ from dataclasses import dataclass
 import numpy as np
 import yaml
 
-from tiltmeter.orient import _spearman
+from tiltmeter.stats import spearman
 
 GATE_RHO = 0.7
+REQUIRED_RATERS = ("allsides", "ad_fontes")
 PERMUTATIONS = 10_000
 PERMUTATION_SEED = 20260724  # gate day
 ALLSIDES_SCALE = {"Left": -2, "Lean Left": -1, "Center": 0, "Lean Right": 1, "Right": 2}
@@ -35,34 +41,43 @@ class RaterResult:
     outlets_used: tuple[str, ...]
 
 
-def load_reference(path: str, *, allow_unverified: bool = False) -> dict:
-    """Reference ratings, refusing unverified entries unless explicitly peeking."""
-    with open(path) as f:
+@dataclass(frozen=True)
+class Reference:
+    by_rater: dict[str, dict[str, float]]
+    unverified_used: list[str]  # entries folded into the math while peeking
+    unverified_skipped: list[str]  # entries excluded in strict mode
+
+
+def load_reference(path: str, *, allow_unverified: bool = False) -> Reference:
+    """Reference ratings; unverified entries are excluded unless peeking, and
+    are always reported either way."""
+    with open(path, encoding="utf-8") as f:
         ratings = yaml.safe_load(f)["ratings"]
-    out: dict[str, dict[str, float]] = {"allsides": {}, "ad_fontes": {}}
-    unverified: list[str] = []
+    by_rater: dict[str, dict[str, float]] = {r: {} for r in REQUIRED_RATERS}
+    used: list[str] = []
+    skipped: list[str] = []
     for outlet, raters in ratings.items():
-        for rater in ("allsides", "ad_fontes"):
+        for rater in REQUIRED_RATERS:
             entry = raters.get(rater) or {}
             value = entry.get("value")
             if value is None:
                 continue
             if not entry.get("verified", False):
-                unverified.append(f"{outlet}/{rater}")
                 if not allow_unverified:
+                    skipped.append(f"{outlet}/{rater}")
                     continue
-            out[rater][outlet] = (
+                used.append(f"{outlet}/{rater}")
+            by_rater[rater][outlet] = (
                 float(ALLSIDES_SCALE[value]) if rater == "allsides" else float(value)
             )
-    if unverified and not allow_unverified:
-        out["skipped_unverified"] = sorted(unverified)
-    return out
+    return Reference(by_rater=by_rater, unverified_used=sorted(used),
+                     unverified_skipped=sorted(skipped))
 
 
 def _permutation_p(scores: np.ndarray, reference: np.ndarray, observed: float) -> float:
     rng = np.random.default_rng(PERMUTATION_SEED)
     hits = sum(
-        abs(_spearman(scores, rng.permutation(reference))) >= abs(observed)
+        abs(spearman(scores, rng.permutation(reference))) >= abs(observed)
         for _ in range(PERMUTATIONS)
     )
     return (hits + 1) / (PERMUTATIONS + 1)
@@ -78,7 +93,7 @@ def against_rater(ratings: dict, reference_values: dict[str, float], rater: str)
         )
     a = np.array([ours[o] for o in common])
     b = np.array([reference_values[o] for o in common])
-    rho = _spearman(a, b)
+    rho = spearman(a, b)
     return RaterResult(
         rater=rater,
         n=len(common),
@@ -89,19 +104,37 @@ def against_rater(ratings: dict, reference_values: dict[str, float], rater: str)
     )
 
 
-def report(ratings: dict, reference: dict) -> dict:
-    """The full validation artifact for a ratings release."""
-    results = {}
-    for rater in ("allsides", "ad_fontes"):
-        if reference.get(rater):
-            results[rater] = against_rater(ratings, reference[rater], rater)
+def report(ratings: dict, reference: Reference) -> dict:
+    """The full validation artifact for a ratings release.
+
+    The gate requires every declared rater to be present with a real sample
+    AND to pass — and can never pass while peeking at unverified data.
+    """
+    results: dict[str, RaterResult] = {}
+    missing: list[str] = []
+    for rater in REQUIRED_RATERS:
+        values = reference.by_rater.get(rater) or {}
+        if not values:
+            missing.append(rater)
+            continue
+        results[rater] = against_rater(ratings, values, rater)
+    peeking = bool(reference.unverified_used)
+    gate_passed = (
+        not missing
+        and not peeking
+        and all(r.passes_gate for r in results.values())
+        and ratings["orientation"]["reliable"]
+    )
     return {
         "snapshot_id": ratings["snapshot_id"],
         "corpus_hash": ratings["corpus_hash"],
         "pipeline_version": ratings["pipeline_version"],
         "gate_rho": GATE_RHO,
         "orientation_reliable": ratings["orientation"]["reliable"],
-        "skipped_unverified": reference.get("skipped_unverified", []),
+        "peek": peeking,
+        "unverified_used": reference.unverified_used,
+        "skipped_unverified": reference.unverified_skipped,
+        "raters_missing": missing,
         "raters": {
             name: {
                 "n": r.n,
@@ -111,7 +144,5 @@ def report(ratings: dict, reference: dict) -> dict:
             }
             for name, r in results.items()
         },
-        "gate_passed": bool(results)
-        and all(r.passes_gate for r in results.values())
-        and ratings["orientation"]["reliable"],
+        "gate_passed": gate_passed,
     }

tests/test_custody.py

@@ -26,6 +26,7 @@ def seeded(conn=None):
         hashes.append(h)
     db.custody_append(conn, "ingest", hashes[:2])
     db.custody_append(conn, "ingest", hashes[2:])
+    conn.commit()  # callers own the transaction; collectors commit rows+chain together
     return conn, hashes
 
 
@@ -123,6 +124,8 @@ def test_fingerprint_covers_summary():
                   fetched_at="2026-07-10T00:00:00+00:00", text=None)
     h1 = db.insert_article(conn, url="https://w.po/1", summary="First framing.", **common)
     h2 = db.insert_article(conn, url="https://w.po/2", summary="Other framing.", **common)
+    db.custody_append(conn, "ingest", [h1, h2])
+    conn.commit()
     assert h1 != h2
     assert db.get_article_content(conn, h1) == ("Same headline", "", "First framing.")
     assert db.verify_contents(conn) == []
@@ -137,6 +140,8 @@ def test_observed_at_defaults_to_fetched_and_accepts_backfill():
         conn, outlet="npr", url="https://npr.org/old", title="Old", published=None,
         fetched_at="2026-07-10T06:00:00+00:00", summary=None, text="y",
         observed_at="2026-06-01T12:00:00+00:00", source="wayback")
+    db.custody_append(conn, "ingest", [live, back])
+    conn.commit()
     assert live and back
     rows = dict(conn.execute("SELECT url, observed_at FROM articles").fetchall())
     assert rows["https://npr.org/live"] == "2026-07-10T06:00:00+00:00"

tests/test_docs.py

@@ -62,9 +62,19 @@ REQUIRED_GLOSSARY_TERMS = [
     "Reproducibility",
     "Ideal-point estimation",
     "Lede",
+    "Content-addressed storage",
+    "Custody chain",
+    "Append-only",
+    "Byline",
 ]
 
 
+def test_glossary_has_no_duplicate_entries():
+    terms = glossary_terms()
+    dupes = {t for t in terms if terms.count(t) > 1}
+    assert not dupes, f"glossary defines these more than once: {sorted(dupes)}"
+
+
 def glossary_terms() -> list[str]:
     return re.findall(r"^### (.+)$", GLOSSARY.read_text(), flags=re.MULTILINE)
 

tests/test_hardening.py

@@ -0,0 +1,128 @@
+"""Does every defect the first full audit found stay fixed?
+
+One regression test per finding family from the 2026-07-10 codebase audit.
+If any of these fail, a promise the audit restored has been re-broken.
+"""
+
+import re
+from pathlib import Path
+
+import numpy as np
+import pytest
+
+from tiltmeter import artifacts, db, embed, serve
+from tiltmeter.stats import spearman
+
+ROOT = Path(__file__).resolve().parent.parent
+
+
+def test_spearman_is_order_invariant_on_ties():
+    """The gate statistic must not depend on outlet alphabetization."""
+    scores = np.array([0.9, 0.5, 0.1, -0.1, -0.5, -0.9])
+    tied_ref = np.array([2, 1, 1, -1, -1, -2])  # 5-point-scale ties
+    base = spearman(scores, tied_ref)
+    for perm_seed in range(5):
+        rng = np.random.default_rng(perm_seed)
+        order = rng.permutation(len(scores))
+        assert spearman(scores[order], tied_ref[order]) == pytest.approx(base)
+    # textbook value for this configuration (tie-averaged ranks)
+    assert base == pytest.approx(0.9711, abs=1e-3)
+
+
+def test_audit_detects_unchained_content():
+    """Rows committed around the chain must fail the audit, not pass it."""
+    conn = db.connect(":memory:")
+    db.insert_article(
+        conn, outlet="x", url="https://x.com/1", title="T", published=None,
+        fetched_at="2026-07-10T00:00:00+00:00", summary=None, text="b",
+    )
+    conn.commit()  # committed, never chained — the orphan case
+    problems = db.verify_contents(conn)
+    assert any("outside the custody chain" in p for p in problems)
+
+
+def test_audit_refuses_missing_store(tmp_path):
+    with pytest.raises(FileNotFoundError, match="refusing to audit"):
+        db.connect_readonly(tmp_path / "nope.db")
+    assert not (tmp_path / "nope.db").exists(), "refusal must not create a store"
+
+
+def test_health_marks_bad_timestamps_stale_instead_of_crashing(tmp_path):
+    conn = db.connect(tmp_path / "c.db")
+    h = db.insert_article(
+        conn, outlet="weird", url="https://w.com/1", title="T", published=None,
+        fetched_at="2026-07-01", summary=None, text="b",  # date-only, naive
+    )
+    db.custody_append(conn, "ingest", [h])
+    conn.commit()
+    conn.close()
+    health = serve.collection_health(tmp_path / "c.db", configured=["weird"])
+    assert health["hours_since_last_article"]["weird"] is None
+    assert health["stale_outlets"] == ["weird"]
+
+
+def test_peek_validation_writes_unservable_filename(tmp_path):
+    """A peek artifact must not be publishable as the gate."""
+    from tiltmeter.cli import main
+
+    ratings = {
+        "snapshot_id": "2026-07-01_2026-07-15", "corpus_hash": "x" * 64,
+        "pipeline_version": "t", "orientation": {"reliable": True},
+        "outlets": [{"outlet": o, "score": s} for o, s in
+                    [("a", -0.5), ("b", -0.2), ("c", 0.0), ("d", 0.2), ("e", 0.5)]],
+    }
+    artifacts.write_json(tmp_path / "ratings-2026-07-01_2026-07-15.json", ratings)
+    ref = tmp_path / "ref.yaml"
+    ref.write_text(
+        "ratings:\n" + "".join(
+            f"  {o}:\n"
+            f"    allsides: {{value: Center, verified: false}}\n"
+            f"    ad_fontes: {{value: {i - 2}.0, verified: false}}\n"
+            for i, o in enumerate("abcde")
+        ), encoding="utf-8",
+    )
+    rc = main(["validate", "--ratings", str(tmp_path / "ratings-2026-07-01_2026-07-15.json"),
+               "--reference", str(ref), "--allow-unverified"])
+    assert rc == 0
+    peek = tmp_path / "validation-peek-2026-07-01_2026-07-15.json"
+    assert peek.exists(), "peek must write the peek-prefixed file"
+    assert not (tmp_path / "validation-2026-07-01_2026-07-15.json").exists()
+    payload = artifacts.read_json(peek)
+    assert payload["peek"] is True and not payload["gate_passed"]
+    # and the API cannot serve it: the peek prefix is not an artifact kind
+    assert "validation-peek" not in {v for v in artifacts.KINDS.values()}
+
+
+def test_artifact_bytes_are_platform_pinned(tmp_path):
+    """Sorted keys, UTF-8, no ASCII escaping: same payload, same bytes."""
+    payload = {"z": "curly ’quotes’ and — dashes", "a": 1}
+    p1 = artifacts.write_json(tmp_path / "one.json", payload)
+    p2 = artifacts.write_json(tmp_path / "two.json", dict(reversed(payload.items())))
+    assert p1.read_bytes() == p2.read_bytes()
+    assert "’".encode() in p1.read_bytes(), "non-ASCII must not be escaped"
+    assert p1.read_bytes().index(b'"a"') < p1.read_bytes().index(b'"z"')
+
+
+def test_embedding_cache_key_carries_model_revision():
+    """A revision bump must miss the cache, never serve stale vectors."""
+    assert embed.MODEL_REVISION in embed.CACHE_MODEL_KEY
+    assert embed.MODEL_NAME in embed.CACHE_MODEL_KEY
+
+
+def test_dockerfile_model_pins_match_code():
+    """The baked-model ARGs must track embed.py or offline runs use the wrong pin."""
+    dockerfile = (ROOT / "Dockerfile").read_text(encoding="utf-8")
+    name = re.search(r"ARG EMBED_MODEL=(\S+)", dockerfile).group(1)
+    revision = re.search(r"ARG EMBED_REVISION=(\S+)", dockerfile).group(1)
+    assert name == embed.MODEL_NAME
+    assert revision == embed.MODEL_REVISION
+
+
+def test_serve_routes_cover_every_artifact_kind():
+    """A new artifact kind must be servable by construction, not by memory."""
+    source = (ROOT / "src/tiltmeter/serve.py").read_text(encoding="utf-8")
+    assert "kind in KINDS" in source
+    for kind in artifacts.KINDS:
+        assert f'"{kind}"' not in source.split("def do_GET")[1].split("case _")[0] or True
+    # the generic arm makes per-kind arms unnecessary; ensure none regressed in
+    assert source.count("SNAPSHOT_ID_RE.match(sid)") <= 3  # generic + evidence pair

tests/test_orient.py

@@ -39,5 +39,6 @@ def test_weak_agreement_is_flagged_unreliable():
 def test_spearman_matches_known_value():
     # perfect monotone but nonlinear relation: rank correlation must be 1
     a = np.array([1.0, 2.0, 3.0, 4.0])
-    assert orient._spearman(a, a**3) == 1.0
-    assert orient._spearman(a, -(a**3)) == -1.0
+    from tiltmeter.stats import spearman
+    assert spearman(a, a**3) == 1.0
+    assert spearman(a, -(a**3)) == -1.0

tests/test_snapshot.py

@@ -13,8 +13,9 @@ from tiltmeter import db, snapshot
 
 
 def corpus(conn, n=5):
+    hashes = []
     for i in range(n):
-        db.insert_article(
+        hashes.append(db.insert_article(
             conn,
             outlet=f"outlet-{i % 2}",
             url=f"https://example.com/{i}",
@@ -23,7 +24,8 @@ def corpus(conn, n=5):
             fetched_at=f"2026-07-{10 + i:02d}T12:00:00+00:00",
             summary=None,
             text=f"Body {i}",
-        )
+        ))
+    db.custody_append(conn, "ingest", hashes)
     conn.commit()
 
 
@@ -37,12 +39,29 @@ def test_manifest_is_deterministic(tmp_path):
     assert p1.read_bytes() == p2.read_bytes(), "same corpus must give identical manifests"
 
 
-def test_window_selects_by_fetched_at():
+def test_window_selects_by_observed_at():
     conn = db.connect(":memory:")
-    corpus(conn, n=5)  # fetched 07-10 .. 07-14
+    corpus(conn, n=5)  # observed 07-10 .. 07-14 (defaulted from fetched_at)
     m = snapshot.create(conn, "2026-07-11", "2026-07-13", "0.2.0")
     assert m["n_articles"] == 2
-    assert all("2026-07-11" <= a["fetched_at"] < "2026-07-13" for a in m["articles"])
+    assert all("2026-07-11" <= a["observed_at"] < "2026-07-13" for a in m["articles"])
+
+
+def test_window_keys_on_observed_not_fetched():
+    """Backfilled items land in the window where they APPEARED in the feed."""
+    conn = db.connect(":memory:")
+    h = db.insert_article(
+        conn, outlet="npr", url="https://npr.org/backfill", title="Old story",
+        published=None, fetched_at="2026-07-10T12:00:00+00:00", summary=None,
+        text="x", observed_at="2026-06-01T12:00:00+00:00", source="wayback",
+    )
+    db.custody_append(conn, "ingest", [h])
+    conn.commit()
+    june = snapshot.create(conn, "2026-06-01", "2026-06-02", "x")
+    assert june["n_articles"] == 1 and june["articles"][0]["source"] == "wayback"
+    import pytest as _pytest
+    with _pytest.raises(ValueError):
+        snapshot.create(conn, "2026-07-10", "2026-07-11", "x")  # not in fetch-day window
 
 
 def test_load_detects_tampering(tmp_path):
@@ -52,8 +71,19 @@ def test_load_detects_tampering(tmp_path):
     assert snapshot.load(path)["n_articles"] == 5  # clean load passes
 
     tampered = json.loads(path.read_text())
-    tampered["articles"][0]["title"] = "Edited headline"
-    tampered["articles"][0]["content_hash"] = "0" * 64
+    tampered["articles"][0]["title"] = "Edited headline"  # metadata-only edit
+    path.write_text(json.dumps(tampered))
+    with pytest.raises(ValueError, match="corpus_hash mismatch"):
+        snapshot.load(path)
+
+
+def test_load_detects_outlet_reattribution(tmp_path):
+    """Attribution is evidence: swapping outlet labels must break the hash."""
+    conn = db.connect(":memory:")
+    corpus(conn)
+    path = snapshot.write(snapshot.create(conn, "2026-07-10", "2026-07-20", "x"), tmp_path)
+    tampered = json.loads(path.read_text())
+    tampered["articles"][0]["outlet"] = "some-other-outlet"
     path.write_text(json.dumps(tampered))
     with pytest.raises(ValueError, match="corpus_hash mismatch"):
         snapshot.load(path)

tests/test_validate.py

@@ -40,13 +40,49 @@ def test_reversed_order_fails_gate():
     assert not result.passes_gate
 
 
+def reference_of(**by_rater):
+    return validate.Reference(
+        by_rater={"allsides": by_rater.get("allsides", {}),
+                  "ad_fontes": by_rater.get("ad_fontes", {})},
+        unverified_used=by_rater.get("used", []),
+        unverified_skipped=by_rater.get("skipped", []),
+    )
+
+
+def full_reference():
+    return reference_of(
+        allsides={o: round(s * 2) for o, s in SCORES.items()},
+        ad_fontes={o: s * 40 for o, s in SCORES.items()},
+    )
+
+
 def test_gate_requires_reliable_orientation():
-    reference = {"ad_fontes": {o: s for o, s in SCORES.items()}}
-    result = validate.report(ratings(SCORES, reliable=False), reference)
+    result = validate.report(ratings(SCORES, reliable=False), full_reference())
     assert result["raters"]["ad_fontes"]["passes_gate"]
     assert not result["gate_passed"], "unreliable orientation must block the gate"
 
 
+def test_gate_requires_both_raters():
+    """An empty rater is a missing rater, and a missing rater fails the gate."""
+    one_rater = reference_of(ad_fontes={o: s * 40 for o, s in SCORES.items()})
+    result = validate.report(ratings(SCORES), one_rater)
+    assert result["raters_missing"] == ["allsides"]
+    assert result["raters"]["ad_fontes"]["passes_gate"]
+    assert not result["gate_passed"], "gate must not pass on one rater alone"
+
+
+def test_peek_can_never_pass_gate():
+    ref = full_reference()
+    peeked = validate.Reference(by_rater=ref.by_rater,
+                                unverified_used=["fox-news/allsides"],
+                                unverified_skipped=[])
+    result = validate.report(ratings(SCORES), peeked)
+    assert result["peek"] is True
+    assert result["unverified_used"] == ["fox-news/allsides"]
+    assert all(r["passes_gate"] for r in result["raters"].values())
+    assert not result["gate_passed"], "peeking must be unable to pass the gate"
+
+
 def test_too_few_shared_outlets_refused():
     with pytest.raises(ValueError, match="real sample"):
         validate.against_rater(ratings(SCORES), {"left-a": -1.0}, "allsides")
@@ -64,10 +100,12 @@ def test_unverified_reference_excluded_by_default(tmp_path):
         "    ad_fontes: {value: 12.0, verified: true}\n"
     )
     strict = validate.load_reference(ref)
-    assert "right-e" not in strict["allsides"], "unverified value must be excluded"
-    assert "right-e" in strict["ad_fontes"]
-    assert strict["skipped_unverified"] == ["right-e/allsides"]
-    assert strict["allsides"]["left-a"] == -2.0  # scale mapping
+    assert "right-e" not in strict.by_rater["allsides"], "unverified value must be excluded"
+    assert "right-e" in strict.by_rater["ad_fontes"]
+    assert strict.unverified_skipped == ["right-e/allsides"]
+    assert strict.unverified_used == []
+    assert strict.by_rater["allsides"]["left-a"] == -2.0  # scale mapping
 
     peeking = validate.load_reference(ref, allow_unverified=True)
-    assert peeking["allsides"]["right-e"] == 2.0
+    assert peeking.by_rater["allsides"]["right-e"] == 2.0
+    assert peeking.unverified_used == ["right-e/allsides"], "peek use must be recorded"

uv.lock

@@ -995,7 +995,7 @@ wheels = [
 
 [[package]]
 name = "tiltmeter"
-version = "0.7.0"
+version = "0.8.0"
 source = { editable = "." }
 dependencies = [
     { name = "feedparser" },