feat: version-lineage report in the crawler ingestion pipeline

When a re-crawl detects a real content delta (a just-ingested root that
supersedes a prior version — content hash changed, not redeploy/ETag
noise the idempotent ingest already no-op'd), the pipeline now surfaces
the page's document chain over time instead of just 'something changed'.

bridge.py: version_chain(conn, uri) walks a URI's documents by ingest_ts
(each content change = new content-addressed doc + supersedes edge);
delta_report() adds the word-level similarity of the latest change;
render_delta_report() prints it. ingest_crawled() detects superseding
roots, emits the lineage report to stderr per changed page, and returns
'deltas' in its summary. Validated on the live russell.ballestrini.net
re-crawl: 223 pages, full redeploy, exactly 1 content change (/about/),
rendered as a 2-version chain (90% similar to prior). 2 tests; suite
2551 passed.

arborist/sources/crawler/bridge.py

@@ -283,6 +283,91 @@ class _CrawledHtmlSource:
                 yield doc
 
 
+# --------------------------------------------------------- version lineage
+
+def _chunk_text(conn, root: str) -> str:
+    """Reassemble a document's text from its content-addressed chunks."""
+    parts: list[str] = []
+    for (blob,) in conn.execute(
+        "SELECT content FROM chunks WHERE document_root=? ORDER BY idx", (root,)
+    ):
+        if blob is None:
+            continue
+        try:
+            from arborist.compress import unpack_chunk
+            parts.append(unpack_chunk(blob))
+        except Exception:  # noqa: BLE001 — plain-text fallback
+            parts.append(
+                blob.decode("utf-8", "replace")
+                if isinstance(blob, (bytes, bytearray)) else str(blob)
+            )
+    return "".join(parts)
+
+
+def version_chain(conn, uri: str) -> list[dict]:
+    """Every document version of a URI over time, oldest → newest.
+
+    Each content change writes a new content-addressed document plus a
+    ``supersedes`` edge (lossless history per the idempotent-ingest
+    invariant), so a URI accumulates a chain of versions. Ordered by
+    ``ingest_ts``."""
+    rows = conn.execute(
+        "SELECT document_root, ingest_ts FROM documents "
+        "WHERE document_uri = ? ORDER BY ingest_ts ASC, document_root ASC",
+        (uri,),
+    ).fetchall()
+    chain: list[dict] = []
+    for r in rows:
+        n = conn.execute(
+            "SELECT count(*) FROM chunks WHERE document_root = ?",
+            (r["document_root"],),
+        ).fetchone()[0]
+        chain.append(
+            {"document_root": r["document_root"],
+             "ingest_ts": r["ingest_ts"], "n_chunks": n}
+        )
+    return chain
+
+
+def delta_report(conn, uri: str, *, max_versions: int = 12) -> dict:
+    """Lineage report for a URI: its document chain over time plus the
+    word-level similarity of the most recent content change."""
+    chain = version_chain(conn, uri)
+    rep: dict = {
+        "uri": uri,
+        "n_versions": len(chain),
+        "versions": chain[-max_versions:],
+    }
+    if len(chain) >= 2:
+        import difflib
+        prev = _chunk_text(conn, chain[-2]["document_root"]).split()
+        cur = _chunk_text(conn, chain[-1]["document_root"]).split()
+        rep["latest_delta"] = {
+            "from_root": chain[-2]["document_root"][:12],
+            "to_root": chain[-1]["document_root"][:12],
+            "similarity": round(
+                difflib.SequenceMatcher(None, prev, cur).ratio(), 3),
+        }
+    return rep
+
+
+def render_delta_report(rep: dict) -> str:
+    """One-paragraph human view of a ``delta_report`` for the crawl log."""
+    import datetime as _dt
+    out = [f"  Δ {rep['uri']} — content changed; "
+           f"{rep['n_versions']} version(s) over time:"]
+    ld = rep.get("latest_delta")
+    for i, v in enumerate(rep["versions"], 1):
+        ts = _dt.datetime.fromtimestamp(
+            v["ingest_ts"], _dt.timezone.utc).strftime("%Y-%m-%d")
+        tail = ""
+        if ld and v["document_root"].startswith(ld["to_root"]):
+            tail = f"   ← {int(ld['similarity'] * 100)}% similar to prior"
+        out.append(f"      v{i} {ts}  {v['document_root'][:12]}…  "
+                   f"{v['n_chunks']} chunks{tail}")
+    return "\n".join(out)
+
+
 def ingest_crawled(
     conn,
     urls: Iterable[str],
@@ -312,6 +397,7 @@ def ingest_crawled(
             "inserted": stats.inserted,
             "http_meta_written": 0,
             "documents": [],
+            "deltas": [],
         }
 
     # Map document_uri → document_root via the documents table (URI is
@@ -351,11 +437,34 @@ def ingest_crawled(
                 }
             )
 
+    # Lineage: a just-ingested root that supersedes a prior version is a
+    # real content delta (the content hash changed — not redeploy/ETag
+    # noise, which the idempotent ingest already no-op'd). Surface the
+    # page's document chain over time so a re-crawl reports WHAT changed,
+    # not merely that it did.
+    import sys as _sys
+    deltas: list[dict] = []
+    seen_uris: set[str] = set()
+    for d in written:
+        if d["document_uri"] in seen_uris:
+            continue
+        superseded = conn.execute(
+            "SELECT 1 FROM edges WHERE src_root = ? "
+            "AND edge_type = 'supersedes' LIMIT 1",
+            (d["document_root"],),
+        ).fetchone()
+        if superseded:
+            seen_uris.add(d["document_uri"])
+            rep = delta_report(conn, d["document_uri"])
+            deltas.append(rep)
+            print(render_delta_report(rep), file=_sys.stderr)
+
     return {
         "seen": stats.seen,
         "inserted": stats.inserted,
         "http_meta_written": len(written),
         "documents": written,
+        "deltas": deltas,
     }
 
 

tests/test_version_lineage.py

@@ -0,0 +1,73 @@
+"""Version lineage in the crawler ingestion pipeline (2026-05-21).
+
+A content change to a URI writes a new content-addressed document + a
+``supersedes`` edge, so a URI accumulates a chain of versions over time.
+A redeploy that doesn't change content is a no-op (same Merkle root) and
+adds NO version. The crawler emits a lineage report when it detects a
+real delta. See arborist/sources/crawler/bridge.py.
+"""
+from __future__ import annotations
+
+from typing import Iterator
+
+from arborist.document import Document
+from arborist.ingest import ingest_source
+from arborist.source import Source
+from arborist.sources.crawler.bridge import (
+    delta_report,
+    render_delta_report,
+    version_chain,
+)
+from arborist.store import connect
+
+
+class FakeSource(Source):
+    source_type = "fake"
+
+    def __init__(self, docs):
+        self.docs = docs
+
+    def iter_documents(self) -> Iterator[Document]:
+        yield from self.docs
+
+
+def _doc(uri: str, content: str) -> Document:
+    return Document(uri=uri, content=content, source_type="fake",
+                    title="page", edges=[])
+
+
+def test_content_change_builds_a_version_chain(tmp_path):
+    conn = connect(tmp_path / "t.db")
+    uri = "test://page"
+    ingest_source(conn, FakeSource([_doc(uri, "alpha bravo charlie delta echo")]))
+    # changed content, same URI -> new root + supersedes edge
+    ingest_source(conn, FakeSource(
+        [_doc(uri, "alpha bravo charlie delta ECHO foxtrot golf")]))
+
+    chain = version_chain(conn, uri)
+    assert len(chain) == 2  # two versions over time
+
+    rep = delta_report(conn, uri)
+    assert rep["n_versions"] == 2
+    assert "latest_delta" in rep
+    assert 0.0 < rep["latest_delta"]["similarity"] < 1.0  # changed, not total
+
+    txt = render_delta_report(rep)
+    assert "content changed" in txt
+    assert "v1" in txt and "v2" in txt
+    assert "similar to prior" in txt
+
+
+def test_identical_recrawl_adds_no_version(tmp_path):
+    # A redeploy with byte-identical content must NOT create a new version
+    # (content-addressed idempotence — the whole point).
+    conn = connect(tmp_path / "t.db")
+    uri = "test://stable"
+    ingest_source(conn, FakeSource([_doc(uri, "same content stays the same")]))
+    ingest_source(conn, FakeSource([_doc(uri, "same content stays the same")]))
+
+    chain = version_chain(conn, uri)
+    assert len(chain) == 1  # no-op re-ingest -> still one version
+    rep = delta_report(conn, uri)
+    assert rep["n_versions"] == 1
+    assert "latest_delta" not in rep