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).
This commit is contained in:
russell@unturf.com 2026-04-27 07:56:49 -04:00
parent 3e60bef516
commit cb3ab5ae83
No known key found for this signature in database
2 changed files with 137 additions and 0 deletions

View file

@ -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,
)

View file

@ -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()