From cb3ab5ae8312b8a6d6c40b91b5dfb874f803a1bb Mon Sep 17 00:00:00 2001 From: "russell@unturf.com" Date: Mon, 27 Apr 2026 07:56:49 -0400 Subject: [PATCH] versioned re-ingest: same URI + changed content writes a 'supersedes' edge When a re-ingest produces a different document_root for an already-seen URI, both versions now coexist in the store. A 'supersedes' edge from the new doc to the prior one keeps the lineage addressable, and the ingest audit body records the supersedes link. Idempotent re-ingest is unchanged: identical bytes -> identical root -> no new doc, no supersedes edge. Three-version chains test that v3->v2 and v2->v1 edges form a walkable history. The old version stays queryable (its chunks may later be evicted to cold for storage savings while remaining provable). --- aborist/ingest.py | 21 ++++++ tests/test_versioned_ingest.py | 116 +++++++++++++++++++++++++++++++++ 2 files changed, 137 insertions(+) create mode 100644 tests/test_versioned_ingest.py diff --git a/aborist/ingest.py b/aborist/ingest.py index e1cb3e0..69b86d9 100644 --- a/aborist/ingest.py +++ b/aborist/ingest.py @@ -81,6 +81,17 @@ def _ingest_document( _upsert_edges(conn, document_root, doc) return False + # Versioned re-ingest: if a prior doc exists for this URI with a + # DIFFERENT root, the source content has changed since last ingest. + # Keep the prior version (lossless history); link the new doc back + # to it via a 'supersedes' edge so the lineage stays addressable. + prior = conn.execute( + "SELECT document_root FROM documents " + "WHERE document_uri = ? AND document_root != ? " + "ORDER BY ingest_ts DESC LIMIT 1", + (doc.uri, document_root), + ).fetchone() + ingest_ts = int(time.time()) conn.execute( "INSERT INTO documents " @@ -130,6 +141,15 @@ def _ingest_document( _upsert_edges(conn, document_root, doc) + # If we found a prior version for this URI, link new -> old. + if prior is not None: + conn.execute( + "INSERT OR IGNORE INTO edges " + "(src_root, dst_root, dst_uri, edge_type, anchor) " + "VALUES (?, ?, ?, 'supersedes', '')", + (document_root, prior["document_root"], doc.uri), + ) + append_audit( conn, event_type="ingest", @@ -141,6 +161,7 @@ def _ingest_document( "chunking_version": chunker.name, "canonicalization_version": CANONICALIZATION_VERSION, "schema_version": SCHEMA_VERSION, + "supersedes": prior["document_root"] if prior is not None else None, }, ts=ingest_ts, ) diff --git a/tests/test_versioned_ingest.py b/tests/test_versioned_ingest.py new file mode 100644 index 0000000..7c1ce79 --- /dev/null +++ b/tests/test_versioned_ingest.py @@ -0,0 +1,116 @@ +"""Versioned re-ingest: same URI, changed content -> 'supersedes' edge.""" + +from __future__ import annotations + +import time +from typing import Iterator + +from aborist.document import Document +from aborist.ingest import ingest_source +from aborist.source import Source +from aborist.store import connect + + +class FakeSource(Source): + source_type = "test" + + def __init__(self, docs: list[Document]): + 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="test", title=uri) + + +def test_unchanged_reingest_is_idempotent(tmp_path): + db = tmp_path / "idem.db" + conn = connect(db) + try: + d = _doc("test://stable", "same content same content same content " * 30) + ingest_source(conn, FakeSource([d])) + # Re-ingesting identical content = same root = idempotent skip. + ingest_source(conn, FakeSource([d])) + n = conn.execute( + "SELECT COUNT(*) FROM documents WHERE document_uri='test://stable'" + ).fetchone()[0] + assert n == 1 + # No supersedes edge — content unchanged. + n_super = conn.execute( + "SELECT COUNT(*) FROM edges WHERE edge_type='supersedes'" + ).fetchone()[0] + assert n_super == 0 + finally: + conn.close() + + +def test_changed_reingest_creates_supersedes_edge(tmp_path): + db = tmp_path / "v.db" + conn = connect(db) + try: + v1 = _doc("test://changing", "version one content " * 30) + ingest_source(conn, FakeSource([v1])) + time.sleep(1.05) # cross integer second so ingest_ts differs + + v2 = _doc("test://changing", "version two content radically different " * 30) + ingest_source(conn, FakeSource([v2])) + + # Two distinct documents share the URI. + rows = conn.execute( + "SELECT document_root FROM documents WHERE document_uri='test://changing' " + "ORDER BY ingest_ts ASC" + ).fetchall() + assert len(rows) == 2 + old_root = rows[0]["document_root"] + new_root = rows[1]["document_root"] + assert old_root != new_root + + # Supersedes edge: new -> old. + edge = conn.execute( + "SELECT * FROM edges WHERE edge_type='supersedes'" + ).fetchone() + assert edge is not None + assert edge["src_root"] == new_root + assert edge["dst_root"] == old_root + assert edge["dst_uri"] == "test://changing" + + # Audit chain records the supersedes link in the new ingest's body. + from json import loads as _loads + ev = conn.execute( + "SELECT body FROM audit_events WHERE subject_root=?", (new_root,) + ).fetchone() + assert _loads(ev["body"])["supersedes"] == old_root + finally: + conn.close() + + +def test_three_version_chain(tmp_path): + """v1 -> v2 -> v3 produces two supersedes edges in a chain.""" + db = tmp_path / "chain.db" + conn = connect(db) + try: + for i, content in enumerate( + ["alpha alpha alpha " * 30, "beta beta beta " * 30, "gamma gamma gamma " * 30] + ): + ingest_source(conn, FakeSource([_doc("test://multi", content)])) + time.sleep(1.05) + + edges = conn.execute( + "SELECT src_root, dst_root FROM edges WHERE edge_type='supersedes' " + "ORDER BY rowid" + ).fetchall() + assert len(edges) == 2 + # Walk the chain: v3 -> v2, v2 -> v1. + roots = conn.execute( + "SELECT document_root FROM documents WHERE document_uri='test://multi' " + "ORDER BY ingest_ts ASC" + ).fetchall() + v1, v2, v3 = (r["document_root"] for r in roots) + + edge_pairs = {(e["src_root"], e["dst_root"]) for e in edges} + assert (v2, v1) in edge_pairs + assert (v3, v2) in edge_pairs + finally: + conn.close()