v0.8.1: audit pass-2 remediation

Savepoints make content+metadata writes atomic per entry; cached_embed runs
no DDL and never commits a caller's batch; cycle windows derive from UTC;
ratings pin the reference-corpus state; thin-sample raters are recorded
gate failures with automation-safe exit codes; snapshot fails loudly on
missing content; audit refuses empty stores and appends custody-heads.jsonl
history; repair adopts orphans as visible 'adopt' batches; timestamps
validated tz-aware UTC at insert; cleanup sweep (shared publish path,
README drift, dead code, encoding pins). 65 tests.

CHANGELOG.md

@@ -8,6 +8,43 @@ requires a version bump and, if it changes methodology, a decision record in
 
 ## [Unreleased]
 
+## [0.8.1] - 2026-07-10
+
+Audit pass 2 (three angles over the pass-1 remediation): 23 findings, all
+fixed. The loop converges: pass 1 found 19, pass 2 found progressively
+narrower issues, each now pinned by a regression test.
+
+### Fixed
+
+- Entry-level savepoints in insert_article/insert_speech: a failure between
+  the content write and the metadata write rolls back both, so a batch
+  commit can never durably orphan unchained content.
+- cached_embed no longer runs DDL (moved to db.connect) and never commits
+  inside a caller's open batch — embedding mid-collection cannot break the
+  rows+chain atomicity contract.
+- cycle derives its window from UTC (was host-local date.today(): a TZ-ahead
+  host would mutate a published release id intra-day) and uses one instant
+  for start/end.
+- Ratings record the exact reference-corpus state (hash + speech count):
+  same manifest + same reference state is the full, checkable
+  reproducibility precondition.
+- validate: thin-sample raters (1-4 shared outlets) are recorded gate
+  failures, not tracebacks; exit codes are automation-safe (0 pass, 2 fail).
+- snapshot fails loudly when a manifested article's content row is missing
+  (was: silently smaller manifest masking store corruption).
+- audit fails on an empty store; emits an append-only custody-heads.jsonl
+  history alongside the head file (external anchoring against wholesale
+  chain recomputation).
+- report regeneration clears stale evidence pages; insert timestamps are
+  validated as timezone-aware and normalized to UTC (string-compared
+  windows stay chronological); file: URI paths no longer mkdir junk.
+- New `tiltmeter repair`: adopts custody-orphaned content (pre-0.8
+  interruptions, partial restores) as an explicit, visible 'adopt' chain
+  batch.
+- Cleanup sweep: shared _publish_release (run and cycle can't diverge),
+  README quickstart/status drift fixed, dead helpers and a vacuous test
+  assertion removed, remaining unpinned-encoding I/O fixed.
+
 ## [0.8.0] - 2026-07-10
 
 Full-codebase audit (8 finder angles, 42 candidates, 19 verified findings)
@@ -274,7 +311,8 @@ human attention and produce trustworthy corpus + fresh dry-run ratings.
   deliberately not used — news and opinion are rated separately by every
   incumbent rater).
 
-[Unreleased]: https://github.com/sudolulo/tiltmeter/compare/v0.8.0...HEAD
+[Unreleased]: https://github.com/sudolulo/tiltmeter/compare/v0.8.1...HEAD
+[0.8.1]: https://github.com/sudolulo/tiltmeter/compare/v0.8.0...v0.8.1
 [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

README.md

@@ -24,9 +24,13 @@ technical spec with sources: [METHODOLOGY.md](METHODOLOGY.md).
 
 ## Status
 
-Pre-alpha (v0.1, milestone M1). The corpus collector is running; scoring lands in
-M2; the first validated ratings release requires ≥2 weeks of corpus (M3). Nothing
-here is a usable rating yet.
+Pre-alpha, pre-validation. The full pipeline runs — collection, scoring,
+sensitivity sweeps, custody audits, and the read-only API — and a deployment
+recomputes everything on a 6-hour `tiltmeter cycle`. The first validated
+ratings release awaits the pre-declared M3 gate (≥2 weeks of corpus, Spearman
+ρ ≥ 0.7 against both incumbent raters). Until then, published numbers carry
+explicit unreliability flags and exist to watch the instrument converge.
+Nothing here is a usable rating yet.
 
 ## Quickstart
 
@@ -52,8 +56,8 @@ Or with Docker (the pinned embedding model is baked into the image, so
 recomputation works offline):
 
 ```sh
-docker compose up -d                        # API on :8477
-docker compose run --rm tiltmeter ingest    # any pipeline command as one-shot
+docker compose up -d                  # API on :8477 + 6-hourly collector
+docker compose run --rm api ingest    # any pipeline command as one-shot
 ```
 
 tiltmeter is a **data layer**: it computes and serves ratings JSON plus evidence

pyproject.toml

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

releases/custody-head.json

@@ -1,10 +1,10 @@
 {
  "custody_head": {
-  "entry_hash": "cb604de99fe1c210ebdde49108168d8571f08f4cf456eebf9fd7352700e8462f",
-  "seq": 20,
-  "ts": "2026-07-10T16:57:50.515431+00:00"
+  "entry_hash": "19ce5759013547de69da70dcf8a0d11cfeb9e4307cad684b4601a5813f6dc03a",
+  "seq": 24,
+  "ts": "2026-07-10T17:12:47.045616+00:00"
  },
  "intact": true,
- "n_contents": 572,
+ "n_contents": 579,
  "problems": []
 }

releases/custody-heads.jsonl

@@ -0,0 +1 @@
+{"entry_hash": "19ce5759013547de69da70dcf8a0d11cfeb9e4307cad684b4601a5813f6dc03a", "n_contents": 579, "seq": 24, "ts": "2026-07-10T17:12:47.045616+00:00"}

scripts/gen_outlets_doc.py

@@ -51,11 +51,11 @@ def render() -> str:
 if __name__ == "__main__":
     content = render()
     if "--check" in sys.argv:
-        current = TARGET.read_text() if TARGET.exists() else ""
+        current = TARGET.read_text(encoding="utf-8") if TARGET.exists() else ""
         if current != content:
             print("docs/outlets.md is stale; run: uv run python scripts/gen_outlets_doc.py")
             sys.exit(1)
         print("docs/outlets.md is in sync")
     else:
-        TARGET.write_text(content)
+        TARGET.write_text(content, encoding="utf-8")
         print(f"wrote {TARGET}")

src/tiltmeter/__init__.py

@@ -7,4 +7,4 @@ 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.8.0"
+__version__ = "0.8.1"

src/tiltmeter/cli.py

@@ -67,21 +67,33 @@ def cmd_reference(args: argparse.Namespace) -> int:
     return 0
 
 
-def cmd_run(args: argparse.Namespace) -> int:
-    from tiltmeter import __version__, artifacts, report, score, snapshot
+def _publish_release(conn, manifest: dict, out_dir: str):
+    """Compute once, publish everything: ratings, stories, evidence pages.
+
+    The single path both manual runs and scheduled cycles go through, so the
+    two can never diverge in what a release contains.
+    """
+    from tiltmeter import __version__, artifacts, report, score
 
-    manifest = snapshot.load(args.manifest)
-    conn = db.connect(args.db)
     result = score.compute(conn, manifest, __version__)
-    ratings_path = artifacts.write(args.out, "ratings", manifest["snapshot_id"], result.ratings)
+    ratings_path = artifacts.write(out_dir, "ratings", manifest["snapshot_id"], result.ratings)
     stories_path = artifacts.write(
-        args.out, "stories", manifest["snapshot_id"], score.stories_json(result, manifest)
+        out_dir, "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,
+        out_dir,
     )
+    return result, ratings_path, stories_path, report_dir
+
+
+def cmd_run(args: argparse.Namespace) -> int:
+    from tiltmeter import snapshot
+
+    manifest = snapshot.load(args.manifest)
+    conn = db.connect(args.db)
+    result, ratings_path, stories_path, report_dir = _publish_release(conn, manifest, args.out)
 
     print(f"ratings: {ratings_path}\nstories: {stories_path}\nevidence: {report_dir}/")
     o = result.ratings["orientation"]
@@ -106,36 +118,29 @@ def cmd_cycle(args: argparse.Namespace) -> int:
     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 datetime import datetime, timedelta, timezone
 
-    from tiltmeter import __version__, artifacts, report, score, snapshot
+    from tiltmeter import __version__, snapshot
 
     rc = cmd_ingest(args)
 
     from tiltmeter import reference
 
     conn = db.connect(args.db)
+    # one UTC "today" for the whole cycle: observed_at values are UTC, so the
+    # window must be too, and start/end must come from the same instant
+    today = datetime.now(timezone.utc).date()
     try:
-        end = date.today()
-        reference.fetch_range(conn, end.isoformat(), args.reference_days, args.congress)
+        reference.fetch_range(conn, today.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()
+    start = (today - timedelta(days=WINDOW_DAYS)).isoformat()
+    end = 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,
-        )
+        _publish_release(conn, manifest, args.out)
         print(f"  rolling release {manifest['snapshot_id']} written")
     except ValueError as exc:
         print(f"  no rolling release: {exc}")
@@ -173,7 +178,8 @@ def cmd_validate(args: argparse.Namespace) -> int:
     if not result["orientation_reliable"]:
         print("  orientation UNRELIABLE — gate cannot pass regardless of rho")
     print(f"  GATE: {'PASSED' if result['gate_passed'] else 'not passed'}  -> {out}")
-    return 0
+    # automation-friendly: 0 = gate passed, 2 = evaluated but not passed
+    return 0 if result["gate_passed"] else 2
 
 
 def cmd_sweep(args: argparse.Namespace) -> int:
@@ -212,11 +218,26 @@ def cmd_audit(args: argparse.Namespace) -> int:
         return 1
     finally:
         conn.close()
+    if n_contents == 0:
+        print("  AUDIT FAILED: store is empty — nothing to attest"
+              " (wrong --db path, or collection never ran)")
+        return 1
     if args.emit:
+        from pathlib import Path
+
         artifacts.write_json(args.emit, {
             "custody_head": head, "n_contents": n_contents,
             "intact": not problems, "problems": problems,
         })
+        # append-only head log: an external copy of this file constrains any
+        # future attempt to rewrite the chain wholesale
+        log_path = Path(args.emit).parent / "custody-heads.jsonl"
+        with open(log_path, "a", encoding="utf-8") as f:
+            import json as _json
+
+            f.write(_json.dumps({"seq": head["seq"], "ts": head["ts"],
+                                 "entry_hash": head["entry_hash"],
+                                 "n_contents": n_contents}, sort_keys=True) + "\n")
     print(f"  contents: {n_contents} items, chain head seq {head['seq']}")
     if problems:
         for p in problems[:20]:
@@ -227,6 +248,22 @@ def cmd_audit(args: argparse.Namespace) -> int:
     return 0
 
 
+def cmd_repair(args: argparse.Namespace) -> int:
+    """Adopt orphaned content into the custody chain, visibly.
+
+    For stores damaged by pre-0.8 collector interruptions or partial
+    restores: orphans are chained in an explicit 'adopt' batch, so the chain
+    records the irregularity instead of hiding it. Audit afterwards.
+    """
+    conn = db.connect(args.db)
+    entry = db.custody_adopt_orphans(conn)
+    if entry is None:
+        print("  nothing to repair — no orphaned content")
+        return 0
+    print(f"  adopted {entry['n_items']} orphaned items as chain seq {entry['seq']}")
+    return 0
+
+
 def cmd_serve(args: argparse.Namespace) -> int:
     from tiltmeter import serve
 
@@ -309,6 +346,11 @@ def main(argv: list[str] | None = None) -> int:
     p_audit.add_argument("--emit", help="also write an audit summary JSON to this path")
     p_audit.set_defaults(func=cmd_audit)
 
+    p_repair = sub.add_parser(
+        "repair", help="adopt custody-orphaned content into the chain (visible 'adopt' batch)"
+    )
+    p_repair.set_defaults(func=cmd_repair)
+
     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,

src/tiltmeter/db.py

@@ -100,7 +100,7 @@ def connect(db_path: str | Path) -> sqlite3.Connection:
     store — auditing above all — uses connect_readonly instead.
     """
     s = str(db_path)
-    if s != ":memory:":
+    if s != ":memory:" and not s.startswith("file:"):
         Path(s).parent.mkdir(parents=True, exist_ok=True)
     conn = sqlite3.connect(s)
     conn.execute("PRAGMA journal_mode=WAL")
@@ -113,6 +113,19 @@ def connect(db_path: str | Path) -> sqlite3.Connection:
             " delete the database (and stale releases) and recollect"
         )
     conn.executescript(SCHEMA_V3)
+    # the embeddings cache is derived data (custody-exempt, rebuildable); its
+    # DDL lives here so no pipeline call ever runs executescript — which would
+    # implicitly commit a half-collected batch — mid-transaction
+    legacy_cache = conn.execute(
+        "SELECT 1 FROM pragma_table_info('embeddings') WHERE name = 'content_hash'"
+    ).fetchone()
+    if legacy_cache:
+        conn.execute("DROP TABLE embeddings")
+    conn.execute(
+        "CREATE TABLE IF NOT EXISTS embeddings ("
+        " text_hash TEXT NOT NULL, model TEXT NOT NULL, vector BLOB NOT NULL,"
+        " PRIMARY KEY (text_hash, model))"
+    )
     conn.execute(f"PRAGMA user_version = {SCHEMA_VERSION}")
     conn.commit()
     return conn
@@ -148,13 +161,6 @@ def _store_content(conn: sqlite3.Connection, chash: str, payload: str) -> None:
     )
 
 
-def get_content(conn: sqlite3.Connection, chash: str) -> str | None:
-    row = conn.execute(
-        "SELECT text_z FROM contents WHERE content_hash = ?", (chash,)
-    ).fetchone()
-    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] = {}
@@ -178,14 +184,6 @@ def split_article_payload(payload: str) -> tuple[str, str, str]:
     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
-    return split_article_payload(payload)
-
-
 def outlet_id(conn: sqlite3.Connection, name: str, first_seen: str | None = None) -> int:
     """Get-or-create the outlet dimension row; outlet names are stored once."""
     row = conn.execute("SELECT id FROM outlets WHERE name = ?", (name,)).fetchone()
@@ -201,6 +199,22 @@ def have_url(conn: sqlite3.Connection, url: str) -> bool:
     return row is not None
 
 
+def utc_timestamp(value: str, field: str) -> str:
+    """Validate and normalize a timestamp to UTC ISO format.
+
+    Snapshot windows compare these as strings, so every stored timestamp must
+    be UTC and offset-aware or windowing silently becomes lexicographic
+    nonsense. Reject naive timestamps; normalize any offset to +00:00.
+    """
+    try:
+        parsed = datetime.fromisoformat(value)
+    except (ValueError, TypeError) as exc:
+        raise ValueError(f"{field} is not an ISO timestamp: {value!r}") from exc
+    if parsed.tzinfo is None:
+        raise ValueError(f"{field} must be timezone-aware UTC, got naive {value!r}")
+    return parsed.astimezone(timezone.utc).isoformat()
+
+
 def insert_article(
     conn: sqlite3.Connection,
     *,
@@ -220,15 +234,26 @@ def insert_article(
     if the URL was already present."""
     if have_url(conn, url):
         return None
+    fetched_at = utc_timestamp(fetched_at, "fetched_at")
+    observed_at = utc_timestamp(observed_at, "observed_at") if observed_at else fetched_at
     chash = content_hash(title, text or "", summary or "")
-    _store_content(conn, chash, _article_payload(title, text or "", summary or ""))
-    conn.execute(
-        "INSERT INTO articles"
-        " (outlet_id, url, url_original, byline, published, observed_at, fetched_at, source,"
-        " content_hash) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)",
-        (outlet_id(conn, outlet, fetched_at), url, url_original, byline,
-         published, observed_at or fetched_at, fetched_at, source, chash),
-    )
+    # savepoint: content row and metadata row land together or not at all — a
+    # failure between them must never strand unchained content in the batch
+    conn.execute("SAVEPOINT insert_item")
+    try:
+        _store_content(conn, chash, _article_payload(title, text or "", summary or ""))
+        conn.execute(
+            "INSERT INTO articles"
+            " (outlet_id, url, url_original, byline, published, observed_at, fetched_at, source,"
+            " content_hash) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)",
+            (outlet_id(conn, outlet, fetched_at), url, url_original, byline,
+             published, observed_at, fetched_at, source, chash),
+        )
+    except Exception:
+        conn.execute("ROLLBACK TO insert_item")
+        conn.execute("RELEASE insert_item")
+        raise
+    conn.execute("RELEASE insert_item")
     return chash
 
 
@@ -252,13 +277,20 @@ def insert_speech(
     ).fetchone()
     if existing:
         return None
-    _store_content(conn, chash, text)
-    conn.execute(
-        "INSERT INTO reference_speeches"
-        " (day, granule, chamber, speaker, state, party, content_hash)"
-        " VALUES (?, ?, ?, ?, ?, ?, ?)",
-        (day, granule, chamber, speaker, state, party, chash),
-    )
+    conn.execute("SAVEPOINT insert_item")
+    try:
+        _store_content(conn, chash, text)
+        conn.execute(
+            "INSERT INTO reference_speeches"
+            " (day, granule, chamber, speaker, state, party, content_hash)"
+            " VALUES (?, ?, ?, ?, ?, ?, ?)",
+            (day, granule, chamber, speaker, state, party, chash),
+        )
+    except Exception:
+        conn.execute("ROLLBACK TO insert_item")
+        conn.execute("RELEASE insert_item")
+        raise
+    conn.execute("RELEASE insert_item")
     return chash
 
 
@@ -350,6 +382,23 @@ def verify_contents(conn: sqlite3.Connection) -> list[str]:
     return problems
 
 
+def custody_adopt_orphans(conn: sqlite3.Connection) -> dict | None:
+    """Chain any collected content that no chain entry covers, as an explicit
+    'adopt' batch. The chain honestly records that these items were adopted
+    late (interrupted pre-0.8 collector, restored backup) rather than
+    pretending they arrived normally. Returns the new entry, or None."""
+    orphans = [
+        r[0] for r in conn.execute(
+            "SELECT content_hash FROM contents"
+            " WHERE content_hash NOT IN (SELECT content_hash FROM custody_items)"
+            " ORDER BY content_hash"
+        )
+    ]
+    entry = custody_append(conn, "adopt", orphans)
+    conn.commit()
+    return entry
+
+
 def custody_verify(conn: sqlite3.Connection) -> list[str]:
     """Walk the whole chain; return problems (empty list = intact)."""
     problems = []

src/tiltmeter/embed.py

@@ -29,16 +29,6 @@ _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 (
-    text_hash TEXT NOT NULL,
-    model TEXT NOT NULL,
-    vector BLOB NOT NULL,
-    PRIMARY KEY (text_hash, model)
-);
-"""
-
-
 def passage(title: str, text: str | None, summary: str | None) -> str:
     """The text we embed for an article: headline plus lede (or feed summary
     as fallback). Everything read here comes from the fingerprinted payload."""
@@ -69,16 +59,11 @@ def cached_embed(conn: sqlite3.Connection, texts: list[str]) -> np.ndarray:
 
     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.
+    The cache table is created by db.connect (derived data, custody-exempt);
+    no DDL happens here, so calling this mid-transaction can never implicitly
+    commit a half-collected batch.
     """
-    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)
+    owns_transaction = not conn.in_transaction
     keys = [hashlib.sha256(t.encode("utf-8")).hexdigest() for t in texts]
     found: dict[str, np.ndarray] = {}
     unique = list(dict.fromkeys(keys))
@@ -99,7 +84,8 @@ def cached_embed(conn: sqlite3.Connection, texts: list[str]) -> np.ndarray:
             "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()
+        if owns_transaction:
+            conn.commit()  # inside a caller's batch, cache rows ride its commit
         found.update(dict(zip(missing, vectors)))
     return np.stack([found[k] for k in keys])
 
@@ -113,16 +99,16 @@ def embed_hashes(conn: sqlite3.Connection, content_hashes: list[str]) -> np.ndar
     """
     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]
+    unique = list(dict.fromkeys(content_hashes))
+    payloads = db.get_contents(conn, unique)
+    absent = [h for h in unique 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])
+    passages = {
+        chash: passage(*db.split_article_payload(payload))
+        for chash, payload in payloads.items()
+    }
+    # cached_embed dedups and preserves row order itself
+    return cached_embed(conn, [passages[h] for h in content_hashes])

src/tiltmeter/orient.py

@@ -29,7 +29,6 @@ class Orientation:
     sign: int  # +1 keep, -1 flip
     correlation: float  # spearman rho between axis and party-language proxy
     reliable: bool
-    proxy_by_outlet: tuple[float, ...]  # cos-to-R minus cos-to-D per outlet
 
 
 def party_means(conn: sqlite3.Connection) -> dict[str, np.ndarray]:
@@ -76,5 +75,4 @@ def orient_sign(axis_positions: list[float], proxy_values: list[float]) -> Orien
         sign=-1 if rho < 0 else 1,
         correlation=rho,
         reliable=abs(rho) >= MIN_ABS_CORRELATION,
-        proxy_by_outlet=tuple(float(p) for p in proxy_values),
     )

src/tiltmeter/report.py

@@ -102,6 +102,8 @@ def render(ratings: dict, stories: list, matrix: np.ndarray, articles: list) ->
 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 stale in directory.glob("*.md"):
+        stale.unlink()  # regeneration must never leave pages from a prior run
     for filename, content in pages.items():
         (directory / filename).write_text(content, encoding="utf-8")
     return directory

src/tiltmeter/score.py

@@ -16,7 +16,7 @@ from dataclasses import dataclass
 
 import numpy as np
 
-from tiltmeter import embed, orient
+from tiltmeter import db, embed, orient
 from tiltmeter.cluster import Story, cluster_articles, coverage_matrix
 from tiltmeter.signals import selection
 
@@ -65,6 +65,15 @@ def compute(conn: sqlite3.Connection, manifest: dict, pipeline_version: str) ->
     )
     s = orientation.sign
 
+    # the reference corpus is a scoring input the manifest does not pin, so
+    # its exact state is recorded here: same manifest + same reference state
+    # (both hashes in the output) is the full reproducibility precondition
+    speech_hashes = [
+        r[0] for r in conn.execute(
+            "SELECT content_hash FROM reference_speeches ORDER BY content_hash"
+        )
+    ]
+
     covered_counts = matrix.sum(axis=1)
     outlets_out = [
         {
@@ -93,6 +102,8 @@ def compute(conn: sqlite3.Connection, manifest: dict, pipeline_version: str) ->
             "method": "party-mean speech embeddings (ADR-0003)",
             "correlation": round(orientation.correlation, 6),
             "reliable": orientation.reliable,
+            "reference_corpus_hash": db.items_hash(speech_hashes),
+            "n_reference_speeches": len(speech_hashes),
         },
         "outlets": outlets_out,
     }

src/tiltmeter/snapshot.py

@@ -34,7 +34,7 @@ def _rows_in_window(conn: sqlite3.Connection, start: str, end: str) -> list[dict
         "SELECT o.name AS outlet, a.url, a.byline, a.published, a.observed_at,"
         " 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"
+        " LEFT 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),
@@ -42,6 +42,13 @@ def _rows_in_window(conn: sqlite3.Connection, start: str, end: str) -> list[dict
     rows = []
     titles: dict[str, str] = {}  # decompress each distinct payload once
     for outlet, url, byline, published, observed, fetched, source, chash, blob in cur:
+        if blob is None:
+            # a manifested article whose content is gone is store corruption;
+            # a silently smaller manifest would mask it — fail loudly instead
+            raise RuntimeError(
+                f"article {url} has no content row ({chash[:12]}…) — store is"
+                " corrupt; run tiltmeter audit"
+            )
         if chash not in titles:
             payload = zlib.decompress(blob).decode("utf-8")
             titles[chash] = db.split_article_payload(payload)[0]

src/tiltmeter/validate.py

@@ -117,7 +117,12 @@ def report(ratings: dict, reference: Reference) -> dict:
         if not values:
             missing.append(rater)
             continue
-        results[rater] = against_rater(ratings, values, rater)
+        try:
+            results[rater] = against_rater(ratings, values, rater)
+        except ValueError:
+            # 1-4 shared outlets: too thin to correlate — a failed rater is a
+            # recorded gate failure, never a crash with no artifact
+            missing.append(rater)
     peeking = bool(reference.unverified_used)
     gate_passed = (
         not missing

tests/test_custody.py

@@ -2,9 +2,9 @@
 
 The chain-of-custody promise, made executable: content edits, deletions,
 history rewrites, and silent schema drift must all be caught by
-`tiltmeter audit`'s two checks (custody_verify + verify_contents). Plus the
-v1→v2 migration must preserve every fingerprint, or every manifest published
-before the migration would become unverifiable.
+`tiltmeter audit`'s two checks (custody_verify + verify_contents). Pre-v3
+stores are refused outright — early-development stores are recollected,
+never migrated — and that refusal is pinned here too.
 """
 
 import sqlite3
@@ -127,7 +127,9 @@ def test_fingerprint_covers_summary():
     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.split_article_payload(
+        db.get_contents(conn, [h1])[h1]
+    ) == ("Same headline", "", "First framing.")
     assert db.verify_contents(conn) == []
 
 

tests/test_docs.py

@@ -155,7 +155,7 @@ def test_outlets_doc_in_sync_with_config():
     )
     mod = importlib.util.module_from_spec(spec)
     spec.loader.exec_module(mod)
-    committed = (ROOT / "docs" / "outlets.md").read_text()
+    committed = (ROOT / "docs" / "outlets.md").read_text(encoding="utf-8")
     assert committed == mod.render(), (
         "docs/outlets.md is stale. Regenerate: uv run python scripts/gen_outlets_doc.py"
     )

tests/test_hardening.py

@@ -47,13 +47,35 @@ def test_audit_refuses_missing_store(tmp_path):
     assert not (tmp_path / "nope.db").exists(), "refusal must not create a store"
 
 
+def test_insert_rejects_naive_timestamps():
+    """Windowing compares strings, so UTC-awareness is enforced at the door."""
+    conn = db.connect(":memory:")
+    with pytest.raises(ValueError, match="timezone-aware"):
+        db.insert_article(
+            conn, outlet="x", url="https://x.com/1", title="T", published=None,
+            fetched_at="2026-07-01", summary=None, text="b",
+        )
+    # offsets are normalized to UTC so string comparison stays chronological
+    h = db.insert_article(
+        conn, outlet="x", url="https://x.com/2", title="T", published=None,
+        fetched_at="2026-07-09T23:00:00-04:00", summary=None, text="b",
+    )
+    assert h
+    row = conn.execute("SELECT observed_at FROM articles").fetchone()[0]
+    assert row == "2026-07-10T03:00:00+00:00"
+
+
 def test_health_marks_bad_timestamps_stale_instead_of_crashing(tmp_path):
+    """Legacy or tampered rows can still hold junk timestamps; the monitoring
+    endpoint must mark them stale, never die."""
     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
+        fetched_at="2026-07-01T00:00:00+00:00", summary=None, text="b",
     )
     db.custody_append(conn, "ingest", [h])
+    # simulate legacy/tampered data: junk timestamp written around the API
+    conn.execute("UPDATE articles SET fetched_at = '2026-07-01' WHERE content_hash = ?", (h,))
     conn.commit()
     conn.close()
     health = serve.collection_health(tmp_path / "c.db", configured=["weird"])
@@ -83,7 +105,7 @@ def test_peek_validation_writes_unservable_filename(tmp_path):
     )
     rc = main(["validate", "--ratings", str(tmp_path / "ratings-2026-07-01_2026-07-15.json"),
                "--reference", str(ref), "--allow-unverified"])
-    assert rc == 0
+    assert rc == 2, "a peek is by definition not a passed gate: exit 2"
     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()
@@ -93,6 +115,72 @@ def test_peek_validation_writes_unservable_filename(tmp_path):
     assert "validation-peek" not in {v for v in artifacts.KINDS.values()}
 
 
+def test_failed_insert_cannot_strand_unchained_content(monkeypatch):
+    """A failure between the content write and the metadata write must roll
+    back both — otherwise the batch commit would durably orphan content."""
+    conn = db.connect(":memory:")
+
+    def boom(*a, **k):
+        raise RuntimeError("simulated failure after content write")
+
+    monkeypatch.setattr(db, "outlet_id", boom)
+    with pytest.raises(RuntimeError, match="simulated"):
+        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()
+    assert conn.execute("SELECT COUNT(*) FROM contents").fetchone()[0] == 0
+
+
+def test_repair_adopts_orphans_visibly():
+    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 around the chain: the orphan case
+    assert any("outside the custody chain" in p for p in db.verify_contents(conn))
+    entry = db.custody_adopt_orphans(conn)
+    assert entry["n_items"] == 1
+    kind = conn.execute("SELECT kind FROM custody_log WHERE seq = ?", (entry["seq"],))
+    assert kind.fetchone()[0] == "adopt", "adoption must be visible in the chain"
+    assert db.verify_contents(conn) == []
+    assert db.custody_verify(conn) == []
+
+
+def test_snapshot_fails_loudly_on_missing_content(tmp_path):
+    from tiltmeter import snapshot
+
+    conn = db.connect(tmp_path / "s.db")
+    h = 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",
+    )
+    db.custody_append(conn, "ingest", [h])
+    conn.commit()
+    conn.close()
+    raw = __import__("sqlite3").connect(tmp_path / "s.db")  # tamperer: no FK pragma
+    raw.execute("DELETE FROM contents WHERE content_hash = ?", (h,))
+    raw.commit()
+    raw.close()
+    with pytest.raises(RuntimeError, match="store is"):
+        snapshot.create(db.connect(tmp_path / "s.db"), "2026-07-10", "2026-07-11", "x")
+
+
+def test_cached_embed_never_commits_inside_a_callers_batch(monkeypatch):
+    """Embedding mid-collection must not commit the half-collected batch."""
+    conn = db.connect(":memory:")
+    monkeypatch.setattr(embed, "embed_texts",
+                        lambda texts: np.zeros((len(texts), 4), dtype=np.float32))
+    conn.execute("BEGIN")
+    conn.execute("INSERT INTO outlets (name, first_seen) VALUES ('x', 't')")
+    embed.cached_embed(conn, ["some text"])
+    assert conn.in_transaction, "cached_embed must not have committed the batch"
+    conn.rollback()
+    assert conn.execute("SELECT COUNT(*) FROM outlets").fetchone()[0] == 0
+
+
 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}
@@ -121,8 +209,6 @@ def test_dockerfile_model_pins_match_code():
 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
+    assert "kind in KINDS" in source, "artifact routes must come from artifacts.KINDS"
     # the generic arm makes per-kind arms unnecessary; ensure none regressed in
     assert source.count("SNAPSHOT_ID_RE.match(sid)") <= 3  # generic + evidence pair

uv.lock

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