shards: integrity & lineage-correctness (sign, collapse revisions, no default-fill bias)

.github/workflows/deploy.yml

@@ -125,12 +125,17 @@ jobs:
           ANSIBLE_HOST_KEY_CHECKING: "false"
           VAULT_ANTHROPIC_API_KEY: ${{ secrets.ANTHROPIC_API_KEY }}
           VAULT_NEWS_API_KEY: ${{ secrets.NEWS_API_KEY }}
+          # Ed25519 shard-signing seed (64 hex chars). Set the repo secret with
+          # `gh secret set ZEITGHOST_SIGNING_KEY`. Empty until then — env.j2
+          # defaults it to '' so the deploy renders and ingest stays unsigned.
+          VAULT_ZEITGHOST_SIGNING_KEY: ${{ secrets.ZEITGHOST_SIGNING_KEY }}
         run: |
           TARGET="${{ inputs.target || 'us-ny1' }}"
           ansible-playbook deploy.yml \
             -i "inventories/${TARGET}/hosts.yml" \
             -e "vault_anthropic_api_key=${VAULT_ANTHROPIC_API_KEY}" \
             -e "vault_news_api_key=${VAULT_NEWS_API_KEY}" \
+            -e "vault_zeitghost_signing_key=${VAULT_ZEITGHOST_SIGNING_KEY}" \
             -e "zeitghost_commit_sha=${{ github.sha }}"
 
       - name: Health check

infra/ansible/templates/env.j2

@@ -4,6 +4,14 @@
 COMPOSE_PROJECT_NAME=zeitghost
 ANTHROPIC_API_KEY={{ vault_anthropic_api_key }}
 NEWS_API_KEY={{ vault_news_api_key }}
+# Ed25519 seed (64 hex chars) for signing shards — mint with
+# `zeitghost gen-signing-key`, store in the vault as vault_zeitghost_signing_key.
+# Empty until provisioned (defaulted so the deploy still renders).
+ZEITGHOST_SIGNING_KEY={{ vault_zeitghost_signing_key | default('') }}
+# Set zeitghost_require_signing: 1 in the inventory to fail-close ingest once
+# the key above is in the vault and verified. Off by default so rolling this
+# out doesn't break ingest before the key is provisioned.
+ZEITGHOST_REQUIRE_SIGNING={{ zeitghost_require_signing | default(0) }}
 ZEITGHOST_FEEDS={{ zeitghost_feeds | default('feeds/newsapi.yaml') }}
 ZEITGHOST_INTERVAL={{ zeitghost_interval | default(3600) }}
 ZEITGHOST_INGEST_LIMIT={{ zeitghost_ingest_limit | default(50) }}

infra/docker/docker-compose.yml

@@ -34,6 +34,10 @@ services:
     environment:
       ANTHROPIC_API_KEY: ${ANTHROPIC_API_KEY:-}
       NEWS_API_KEY: ${NEWS_API_KEY:-}
+      # Shard signing — seed (secret) + fail-closed switch. Values live in .env
+      # (rendered 0600 by Ansible from the vault); only the references are here.
+      ZEITGHOST_SIGNING_KEY: ${ZEITGHOST_SIGNING_KEY:-}
+      ZEITGHOST_REQUIRE_SIGNING: ${ZEITGHOST_REQUIRE_SIGNING:-0}
       ZEITGHOST_FEEDS: ${ZEITGHOST_FEEDS:-feeds/newsapi.yaml}
       ZEITGHOST_INTERVAL: ${ZEITGHOST_INTERVAL:-3600}
       ZEITGHOST_INGEST_LIMIT: ${ZEITGHOST_INGEST_LIMIT:-50}

tests/test_shard_integrity.py

@@ -0,0 +1,226 @@
+"""Shard integrity & lineage-correctness tests.
+
+Covers three behaviors:
+- #5  bias_score is never default-filled (skip an unscored article).
+- #2  load collapses a revision chain to the latest shard per entity.
+- #4  shards are signed when (and only when) a signing seed is configured.
+"""
+
+import os
+
+import pytest
+
+from cryptography.hazmat.primitives import serialization
+from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
+
+from zeitghost.bias import AnalyzedArticle
+from zeitghost.fetcher import Article
+
+
+def _article(url, score=0.5):
+    return AnalyzedArticle(
+        original=Article(title="T", url=url, summary="S", source_name="AP",
+                         published="2026-05-12T00:00:00+00:00",
+                         categories=["politics"]),
+        bias_score=score, bias_label="center",
+        variant_left_title="L", variant_left_summary="L sum",
+        variant_right_title="R", variant_right_summary="R sum",
+    )
+
+
+def _pubkey(seed: bytes) -> bytes:
+    return (Ed25519PrivateKey.from_private_bytes(seed).public_key()
+            .public_bytes(encoding=serialization.Encoding.Raw,
+                         format=serialization.PublicFormat.Raw))
+
+
+# --- #5: bias_score skip-not-default --------------------------------------
+
+def test_parse_bias_score_skips_missing_not_defaults():
+    """A missing or non-numeric score yields None (caller skips) — never a
+    silent 0.5 that would mislabel an unscored article as 'center'."""
+    from zeitghost.bias import _parse_bias_score
+
+    # Present and numeric — passes through, including the valid 0.0 edge.
+    assert _parse_bias_score({"bias_score": 0.0}) == 0.0
+    assert _parse_bias_score({"bias_score": 0.73}) == pytest.approx(0.73)
+    assert _parse_bias_score({"bias_score": "0.7"}) == pytest.approx(0.7)
+    # Absent / null / unparseable — None, so analyze_article returns None.
+    assert _parse_bias_score({}) is None
+    assert _parse_bias_score({"bias_score": None}) is None
+    assert _parse_bias_score({"bias_score": "left-ish"}) is None
+    assert _parse_bias_score({"bias_score": {}}) is None
+
+
+# --- #2: load collapses revision chains to latest --------------------------
+
+def test_load_returns_latest_revision_only(tmp_path):
+    """Two revisions of the same article (chained via parent_shard_id) collapse
+    to a single card on load — the newest one."""
+    from zeitghost.shards import (
+        init_store, article_to_internal_shard, load_articles_from_shards,
+        build_lineage_index, SCOPE_INTERNAL,
+    )
+    store = init_store(tmp_path / "shards")
+    url = "https://e.com/evolving-story"
+
+    a = _article(url, score=0.30)
+    first = article_to_internal_shard(a, store)
+
+    a2 = _article(url, score=0.80)  # re-analysis, same URL → same entity
+    lineage = build_lineage_index(store, SCOPE_INTERNAL)
+    second = article_to_internal_shard(a2, store, lineage_index=lineage)
+    assert second != first
+
+    loaded = load_articles_from_shards(store)
+    assert len(loaded) == 1
+    assert loaded[0].shard_id == second
+    assert loaded[0].parent_shard_id == first
+    assert loaded[0].bias_score == pytest.approx(0.80)
+
+
+def test_load_keeps_distinct_entities_separate(tmp_path):
+    """Collapse is per-entity — two different articles both load."""
+    from zeitghost.shards import (
+        init_store, article_to_internal_shard, load_articles_from_shards,
+    )
+    store = init_store(tmp_path / "shards")
+    article_to_internal_shard(_article("https://e.com/a"), store)
+    article_to_internal_shard(_article("https://e.com/b"), store)
+
+    loaded = load_articles_from_shards(store)
+    assert {a.original.url for a in loaded} == {"https://e.com/a", "https://e.com/b"}
+
+
+# --- #4: opt-in signing ----------------------------------------------------
+
+def test_shard_unsigned_without_seed(tmp_path):
+    from spiritwriter.fabric.store import ShardStore  # noqa: F401 (type clarity)
+    from zeitghost.shards import (
+        init_store, article_to_internal_shard, SCOPE_INTERNAL,
+    )
+    store = init_store(tmp_path / "shards")
+    article_to_internal_shard(_article("https://e.com/unsigned"), store)
+
+    [shard] = list(store.by_scope(SCOPE_INTERNAL))
+    assert shard.signature is None
+    assert shard.created_by is None
+
+
+def test_shard_signed_with_seed_verifies_and_round_trips(tmp_path):
+    """A signed shard persists its signature + created_by, and the signature
+    verifies against the seed's public key after a store round-trip."""
+    from spiritwriter.fabric.shard import pubkey_thumbprint
+    from zeitghost.shards import (
+        init_store, article_to_internal_shard, article_to_sw_shard,
+        SCOPE_INTERNAL, SCOPE_SW_ARTICLE,
+    )
+    seed = os.urandom(32)
+    pub = _pubkey(seed)
+    store = init_store(tmp_path / "shards")
+
+    article_to_internal_shard(_article("https://e.com/signed"), store,
+                              signing_seed=seed)
+    article_to_sw_shard(_article("https://e.com/signed"), store,
+                        signing_seed=seed)
+
+    for scope in (SCOPE_INTERNAL, SCOPE_SW_ARTICLE):
+        [shard] = list(store.by_scope(scope))
+        assert shard.signature is not None
+        assert shard.created_by == pubkey_thumbprint(pub)
+        # verify() raises on a bad signature; True means the chain holds.
+        assert shard.verify(pub) is True
+
+
+def test_tampered_signature_fails_verify(tmp_path):
+    """Flipping a byte of the signature must make verify() reject it — guards
+    against a future regression that signs the wrong payload."""
+    from cryptography.exceptions import InvalidSignature
+    from zeitghost.shards import (
+        init_store, article_to_internal_shard, SCOPE_INTERNAL,
+    )
+    seed = os.urandom(32)
+    pub = _pubkey(seed)
+    store = init_store(tmp_path / "shards")
+    article_to_internal_shard(_article("https://e.com/tamper"), store,
+                              signing_seed=seed)
+
+    [shard] = list(store.by_scope(SCOPE_INTERNAL))
+    # Corrupt one hex nibble of the signature (wraps so 'f' stays valid hex).
+    sig = shard.signature
+    flipped = ("e" if sig[0] != "e" else "d") + sig[1:]
+    shard.signature = flipped
+    with pytest.raises(InvalidSignature):
+        shard.verify(pub)
+
+
+# --- #5 end-to-end: analyze_article drops an unscored article --------------
+
+def test_analyze_article_returns_none_when_bias_score_missing(monkeypatch):
+    """End-to-end: when the LLM response omits bias_score, analyze_article
+    returns None (skip) rather than constructing a default-0.5 article."""
+    import asyncio
+    import zeitghost.bias as bias
+
+    class _FakeProvider:
+        async def query(self, prompt, model=None):
+            # Valid JSON, variants present, but NO bias_score key.
+            return ('{"bias_label": "center", '
+                    '"variant_left": {"title": "L", "summary": "ls"}, '
+                    '"variant_right": {"title": "R", "summary": "rs"}}')
+
+    monkeypatch.setattr(bias, "_get_provider", lambda: _FakeProvider())
+
+    art = Article(title="T", url="https://e.com/no-score", summary="s",
+                  source_name="src", published="2026-05-12T00:00:00+00:00")
+    assert asyncio.run(bias.analyze_article(art)) is None
+
+
+# --- resolve_signing_seed --------------------------------------------------
+
+def test_resolve_signing_seed_from_env(monkeypatch):
+    from zeitghost.shards import resolve_signing_seed, SIGNING_KEY_NAME
+    seed = os.urandom(32)
+    monkeypatch.setenv(SIGNING_KEY_NAME, seed.hex())
+    assert resolve_signing_seed() == seed
+
+
+def test_resolve_signing_seed_absent_is_none(monkeypatch):
+    from zeitghost.shards import resolve_signing_seed, SIGNING_KEY_NAME
+    monkeypatch.delenv(SIGNING_KEY_NAME, raising=False)
+    # No env var (and the test keychain won't have this key) → opt-out.
+    assert resolve_signing_seed() is None
+
+
+def test_resolve_signing_seed_malformed_is_none(monkeypatch):
+    """A fat-fingered key degrades to unsigned rather than crashing ingest."""
+    from zeitghost.shards import resolve_signing_seed, SIGNING_KEY_NAME
+    monkeypatch.setenv(SIGNING_KEY_NAME, "not-hex-at-all")
+    assert resolve_signing_seed() is None
+    monkeypatch.setenv(SIGNING_KEY_NAME, "ab")  # valid hex, but 1 byte ≠ 32
+    assert resolve_signing_seed() is None
+
+
+# --- signing_required (prod fail-closed switch) ----------------------------
+
+def test_signing_required_off_by_default(monkeypatch):
+    from zeitghost.shards import signing_required
+    monkeypatch.delenv("ZEITGHOST_REQUIRE_SIGNING", raising=False)
+    assert signing_required() is False
+    assert signing_required(flag=False) is False
+
+
+def test_signing_required_flag_wins(monkeypatch):
+    from zeitghost.shards import signing_required
+    monkeypatch.delenv("ZEITGHOST_REQUIRE_SIGNING", raising=False)
+    assert signing_required(flag=True) is True
+
+
+def test_signing_required_env_truthy_variants(monkeypatch):
+    from zeitghost.shards import signing_required
+    for truthy in ("1", "true", "TRUE", "yes", "on"):
+        monkeypatch.setenv("ZEITGHOST_REQUIRE_SIGNING", truthy)
+        assert signing_required() is True, truthy
+    for falsy in ("0", "false", "no", "", "off"):
+        monkeypatch.setenv("ZEITGHOST_REQUIRE_SIGNING", falsy)
+        assert signing_required() is False, falsy

tests/test_smoke.py

@@ -474,10 +474,15 @@ def test_shard_metadata_round_trip(tmp_path):
     shard_id_2 = article_to_internal_shard(a, store, lineage_index=lineage)
     assert shard_id_2 != shard_id_1
 
-    # Loading now returns both revisions; the newer one points at the older
+    # Loading collapses the revision chain to the latest shard per entity, so
+    # the re-analyzed article renders as ONE card (its newest revision), not
+    # two — while still carrying the parent link back to the prior shard.
     revisions = load_articles_from_shards(store)
-    by_id = {r.shard_id: r for r in revisions}
-    assert by_id[shard_id_2].parent_shard_id == shard_id_1
+    assert len(revisions) == 1
+    latest = revisions[0]
+    assert latest.shard_id == shard_id_2
+    assert latest.parent_shard_id == shard_id_1
+    assert latest.bias_score == pytest.approx(0.74)
 
 
 def test_legacy_dump_helpers_smoke():

zeitghost/bias.py

@@ -172,6 +172,24 @@ def bias_lean_display(score: float, center_tolerance: float = 0.025) -> str:
 """
 
 
+def _parse_bias_score(data: dict) -> float | None:
+    """Bias score from the LLM JSON, or None if absent/non-numeric.
+
+    Returns None rather than defaulting to 0.5 — an article we couldn't score
+    must be *skipped*, never silently published as "center" (robustness
+    invariant #1: skip or guard, never default-fill). A literal 0.0 (full
+    left) is a valid score and passes through; only a missing or unparseable
+    value yields None.
+    """
+    raw = data.get("bias_score")
+    if raw is None:
+        return None
+    try:
+        return float(raw)
+    except (TypeError, ValueError):
+        return None
+
+
 def _extract_json(text: str) -> dict | None:
     """Extract first valid JSON object from LLM response.
 
@@ -201,8 +219,14 @@ def _extract_json(text: str) -> dict | None:
         return None
 
 
-async def analyze_article(article: Article) -> AnalyzedArticle | None:
-    """Analyze one article — compute bias and generate both variants."""
+async def analyze_article(article: Article,
+                          stats: dict | None = None) -> AnalyzedArticle | None:
+    """Analyze one article — compute bias and generate both variants.
+
+    Pass a mutable `stats` dict to tally why articles are dropped — currently
+    the `no_score` reason (skipped for a missing/invalid bias_score). Lets
+    `analyze_batch` surface the count so a silent feed-drop is noticeable.
+    """
     provider = _get_provider()
     # Prefer the trafilatura-extracted body when present (richer context for
     # Claude); fall back to NewsAPI's terse description if body fetch failed.
@@ -221,11 +245,21 @@ async def analyze_article(article: Article) -> AnalyzedArticle | None:
             log.debug("Raw response: %s", response[:500])
             return None
 
+        bias_score = _parse_bias_score(data)
+        if bias_score is None:
+            # No usable score → skip rather than default-fill to 0.5, which
+            # would mislabel an unscored article as "center" (invariant #1).
+            log.warning("Missing/invalid bias_score for '%s' — skipping",
+                        article.title[:50])
+            if stats is not None:
+                stats["no_score"] = stats.get("no_score", 0) + 1
+            return None
+
         left = data.get("variant_left", {}) or {}
         right = data.get("variant_right", {}) or {}
         return AnalyzedArticle(
             original=article,
-            bias_score=float(data.get("bias_score", 0.5)),
+            bias_score=bias_score,
             bias_label=data.get("bias_label", "center"),
             variant_left_title=left.get("title", article.title),
             variant_left_summary=left.get("summary", article.summary),
@@ -242,11 +276,23 @@ async def analyze_article(article: Article) -> AnalyzedArticle | None:
 
 
 async def analyze_batch(articles: list[Article]) -> list[AnalyzedArticle]:
-    """Analyze a batch of articles. Skips any that fail to parse."""
+    """Analyze a batch of articles. Skips any that fail to parse.
+
+    Logs a WARNING with the count of articles dropped for a missing/invalid
+    bias_score — that path silently shrinks the feed (by design, vs. the old
+    default-fill), so the count surfaces an LLM regression that starts dropping
+    a chunk of articles. Logging propagates to the CLI's RichHandler, so it
+    shows on the operator's console during `zeitghost ingest`.
+    """
     results = []
+    stats: dict[str, int] = {}
     for article in articles:
-        analyzed = await analyze_article(article)
+        analyzed = await analyze_article(article, stats=stats)
         if analyzed is not None:
             results.append(analyzed)
     log.info("Analyzed %d articles, %d succeeded", len(articles), len(results))
+    n_unscored = stats.get("no_score", 0)
+    if n_unscored:
+        log.warning("%d article(s) skipped — no usable bias_score in the LLM "
+                    "response (not default-filled to center)", n_unscored)
     return results