From a28345e3f7acc5bd052c826aa35e84072199bc01 Mon Sep 17 00:00:00 2001 From: Shashank Shekhar Singh Date: Sat, 26 Sep 2026 15:51:53 +0530 Subject: [PATCH] memory: an imported retraction stays retracted on the LadybugDB backend MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Closes #110, taking option 3 of the three the issue costed — persist `superseded_by` as a column and keep the edge derived from it. `superseded_by` was reconstructed from the SUPERSEDED_BY edge, and `add()` cannot create an edge towards a claim the store does not have yet. In the natural import order it does not have it: `all_claims()` returns oldest-first, so a superseded claim arrives before the claim that superseded it. The edge was skipped while `superseded_at` was written, so the same row said "retracted" and `superseded_by IS NULL` said "current" — `Claim.is_current` believed the second, and `current("svc")` returned both sides of a correction. Two of three backends agreed; this one contradicted itself. `supersede()` was never affected because it writes the edge itself, which is why every test that went through it passed. The column is what reads project now, and `current()` filters on it rather than on an OPTIONAL MATCH, which is both tidier and the actual fix — filtering on the edge is what returned a retracted claim as live. The edge remains, because it is the provenance chain this backend exists for and `cypher()` walks it. `_write` reconciles it in both directions on every write: forward when this claim names a superseder that is present, backward when a stored claim names *this* one and could not have an edge until now. So the edge is complete once both ends have arrived, in either order, and a stale one is dropped first — `add` is an upsert, for the same reason the entity edges are rebuilt rather than merged. Option 1 (fail closed) was rejected because it breaks replaying `all_claims()` in its own returned order, and option 2 (a stub node) because the stub reads back as a ValidationError and has no `seq`, which is what `test_add_is_an_upsert_that_keeps_insertion_order` is about. The cost option 3 was costed at is the schema change, and it is handled rather than assumed: `CREATE NODE TABLE IF NOT EXISTS` does not alter an existing table, so `_migrate_superseded_by` adds the column and backfills it from the edges an older database does have. Verified against a database built with the driver in the shape the old code created, not a checked-in fixture: the link is recovered, `seq` does not restart, and a second open is a no-op. A driver that cannot add the column raises with an explanation instead of writing rows that are wrong. The module docstring said `superseded_by` is "an *edge* rather than a foreign key in a column". It is both now, and says so. Verified: 2197 selected, 13 deselected, ruff clean. Five of the eight new tests are red without the fix; the other three guard the mechanism it adds. Co-Authored-By: Claude Opus 5 (1M context) --- docs/deep-dive.md | 2 +- grapharc/memory/ladybug_store.py | 126 ++++++++++++++++++---- tests/test_ladybug_store.py | 172 ++++++++++++++++++++++++++++++- 3 files changed, 278 insertions(+), 22 deletions(-) diff --git a/docs/deep-dive.md b/docs/deep-dive.md index b1328c2..6ade549 100644 --- a/docs/deep-dive.md +++ b/docs/deep-dive.md @@ -254,7 +254,7 @@ A stable system is not one that claims to have no edges — it is one whose edge - **`.env` and `grapharc.toml` follow the same discovery rule: the working directory, and nowhere else.** Neither searches parent directories — a run must not be governed by a file you did not know about, and must not be *billed* to one either. **This is a behaviour change:** the credential loader used to walk up to `/`, so a `.env` in an ancestor directory (a `$HOME` one on a shared box, a client project one above a demo checkout) was picked up silently. If you relied on that, move the file into the directory you run from, `export` the variable, or pass `env_file=` to name it explicitly. A real environment variable still beats any file. - **`grapharc run` has no budget unless you give it one.** Set any of `--max-tokens`, `--max-iterations`, `--max-seconds`, or `--max-concurrency`; without them each dimension is unlimited and the gate admits a topology of any worst-case cost. -**Verified this pass:** `pytest` → green, 2,190 selected and 13 deselected (the live ones); `ruff check .` clean; all eight `grapharc demo` stages green, plus the `trace` / `metrics` / `viz` / `replay` tour against a freshly recorded demo trace; the wheel builds and imports all submodules in a clean virtualenv with `[all]`, and `0.1.8` on PyPI is that wheel. The counts are a snapshot, not a property of the project — `pytest` re-derives them in one command, which is the only reason they are quoted, and `tests/test_deep_dive.py` fails this line rather than letting it drift. +**Verified this pass:** `pytest` → green, 2,197 selected and 13 deselected (the live ones); `ruff check .` clean; all eight `grapharc demo` stages green, plus the `trace` / `metrics` / `viz` / `replay` tour against a freshly recorded demo trace; the wheel builds and imports all submodules in a clean virtualenv with `[all]`, and `0.1.8` on PyPI is that wheel. The counts are a snapshot, not a property of the project — `pytest` re-derives them in one command, which is the only reason they are quoted, and `tests/test_deep_dive.py` fails this line rather than letting it drift. [ROADMAP.md](../ROADMAP.md) tracks what is built and what is not, item by item. diff --git a/grapharc/memory/ladybug_store.py b/grapharc/memory/ladybug_store.py index df5e9f8..a529421 100644 --- a/grapharc/memory/ladybug_store.py +++ b/grapharc/memory/ladybug_store.py @@ -8,10 +8,10 @@ LadybugDB is an embedded property-graph database with Cypher — a fork of Kuzu, revived in 2025 after Apple acquired and closed it. Embedded means the same deal SQLite offers: a path on disk, no server, no daemon to run. What it adds -over the SQLite backend is that `superseded_by` is an *edge* rather than a -foreign key in a column, and the subject and object of every claim are `Entity` -nodes, so "what did run #12 believe that run #37 corrected" is a path, and you -can ask it in Cypher without going through Python at all:: +over the SQLite backend is that `superseded_by` is an *edge* as well as a +column, and the subject and object of every claim are `Entity` nodes, so "what +did run #12 believe that run #37 corrected" is a path, and you can ask it in +Cypher without going through Python at all:: store = LadybugMemoryStore("memory.lbdb") store.cypher( @@ -72,6 +72,7 @@ run_id STRING, confidence DOUBLE, superseded_at STRING, + superseded_by STRING, subject_norm STRING, predicate_norm STRING, seq INT64 @@ -96,14 +97,26 @@ "CREATE REL TABLE IF NOT EXISTS SUPERSEDED_BY(FROM Claim TO Claim)", ) -# `superseded_by` is not a column — it is reconstructed from the edge, so every -# read pairs its MATCH with this OPTIONAL MATCH and this projection. +# `superseded_by` is stored as a column *and* walkable as an edge, and the +# column is the one reads project. It was edge-only, reconstructed by an +# OPTIONAL MATCH on every read, which reads well and loses data: `add()` cannot +# create an edge to a claim the store does not have yet, and in the natural +# import order it does not have it — `all_claims()` returns oldest-first, so a +# superseded claim arrives before the claim that superseded it. The edge was +# silently skipped while `superseded_at` was written, so the row said "retracted" +# and `superseded_by IS NULL` said "current", and `current()` returned both +# sides of a correction (issue #110). +# +# The edge is still what you walk in Cypher — it is the provenance chain this +# backend exists for, and `_write` reconciles it in both directions so it is +# complete once both ends have arrived, whatever order they arrived in. What +# changed is which of the two survives an import that has only seen one end. _OPTIONAL_SUPERSEDER = "OPTIONAL MATCH (c)-[:SUPERSEDED_BY]->(n:Claim)" _PROJECTION = """ c.id AS id, c.subject AS subject, c.predicate AS predicate, c.object AS object, c.source AS source, c.observed_at AS observed_at, c.run_id AS run_id, c.confidence AS confidence, c.superseded_at AS superseded_at, - n.id AS superseded_by + c.superseded_by AS superseded_by """ _UPSERT = """ @@ -112,11 +125,13 @@ c.subject=$subject, c.predicate=$predicate, c.object=$object, c.source=$source, c.observed_at=$observed_at, c.run_id=$run_id, c.confidence=$confidence, c.superseded_at=$superseded_at, + c.superseded_by=$superseded_by, c.subject_norm=$subject_norm, c.predicate_norm=$predicate_norm, c.seq=$seq ON MATCH SET c.subject=$subject, c.predicate=$predicate, c.object=$object, c.source=$source, c.observed_at=$observed_at, c.run_id=$run_id, c.confidence=$confidence, c.superseded_at=$superseded_at, + c.superseded_by=$superseded_by, c.subject_norm=$subject_norm, c.predicate_norm=$predicate_norm """ @@ -164,6 +179,7 @@ def _params(claim: Claim, seq: int) -> dict[str, Any]: "run_id": claim.run_id, "confidence": float(claim.confidence), "superseded_at": claim.superseded_at, + "superseded_by": claim.superseded_by, "subject_norm": _normalize(claim.subject), "predicate_norm": _normalize(claim.predicate), "seq": seq, @@ -200,8 +216,47 @@ def __init__( if not read_only: for statement in _SCHEMA: self._conn.execute(statement) + self._migrate_superseded_by() self._seq = self._next_seq() + def _migrate_superseded_by(self) -> None: + """Add the `superseded_by` column to a database that predates it. + + `CREATE NODE TABLE IF NOT EXISTS` does not alter an existing table, so a + database written before #110 has every other column and not this one. + Adding it leaves the column NULL on every row, which would read as "no + claim was ever superseded" — so the existing edges are the thing to + trust here, and the backfill copies them into the column. Those edges + were written by `supersede()`, which always created them; it is the + `add()` path that never did, and that path left nothing to recover. + + `ALTER TABLE` raises on a database that already has the column, which is + every database created since. That is the expected outcome, not an + error, so it is swallowed — narrowly, by re-reading the schema + afterwards rather than by assuming. + """ + try: + self._conn.execute("ALTER TABLE Claim ADD superseded_by STRING") + except Exception: + # Either the column is already there (the common case) or the + # driver rejected the statement. The projection below decides which. + pass + try: + self._conn.execute("MATCH (c:Claim) RETURN c.superseded_by LIMIT 1").get_all() + except Exception as exc: # pragma: no cover - a driver too old to alter + raise RuntimeError( + "this LadybugDB database has no `superseded_by` column and it " + "could not be added, so a superseded claim cannot be stored " + "correctly (see issue #110). Re-create the store from " + f"`all_claims()` of a copy, or use SQLiteMemoryStore. Cause: {exc}" + ) from exc + # Backfill from the edges, which are the only record an older database + # has. Idempotent: re-running sets the same ids. + self._conn.execute( + "MATCH (c:Claim)-[:SUPERSEDED_BY]->(n:Claim) " + "WHERE c.superseded_by IS NULL SET c.superseded_by = n.id" + ) + def _next_seq(self) -> int: """Resume the insertion counter where the last process left it.""" rows = self._conn.execute("MATCH (c:Claim) RETURN max(c.seq)").get_all() @@ -239,10 +294,7 @@ def _transaction(self) -> Iterator[None]: self._conn.execute("COMMIT") def _query(self, where: str, params: dict[str, Any], order: str = "c.seq") -> list[Claim]: - cypher = ( - f"MATCH (c:Claim) {where} {_OPTIONAL_SUPERSEDER} " - f"RETURN {_PROJECTION} ORDER BY {order}" - ) + cypher = f"MATCH (c:Claim) {where} RETURN {_PROJECTION} ORDER BY {order}" with self._lock: result = self._conn.execute(cypher, params) return [_to_claim(row) for row in result.rows_as_dict()] @@ -275,6 +327,40 @@ def _write(self, claim: Claim) -> None: f"MERGE (c)-[:{rel}]->(e)", {"id": claim.id, "name": name}, ) + self._reconcile_supersession(claim.id) + + def _reconcile_supersession(self, claim_id: str) -> None: + """Make the SUPERSEDED_BY edges agree with the columns, both ways round. + + Called for every write, because a claim can arrive at either end of a + correction first and the edge needs both ends to exist: + + - *forward*: this claim names a superseder. If that claim is present the + edge is created; if it is not, the column still records it and this + runs again when the superseder arrives. + - *backward*: an already-stored claim names **this** one as its + superseder, and could not have an edge until now. + + The stale edge is dropped first, because `add` is an upsert: re-adding + an id whose `superseded_by` changed must not leave the old edge behind, + for the same reason the entity edges above are rebuilt rather than + merged. + """ + self._conn.execute( + "MATCH (c:Claim {id: $id})-[r:SUPERSEDED_BY]->(n:Claim) " + "WHERE n.id <> coalesce(c.superseded_by, '') DELETE r", + {"id": claim_id}, + ) + self._conn.execute( + "MATCH (c:Claim {id: $id}), (n:Claim) WHERE c.superseded_by = n.id " + "MERGE (c)-[:SUPERSEDED_BY]->(n)", + {"id": claim_id}, + ) + self._conn.execute( + "MATCH (c:Claim), (n:Claim {id: $id}) WHERE c.superseded_by = n.id " + "MERGE (c)-[:SUPERSEDED_BY]->(n)", + {"id": claim_id}, + ) def add(self, claim: Claim) -> Claim: with self._transaction(): @@ -301,8 +387,9 @@ def supersede(self, old_id: str, new_claim: Claim) -> Claim: {"old": old_id, "new": new_claim.id}, ) self._conn.execute( - "MATCH (o:Claim {id: $old}) SET o.superseded_at = $at", - {"old": old_id, "at": _now()}, + "MATCH (o:Claim {id: $old}) " + "SET o.superseded_at = $at, o.superseded_by = $new", + {"old": old_id, "at": _now(), "new": new_claim.id}, ) return new_claim @@ -313,13 +400,12 @@ def current(self, subject: str, predicate: str | None = None) -> list[Claim]: if predicate is not None: where += " AND c.predicate_norm = $predicate" params["predicate"] = _normalize(predicate) - # The superseded test is on the edge, so it has to follow the OPTIONAL - # MATCH rather than ride along in the WHERE above. - cypher = ( - f"MATCH (c:Claim) {where} {_OPTIONAL_SUPERSEDER} " - f"WITH c, n WHERE n IS NULL " - f"RETURN {_PROJECTION} ORDER BY c.seq" - ) + # The superseded test is on the column now, so it rides along in the + # WHERE above instead of needing a WITH after an OPTIONAL MATCH. That is + # not only tidier: filtering on the edge is what returned both sides of + # an imported correction, because the edge was the half that got lost. + where += " AND c.superseded_by IS NULL" + cypher = f"MATCH (c:Claim) {where} RETURN {_PROJECTION} ORDER BY c.seq" with self._lock: result = self._conn.execute(cypher, params) return [_to_claim(row) for row in result.rows_as_dict()] diff --git a/tests/test_ladybug_store.py b/tests/test_ladybug_store.py index d55d501..7373344 100644 --- a/tests/test_ladybug_store.py +++ b/tests/test_ladybug_store.py @@ -27,7 +27,7 @@ import pytest -from grapharc.memory import Claim, LadybugMemoryStore, SQLiteMemoryStore +from grapharc.memory import Claim, LadybugMemoryStore, MemoryStore, SQLiteMemoryStore from grapharc.memory.index import ClaimIndex from grapharc.memory.ladybug_store import _load_driver from grapharc.memory.retrieval import search @@ -449,3 +449,173 @@ def _run_child(tmp_path: Path, name: str, body: str, db: Path, *args: str) -> st if line.startswith("RESULT "): return line[len("RESULT ") :] raise AssertionError(f"child emitted no result:\nstdout={proc.stdout}\n{proc.stderr}") + + +# -------------------------------------------------------------------------- +# An already-superseded claim, arriving through add() (#110) +# -------------------------------------------------------------------------- +# +# `supersede()` creates the SUPERSEDED_BY edge itself, so every path that goes +# through it agreed with the other backends and the comparison test above +# passed. `add()` never created the edge, and `superseded_by` was read off it — +# so a claim that arrived *already superseded* landed with `superseded_at` set +# and `superseded_by` NULL, which `Claim.is_current` reads as current. The store +# then contradicted itself, and `current()` returned both sides of a correction. +# +# That is every import, replay, backup restore, and copy between backends. + + +def _corrected_pair() -> list[Claim]: + """A claim and its correction, as an export would carry them: oldest first, + the older one already naming its superseder.""" + source = SQLiteMemoryStore(":memory:") + source.add(Claim(id="a", subject="svc", predicate="owner", object="alice", source="t")) + source.supersede( + "a", Claim(id="b", subject="svc", predicate="owner", object="bob", source="t") + ) + return [source.get("a"), source.get("b")] + + +def test_an_imported_retraction_does_not_come_back_as_current(store): + """The bug: `current('svc')` answered alice *and* bob.""" + for claim in _corrected_pair(): + store.add(claim) + + retracted = store.get("a") + assert retracted.superseded_by == "b" + assert retracted.is_current is False + assert [c.id for c in store.current("svc")] == ["b"] + + +def test_the_import_survives_the_reverse_order_too(store): + """`all_claims()` is oldest-first, so the superseded claim normally arrives + before its superseder — but nothing guarantees the order a caller replays + in, and an edge cannot be created towards a claim that is not there yet.""" + for claim in reversed(_corrected_pair()): + store.add(claim) + + assert store.get("a").superseded_by == "b" + assert [c.id for c in store.current("svc")] == ["b"] + + +def test_no_claim_is_current_while_its_superseded_at_is_set(store): + """The self-contradiction, asserted as the property it is: these two columns + describe the same fact and cannot disagree.""" + for claim in _corrected_pair(): + store.add(claim) + + for claim in store.all_claims(): + if claim.superseded_at is not None: + assert claim.is_current is False, claim.id + for claim in store.current("svc"): + assert claim.superseded_at is None, claim.id + + +def test_the_three_backends_agree_on_an_imported_correction(store, tmp_path): + """The equivalence that broke was the one nothing asked for. Asked now.""" + exported = _corrected_pair() + + sqlite_store = SQLiteMemoryStore(tmp_path / "compare.sqlite") + in_memory = MemoryStore() + for claim in exported: + store.add(claim) + sqlite_store.add(claim) + in_memory.add(claim) + + assert _shape(store.all_claims()) == _shape(sqlite_store.all_claims()) + assert _shape(store.all_claims()) == _shape(in_memory.all_claims()) + assert ( + [c.id for c in store.current("svc")] + == [c.id for c in sqlite_store.current("svc")] + == [c.id for c in in_memory.current("svc")] + ) + + +def test_the_edge_is_still_walkable_after_an_import(store): + """The column is what reads project; the edge is what this backend exists + for. `_write` reconciles it in both directions, so it must be there after an + import in either order — not just after `supersede()`.""" + for claim in reversed(_corrected_pair()): + store.add(claim) + + walked = store.cypher("MATCH (o:Claim)-[:SUPERSEDED_BY]->(n:Claim) RETURN o.id, n.id") + assert walked == [["a", "b"]] + + +def test_an_upsert_that_clears_the_link_removes_the_stale_edge(store): + """`add` is an upsert, and the entity edges are rebuilt for this same + reason: re-adding an id whose `superseded_by` changed must not leave the + old edge pointing where it used to.""" + for claim in _corrected_pair(): + store.add(claim) + store.add(store.get("a").model_copy(update={"superseded_by": None})) + + assert store.get("a").superseded_by is None + assert store.get("a").is_current is True + assert store.cypher("MATCH (:Claim)-[:SUPERSEDED_BY]->(:Claim) RETURN 1") == [] + + +def test_a_database_written_before_the_column_existed_is_migrated(driver, tmp_path): + """`CREATE NODE TABLE IF NOT EXISTS` does not alter an existing table, so a + database written before #110 has every column but `superseded_by`. + + Adding it leaves it NULL on every row, which would read as "nothing was ever + superseded" — so the migration backfills from the SUPERSEDED_BY edges, which + are what an older database does have: `supersede()` always wrote them. (The + `add()` path never did, and that is the data #110 lost for good; this + recovers what was recoverable.) + + The old schema is written here with the driver directly, rather than by + checking in a fixture database, so this tests the migration against the + shape the code actually used to create. + """ + path = tmp_path / "pre-110.lbdb" + database = driver.Database(str(path)) + connection = driver.Connection(database) + connection.execute( + """CREATE NODE TABLE Claim( + id STRING PRIMARY KEY, subject STRING, predicate STRING, object STRING, + source STRING, observed_at STRING, run_id STRING, confidence DOUBLE, + superseded_at STRING, subject_norm STRING, predicate_norm STRING, seq INT64)""" + ) + connection.execute("CREATE NODE TABLE Entity(name STRING PRIMARY KEY, display STRING)") + for statement in ( + "CREATE REL TABLE ABOUT(FROM Claim TO Entity)", + "CREATE REL TABLE MENTIONS(FROM Claim TO Entity)", + "CREATE REL TABLE SUPERSEDED_BY(FROM Claim TO Claim)", + ): + connection.execute(statement) + for claim_id, obj, superseded_at, seq in ( + ("a", "alice", "2026-01-01T00:00:00+00:00", 0), + ("b", "bob", None, 1), + ): + connection.execute( + "CREATE (c:Claim {id: $id, subject: 'svc', predicate: 'owner', object: $obj, " + "source: 't', observed_at: '2026-01-01T00:00:00+00:00', run_id: NULL, " + "confidence: 1.0, superseded_at: $at, subject_norm: 'svc', " + "predicate_norm: 'owner', seq: $seq})", + {"id": claim_id, "obj": obj, "at": superseded_at, "seq": seq}, + ) + connection.execute( + "MATCH (o:Claim {id: 'a'}), (n:Claim {id: 'b'}) CREATE (o)-[:SUPERSEDED_BY]->(n)" + ) + connection.close() + database.close() + + store = LadybugMemoryStore(path) + try: + assert store.get("a").superseded_by == "b" + assert store.get("a").is_current is False + assert [c.id for c in store.current("svc")] == ["b"] + # The insertion counter must not restart and collide with existing rows. + assert store._seq == 2 + finally: + store.close() + + # Opening it again re-runs the migration, which must be a no-op rather than + # an error: `ALTER TABLE` on a column that now exists is the common case. + reopened = LadybugMemoryStore(path) + try: + assert reopened.get("a").superseded_by == "b" + finally: + reopened.close()