From eb9ffba5bdd07c9b4cc891578cab32494bd58e98 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 5 Nov 2026 02:26:21 +0000 Subject: [PATCH 1/5] fix(dedupe): rewrite content_hash with every content-changing merge (#623) merge_duplicate_chunk and merge_existing_chunk_content rewrote chunks.content into a merged-originals body and refreshed dedupe/simhash fields, but left content_hash holding the hash of the pre-merge body. The drain prefers a stored hash, so it rejected every fresh enrichment event for the merged row. The content-changing UPDATE now also sets content_hash = canonical_content_hash(merged_content), inside the existing merge savepoint. No-op merges still omit the column, so replays stay idempotent. Schemas without a content_hash column are unaffected. Adds python -m brainlayer.content_hash_repair: an explicit, dry-run-first, bounded, identity-checked repair for rows the old writer left stale. It refuses the live DB and never creates one; nothing runs it on startup. Refs #623 Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01L3NoZHSCNJgDBeHiBoPu42 --- src/brainlayer/content_hash_repair.py | 152 ++++++++++++++++++++++++++ src/brainlayer/dedupe.py | 19 +++- tests/test_content_hash_repair.py | 141 ++++++++++++++++++++++++ tests/test_dedupe_content_hash.py | 134 +++++++++++++++++++++++ 4 files changed, 445 insertions(+), 1 deletion(-) create mode 100644 src/brainlayer/content_hash_repair.py create mode 100644 tests/test_content_hash_repair.py create mode 100644 tests/test_dedupe_content_hash.py diff --git a/src/brainlayer/content_hash_repair.py b/src/brainlayer/content_hash_repair.py new file mode 100644 index 000000000..3032158af --- /dev/null +++ b/src/brainlayer/content_hash_repair.py @@ -0,0 +1,152 @@ +"""Bounded repair of merged rows whose stored content_hash went stale (#623). + +Before #623 was fixed, ``merge_duplicate_chunk`` and ``merge_existing_chunk_content`` +rewrote ``chunks.content`` into a ``[Merged duplicate originals]`` body without +rewriting ``content_hash``. The drain prefers a stored hash, so every fresh +enrichment event for such a row was rejected as stale, forever. + +This tool rewrites ONLY that column, ONLY on those rows: + +* **Explicit DB.** ``--db`` is required and must already exist. The live + canonical DB (and whatever this process resolves as live) is refused outright; + run it on a copy. Nothing calls this on startup. +* **Dry-run first.** Without ``--apply`` the DB is opened read-only and only the + candidate count and ids are reported. +* **Narrow scope.** A row qualifies only when its content starts with the merge + marker, its ``content_hash`` is non-NULL, and that hash differs from + ``sha256(content.strip())``. NULL hashes and legacy-scheme hashes on ordinary + rows are not this defect and are left alone. +* **Bounded.** At most ``--max-rows`` rows per run, written in ``--batch-size`` + transactions; the report says when more remain. +* **Identity-checked.** Each write is ``WHERE id = ? AND content = ? AND + content_hash = ?`` with the values read at scan time, so a row changed in + between is skipped and counted, never stamped with a hash of other text. +* **Idempotent.** A second run finds nothing to do. + +Content, provenance, FTS rows and vectors are untouched: ``content_hash`` is not +an FTS trigger column, and the embedded text does not change. +""" + +from __future__ import annotations + +import argparse +import json +from pathlib import Path +from typing import Any + +import apsw + +from .chunk_origin_wipe import assert_not_live_db +from .chunk_write import canonical_content_hash +from .dedupe import _MERGED_MARKER + +DEFAULT_BATCH_SIZE = 500 +DEFAULT_MAX_ROWS = 5_000 +SAMPLE_IDS = 10 + + +def _find_candidates(conn: apsw.Connection, *, scan_batch: int, max_rows: int) -> tuple[list[tuple], int, bool]: + candidates: list[tuple[str, str, str, str]] = [] + scanned = 0 + last_rowid = -1 + while len(candidates) < max_rows: + rows = conn.execute( + """ + SELECT rowid, id, content, content_hash FROM chunks + WHERE rowid > ? AND content_hash IS NOT NULL AND substr(content, 1, ?) = ? + ORDER BY rowid LIMIT ? + """, + (last_rowid, len(_MERGED_MARKER), _MERGED_MARKER, scan_batch), + ).fetchall() + if not rows: + return candidates, scanned, False + for rowid, chunk_id, content, stored_hash in rows: + last_rowid = rowid + scanned += 1 + expected = canonical_content_hash(content) + if stored_hash != expected: + candidates.append((str(chunk_id), content, stored_hash, expected)) + if len(candidates) == max_rows: + break + more = ( + conn.execute( + "SELECT 1 FROM chunks WHERE rowid > ? AND content_hash IS NOT NULL AND substr(content, 1, ?) = ? LIMIT 1", + (last_rowid, len(_MERGED_MARKER), _MERGED_MARKER), + ).fetchone() + is not None + ) + return candidates, scanned, more + + +def repair_merged_content_hashes( + db_path: str | Path, + *, + apply: bool = False, + batch_size: int = DEFAULT_BATCH_SIZE, + max_rows: int = DEFAULT_MAX_ROWS, +) -> dict[str, Any]: + if batch_size < 1 or max_rows < 1: + raise ValueError("batch_size and max_rows must be positive") + resolved = assert_not_live_db(Path(db_path)) + if not resolved.is_file(): + raise FileNotFoundError(f"no database at {resolved}; this tool never creates one") + flags = apsw.SQLITE_OPEN_READWRITE if apply else apsw.SQLITE_OPEN_READONLY + conn = apsw.Connection(str(resolved), flags=flags) + try: + conn.setbusytimeout(30_000) + if "content_hash" not in {row[1] for row in conn.execute("PRAGMA table_info(chunks)")}: + raise RuntimeError("chunks.content_hash does not exist; nothing to repair") + candidates, scanned, more = _find_candidates(conn, scan_batch=batch_size, max_rows=max_rows) + repaired = skipped = batches = 0 + if apply: + for start in range(0, len(candidates), batch_size): + conn.execute("BEGIN IMMEDIATE") + try: + for chunk_id, content, stored_hash, expected in candidates[start : start + batch_size]: + conn.execute( + "UPDATE chunks SET content_hash = ? WHERE id = ? AND content = ? AND content_hash = ?", + (expected, chunk_id, content, stored_hash), + ) + if conn.changes(): + repaired += 1 + else: + skipped += 1 + conn.execute("COMMIT") + except BaseException: + conn.execute("ROLLBACK") + raise + batches += 1 + finally: + conn.close() + return { + "mode": "apply" if apply else "dry-run", + "merged_rows_scanned": scanned, + "stale": len(candidates), + "repaired": repaired, + "skipped_identity_changed": skipped, + "batches": batches, + "more_remaining": more, + "sample_ids": [candidate[0] for candidate in candidates[:SAMPLE_IDS]], + } + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser( + prog="python -m brainlayer.content_hash_repair", + description="Repair stale content_hash on merged-duplicate rows (#623). Dry-run unless --apply; " + "refuses the live DB.", + ) + parser.add_argument("--db", type=Path, required=True, help="Path to a COPY of the database") + parser.add_argument("--apply", action="store_true", help="Write the repair (default: dry-run)") + parser.add_argument("--batch-size", type=int, default=DEFAULT_BATCH_SIZE) + parser.add_argument("--max-rows", type=int, default=DEFAULT_MAX_ROWS) + args = parser.parse_args(argv) + result = repair_merged_content_hashes( + args.db.expanduser(), apply=args.apply, batch_size=args.batch_size, max_rows=args.max_rows + ) + print(json.dumps(result, sort_keys=True)) + return 0 + + +if __name__ == "__main__": # pragma: no cover + raise SystemExit(main()) diff --git a/src/brainlayer/dedupe.py b/src/brainlayer/dedupe.py index 29331f21d..2110732f2 100644 --- a/src/brainlayer/dedupe.py +++ b/src/brainlayer/dedupe.py @@ -521,8 +521,14 @@ def _changed_merge_fields( merged_tags: list[str], created_at: str | None, existing_fingerprints: tuple[Any, ...], + has_content_hash: bool, ) -> dict[str, Any]: - """Omit unchanged indexed columns: SQLite UPDATE OF fires even for equal values.""" + """Omit unchanged indexed columns: SQLite UPDATE OF fires even for equal values. + + Every content change also rewrites ``content_hash`` in the same UPDATE (#623): + a stale stored hash made the drain reject fresh enrichment for the merged + body forever. A no-op merge leaves ``content_hash`` exactly as it was. + """ fields = compute_dedupe_fields(merged_content, created_at) fingerprints = { "dedupe_hash": fields.dedupe_hash, @@ -543,6 +549,11 @@ def _changed_merge_fields( content=merged_content, char_count=len(merged_content), ) + if has_content_hash: + # Lazy: chunk_write imports this module. One hash implementation only. + from .chunk_write import canonical_content_hash + + updates["content_hash"] = canonical_content_hash(merged_content) merged_tags_json = json.dumps(merged_tags) if merged_tags else None # Legacy CSV must be rewritten: only valid JSON populates chunk_tags. # Exact representation equality also preserves canonical ordering/formatting. @@ -637,6 +648,10 @@ def _rollback_failed_merge(cursor: Any, savepoint: str, original: Exception) -> return True +def _has_content_hash_column(cursor: Any) -> bool: + return "content_hash" in {row[1] for row in cursor.execute("PRAGMA table_info(chunks)")} + + def _invalidate_merged_vectors(cursor: Any, chunk_id: str) -> None: """Remove stale vectors within the merge savepoint.""" for table in ("chunk_vectors", "chunk_vectors_binary"): @@ -702,6 +717,7 @@ def merge_duplicate_chunk( merged_tags=merged_tags, created_at=existing_created, existing_fingerprints=row[7:], + has_content_hash=_has_content_hash_column(cursor), ) ) if merged_importance is not None: @@ -903,6 +919,7 @@ def merge_existing_chunk_content( merged_tags=merged_tags, created_at=existing_created, existing_fingerprints=row[7:], + has_content_hash=_has_content_hash_column(cursor), ) ) if merged_importance is not None: diff --git a/tests/test_content_hash_repair.py b/tests/test_content_hash_repair.py new file mode 100644 index 000000000..5f22a8066 --- /dev/null +++ b/tests/test_content_hash_repair.py @@ -0,0 +1,141 @@ +"""#623 bounded repair: only stale merged rows, dry-run first, never the live DB.""" + +from __future__ import annotations + +import hashlib + +import apsw +import pytest + +from brainlayer import content_hash_repair +from brainlayer.content_hash_repair import repair_merged_content_hashes +from brainlayer.dedupe import merge_existing_chunk_content +from brainlayer.vector_store import VectorStore + + +def _sha(content: str) -> str: + return hashlib.sha256(content.strip().encode("utf-8")).hexdigest() + + +def _insert(conn, chunk_id: str, content: str, content_hash: str | None) -> None: + conn.execute( + """INSERT INTO chunks(id, content, metadata, source_file, content_hash, created_at, seen_count) + VALUES (?, ?, '{}', 'test.jsonl', ?, '2026-10-22T01:00:00Z', 1)""", + (chunk_id, content, content_hash), + ) + + +def _stale_merge(conn, chunk_id: str, original: str) -> str: + """Leave the exact pre-fix state: merged body, hash of the pre-merge body.""" + _insert(conn, chunk_id, original, _sha(original)) + merge_existing_chunk_content( + conn, chunk_id=chunk_id, incoming={"content": original + " plus a later revision"}, ensure_schema=False + ) + conn.execute("UPDATE chunks SET content_hash = ? WHERE id = ?", (_sha(original), chunk_id)) + return conn.execute("SELECT content FROM chunks WHERE id = ?", (chunk_id,)).fetchone()[0] + + +def _hashes(db) -> dict[str, str | None]: + conn = apsw.Connection(str(db), flags=apsw.SQLITE_OPEN_READONLY) + try: + return dict(conn.execute("SELECT id, content_hash FROM chunks ORDER BY id").fetchall()) + finally: + conn.close() + + +@pytest.fixture +def copy_db(tmp_path): + db = tmp_path / "copy.db" + store = VectorStore(db) + conn = store.conn + merged = { + cid: _stale_merge(conn, cid, f"Original note {cid} about the copper lantern") for cid in ("m1", "m2", "m3") + } + _insert(conn, "plain-legacy", "Ordinary row with a legacy hash scheme", "abcd1234abcd1234") + _insert(conn, "merged-null", "[Merged duplicate originals]\n\n␞ 1. a\n\n␞ 2. b", None) + fts_before = conn.execute("SELECT COUNT(*) FROM chunks_fts").fetchone()[0] + store.close() + return db, merged, fts_before + + +def test_dry_run_reports_and_writes_nothing(copy_db): + db, merged, _ = copy_db + before = _hashes(db) + result = repair_merged_content_hashes(db) + assert result["mode"] == "dry-run" + assert result["stale"] == 3 and result["repaired"] == 0 + assert sorted(result["sample_ids"]) == ["m1", "m2", "m3"] + assert _hashes(db) == before + + +def test_apply_repairs_only_stale_merged_rows_and_is_idempotent(copy_db): + db, merged, fts_before = copy_db + result = repair_merged_content_hashes(db, apply=True, batch_size=2) + assert (result["stale"], result["repaired"], result["batches"]) == (3, 3, 2) + after = _hashes(db) + for chunk_id, content in merged.items(): + assert after[chunk_id] == _sha(content) + assert after["plain-legacy"] == "abcd1234abcd1234" + assert after["merged-null"] is None + + again = repair_merged_content_hashes(db, apply=True) + assert (again["stale"], again["repaired"]) == (0, 0) + assert _hashes(db) == after + + store = VectorStore(db) + try: + assert store.conn.execute("SELECT COUNT(*) FROM chunks_fts").fetchone()[0] == fts_before + from brainlayer.drain import _apply_enrichment + + _apply_enrichment( + store.conn, {"chunk_id": "m1", "content_hash": _sha(merged["m1"]), "enrichment": {"summary": "fresh"}} + ) + assert store.conn.execute("SELECT summary FROM chunks WHERE id = 'm1'").fetchone()[0] == "fresh" + finally: + store.close() + + +def test_apply_is_bounded_per_run(copy_db): + db, _, _ = copy_db + first = repair_merged_content_hashes(db, apply=True, max_rows=2) + assert (first["repaired"], first["more_remaining"]) == (2, True) + second = repair_merged_content_hashes(db, apply=True, max_rows=2) + assert (second["repaired"], second["more_remaining"]) == (1, False) + + +def test_row_changed_after_scan_is_skipped_not_stamped(copy_db, monkeypatch): + db, _, _ = copy_db + real_find = content_hash_repair._find_candidates + + def find_then_concurrent_edit(conn, **kwargs): + found = real_find(conn, **kwargs) + other = apsw.Connection(str(db)) + other.execute("UPDATE chunks SET content = content || ' edited' WHERE id = 'm2'") + other.close() + return found + + monkeypatch.setattr(content_hash_repair, "_find_candidates", find_then_concurrent_edit) + result = repair_merged_content_hashes(db, apply=True) + assert (result["repaired"], result["skipped_identity_changed"]) == (2, 1) + + +def test_refuses_live_db_and_missing_db(copy_db, monkeypatch, tmp_path): + db, _, _ = copy_db + before = _hashes(db) + monkeypatch.setattr("brainlayer.chunk_origin_wipe._live_db_candidates", lambda: [db]) + with pytest.raises(RuntimeError, match="refusing to write the live BrainLayer DB"): + repair_merged_content_hashes(db, apply=True) + assert _hashes(db) == before + + missing = tmp_path / "absent.db" + with pytest.raises(FileNotFoundError): + repair_merged_content_hashes(missing, apply=True) + assert not missing.exists() + + +def test_cli_defaults_to_dry_run(copy_db, capsys): + db, _, _ = copy_db + before = _hashes(db) + assert content_hash_repair.main(["--db", str(db)]) == 0 + assert '"mode": "dry-run"' in capsys.readouterr().out + assert _hashes(db) == before diff --git a/tests/test_dedupe_content_hash.py b/tests/test_dedupe_content_hash.py new file mode 100644 index 000000000..06cc8c216 --- /dev/null +++ b/tests/test_dedupe_content_hash.py @@ -0,0 +1,134 @@ +"""#623: a content-changing dedupe merge must keep content_hash == sha256(content.strip()). + +Both merge writers run against the production ``VectorStore`` schema (FTS +triggers, vector tables, content_hash index) on a private tmp DB. +""" + +from __future__ import annotations + +import hashlib + +import apsw +import pytest + +from brainlayer.dedupe import merge_duplicate_chunk, merge_existing_chunk_content +from brainlayer.vector_store import VectorStore + +ORIGINAL = "Quartz retrieval original about the lantern ledger" +INCOMING = "Quartz retrieval original about the lantern ledger, now with the amber footnote" + + +def _sha(content: str) -> str: + return hashlib.sha256(content.strip().encode("utf-8")).hexdigest() + + +@pytest.fixture +def store(tmp_path): + vs = VectorStore(tmp_path / "content-hash.db") + vs.conn.execute( + """INSERT INTO chunks(id, content, metadata, source_file, content_hash, created_at, seen_count) + VALUES ('canonical', ?, '{}', 'test.jsonl', ?, '2026-10-22T01:00:00Z', 1)""", + (ORIGINAL, _sha(ORIGINAL)), + ) + try: + yield vs + finally: + vs.close() + + +def _merge(conn, merge_kind: str, content: str) -> bool: + incoming = {"content": content, "created_at": "2026-10-22T02:00:00Z"} + conn.execute("BEGIN IMMEDIATE") + try: + if merge_kind == "duplicate": + changed = merge_duplicate_chunk( + conn, + canonical_id="canonical", + duplicate_id="duplicate", + incoming=incoming, + mechanism="simhash", + hamming_distance_value=3, + ensure_schema=False, + ) + else: + changed = merge_existing_chunk_content(conn, chunk_id="canonical", incoming=incoming, ensure_schema=False) + conn.execute("COMMIT") + return changed + except Exception: + conn.execute("ROLLBACK") + raise + + +def _row(conn): + return conn.execute( + "SELECT content, content_hash, dedupe_hash, simhash FROM chunks WHERE id = 'canonical'" + ).fetchone() + + +@pytest.mark.parametrize("merge_kind", ["duplicate", "content"]) +def test_content_changing_merge_rewrites_content_hash(store, merge_kind): + assert _merge(store.conn, merge_kind, INCOMING) is True + + content, content_hash, dedupe_hash, simhash = _row(store.conn) + assert content.startswith("[Merged duplicate originals]") + assert ORIGINAL in content and INCOMING in content # retained originals + assert dedupe_hash and simhash + assert content_hash == _sha(content) + + +@pytest.mark.parametrize("merge_kind", ["duplicate", "content"]) +def test_replayed_merge_is_idempotent_and_leaves_hash_alone(store, merge_kind): + _merge(store.conn, merge_kind, INCOMING) + first = _row(store.conn) + + assert _merge(store.conn, merge_kind, INCOMING) is False + assert _merge(store.conn, merge_kind, ORIGINAL) is False + assert _row(store.conn) == first + + +@pytest.mark.parametrize("merge_kind", ["duplicate", "content"]) +def test_no_op_merge_does_not_touch_content_hash(store, merge_kind): + # A no-op merge is not a repair path: whatever content_hash holds stays. + store.conn.execute("UPDATE chunks SET content_hash = 'legacy-16charhash' WHERE id = 'canonical'") + assert _merge(store.conn, merge_kind, ORIGINAL) is False + assert _row(store.conn)[:2] == (ORIGINAL, "legacy-16charhash") + + +@pytest.mark.parametrize("merge_kind", ["duplicate", "content"]) +def test_failed_merge_rolls_back_content_and_hash_together(store, merge_kind): + before = _row(store.conn) + store.conn.execute( + """CREATE TRIGGER fail_after_chunk_update BEFORE INSERT ON dedupe_audit BEGIN + SELECT RAISE(ABORT, 'injected audit failure'); + END""" + ) + with pytest.raises(apsw.ConstraintError, match="injected audit failure"): + _merge(store.conn, merge_kind, INCOMING) + assert _row(store.conn) == before + + +def test_drain_same_id_merge_lets_fresh_enrichment_apply(store): + """The issue's symptom on the drain writer: a fresh event for the merged body was rejected.""" + from brainlayer.drain import _apply_enrichment, _insert_or_merge_chunk + + conn = store.conn + conn.execute("BEGIN IMMEDIATE") + _insert_or_merge_chunk( + conn, + { + "id": "canonical", + "content": INCOMING, + "metadata": "{}", + "source_file": "test.jsonl", + "created_at": "2026-10-22T02:00:00Z", + }, + ) + conn.execute("COMMIT") + content = _row(conn)[0] + assert content.startswith("[Merged duplicate originals]") + + event = {"chunk_id": "canonical", "content_hash": _sha(content), "enrichment": {"summary": "fresh"}} + _apply_enrichment(conn, event) + summary, stored_hash = conn.execute("SELECT summary, content_hash FROM chunks WHERE id = 'canonical'").fetchone() + assert summary == "fresh" + assert stored_hash == _sha(content) From 01ccd0c3d1a6987fbe7ebd31759c36a30795ca8d Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 5 Nov 2026 04:46:27 +0000 Subject: [PATCH 2/5] test(dedupe): adversarial DB-copy proof for content_hash merge + repair (Refs #623) Stacked on #1637. Synthetic production-schema copies only: seeded merge sequences on both writers, a 2.1k-row batched repair that must change only content_hash, a mid-batch fault that must roll back the whole batch, a forced scan/write race with the fixed writer, a real subprocess waiting on a held writer lock, and live-DB refusal via BRAINLAYER_DB and a hardlink. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01L3NoZE5vtwWCuuptxjhLbS --- tests/test_content_hash_repair_dbcopy.py | 317 +++++++++++++++++++++++ 1 file changed, 317 insertions(+) create mode 100644 tests/test_content_hash_repair_dbcopy.py diff --git a/tests/test_content_hash_repair_dbcopy.py b/tests/test_content_hash_repair_dbcopy.py new file mode 100644 index 000000000..4be31c149 --- /dev/null +++ b/tests/test_content_hash_repair_dbcopy.py @@ -0,0 +1,317 @@ +"""Adversarial DB-copy proof for #623 (stacked on #1637): both merge writers and the repair. + +Every fixture is a private synthetic copy built with the production ``VectorStore`` +schema (FTS triggers, vec0 tables, the content_hash index). Nothing here opens, +copies or names a real database. + +The invariant under attack: after a content-changing merge, and after the repair, +``content_hash == sha256(content.strip())``; every other column, FTS row and +vector row is left byte-identical. +""" + +from __future__ import annotations + +import hashlib +import json +import os +import random +import subprocess +import sys +import time + +import apsw +import pytest +import sqlite_vec + +from brainlayer import content_hash_repair +from brainlayer.content_hash_repair import repair_merged_content_hashes +from brainlayer.dedupe import ( + _MERGED_MARKER, + _merged_originals, + _whitespace_normalized, + compute_dedupe_fields, + merge_duplicate_chunk, + merge_existing_chunk_content, +) +from brainlayer.vector_store import VectorStore + +CREATED = "2026-10-22T01:00:00Z" + + +def _sha(content: str) -> str: + return hashlib.sha256(content.strip().encode("utf-8")).hexdigest() + + +def _insert(conn, chunk_id, content, content_hash, *, source_file="synthetic.jsonl", project="synthetic"): + conn.execute( + """INSERT INTO chunks(id, content, metadata, source_file, project, content_hash, created_at, seen_count) + VALUES (?, ?, ?, ?, ?, ?, ?, 1)""", + (chunk_id, content, json.dumps({"origin": chunk_id}), source_file, project, content_hash, CREATED), + ) + + +def _merge(conn, kind: str, chunk_id: str, content: str, n: int) -> bool: + incoming = {"content": content, "created_at": f"2026-10-22T02:{n % 60:02d}:00Z"} + conn.execute("BEGIN IMMEDIATE") + try: + if kind == "duplicate": + changed = merge_duplicate_chunk( + conn, + canonical_id=chunk_id, + duplicate_id=f"{chunk_id}-dup-{n}", + incoming=incoming, + mechanism="simhash", + hamming_distance_value=3, + ensure_schema=False, + ) + else: + changed = merge_existing_chunk_content(conn, chunk_id=chunk_id, incoming=incoming, ensure_schema=False) + conn.execute("COMMIT") + return changed + except BaseException: + conn.execute("ROLLBACK") + raise + + +# Inputs chosen to hit every branch of _merged_content: framing escapes, legacy +# bodies, whitespace-only/empty incoming, surrounding whitespace, re-sent originals. +TRICKY = [ + " copper lantern note with leading and trailing whitespace \n", + "record separator ␞ and escape ␛ inside an original", + "escaped-looking ␛R and ␛E sequences", + "", + " \n\t ", + "שלום unicode original — em dash, ñ, 漢字", + _MERGED_MARKER + " an original that merely starts with the marker", + "1. numbered list\n\n2. that looks like a legacy heading", +] + + +@pytest.mark.parametrize("kind", ["duplicate", "content"]) +@pytest.mark.parametrize("seed", range(6)) +def test_merge_sequences_keep_hash_equal_to_current_trimmed_content(tmp_path, kind, seed): + rng = random.Random(seed) + store = VectorStore(tmp_path / "seq.db") + conn = store.conn + try: + start = rng.choice( + [ + "plain starting body for the ledger", + _MERGED_MARKER + "\n\n1. legacy first original\n\n2. legacy second original", + ] + ) + _insert(conn, "c", start, _sha(start)) + sent = [] + for n in range(12): + incoming = rng.choice(TRICKY + sent + [f"fresh original {seed}-{n} about the amber footnote"]) + before_content, before_hash = conn.execute( + "SELECT content, content_hash FROM chunks WHERE id='c'" + ).fetchone() + changed = _merge(conn, kind, "c", incoming, n) + content, stored, dedupe_hash, simhash = conn.execute( + "SELECT content, content_hash, dedupe_hash, simhash FROM chunks WHERE id='c'" + ).fetchone() + assert changed == (content != before_content) + if changed: + assert stored == _sha(content), f"seed={seed} step={n}: stale hash after content-changing merge" + else: + assert (content, stored) == (before_content, before_hash), "no-op merge touched content/hash" + fields = compute_dedupe_fields(content, CREATED) + assert (dedupe_hash, simhash) == (fields.dedupe_hash, fields.simhash) + if incoming.strip(): + sent.append(incoming) + # Retained original text: every non-empty original is still recoverable. + normalized = {_whitespace_normalized(o) for o in _merged_originals(content)} + assert ( + _whitespace_normalized(incoming) in normalized + or _whitespace_normalized(content) == (_whitespace_normalized(incoming)) + or start.startswith(_MERGED_MARKER) + ) + finally: + store.close() + + +def _stale_merged(conn, chunk_id: str, n: int) -> None: + """The exact state the pre-fix writer left: merged body, hash of the pre-merge body.""" + original = f"original {chunk_id} about the copper lantern, revision {n}" + _insert(conn, chunk_id, original, _sha(original)) + merge_existing_chunk_content( + conn, chunk_id=chunk_id, incoming={"content": original + " plus a later amber footnote"}, ensure_schema=False + ) + conn.execute("UPDATE chunks SET content_hash = ? WHERE id = ?", (_sha(original), chunk_id)) + + +def _snapshot(db) -> dict: + conn = apsw.Connection(str(db), flags=apsw.SQLITE_OPEN_READONLY) + conn.enable_load_extension(True) + conn.load_extension(sqlite_vec.loadable_path()) + try: + cols = [r[1] for r in conn.execute("PRAGMA table_info(chunks)")] + others = [c for c in cols if c != "content_hash"] + rows = conn.execute(f"SELECT {', '.join(others)} FROM chunks ORDER BY id").fetchall() + return { + "rows": rows, + "hashes": dict(conn.execute("SELECT id, content_hash FROM chunks").fetchall()), + "fts": conn.execute("SELECT chunk_id, content FROM chunks_fts ORDER BY chunk_id").fetchall(), + "vectors": conn.execute("SELECT chunk_id, embedding FROM chunk_vectors ORDER BY chunk_id").fetchall(), + "integrity": conn.execute("PRAGMA integrity_check").fetchone()[0], + "indexes": sorted(r[1] for r in conn.execute("PRAGMA index_list(chunks)")), + } + finally: + conn.close() + + +@pytest.fixture +def big_copy(tmp_path): + """~2.4k-row synthetic copy: stale/fresh/NULL/legacy merged rows, plain stale rows, vectors.""" + db = tmp_path / "copy.db" + store = VectorStore(db) + conn = store.conn + stale = [f"stale-{i:04d}" for i in range(900)] + conn.execute("BEGIN IMMEDIATE") + for i, chunk_id in enumerate(stale): + _stale_merged(conn, chunk_id, i) + for i in range(300): + _insert(conn, f"fresh-{i:04d}", f"plain fresh row {i}", None) + merge_existing_chunk_content(conn, chunk_id=f"fresh-{i:04d}", incoming={"content": f"other {i}"}) + for i in range(100): + body = f"{_MERGED_MARKER}\n\n␞ 1. null hash {i}\n\n␞ 2. second {i}" + _insert(conn, f"mnull-{i:04d}", body, None) + for i in range(800): # Ordinary rows: legacy-scheme, edited-later (stale) and canonical hashes. + content = f"ordinary row {i} with sixteen chars" + stored = [content[:16].encode().hex()[:16], _sha(content + " before an edit"), _sha(content)][i % 3] + _insert(conn, f"plain-{i:04d}", content, stored) + for chunk_id in (*stale[:50], "plain-0001", "plain-0002"): + store._upsert_chunk_vector(conn.cursor(), chunk_id, [0.001 * (len(chunk_id) % 7)] * 1024) + conn.execute("COMMIT") + store.close() + return db, stale + + +def test_bounded_batched_repair_converges_and_touches_only_content_hash(big_copy): + db, stale = big_copy + before = _snapshot(db) + dry = repair_merged_content_hashes(db, batch_size=37, max_rows=250) + assert (dry["stale"], dry["repaired"], dry["more_remaining"]) == (250, 0, True) + assert _snapshot(db) == before + + runs = [] + while True: + result = repair_merged_content_hashes(db, apply=True, batch_size=37, max_rows=250) + runs.append(result) + if result["stale"] == 0: + break + assert len(runs) < 10, "repair did not converge" + assert [r["repaired"] for r in runs] == [250, 250, 250, 150, 0] + assert all(r["skipped_identity_changed"] == 0 for r in runs) + + after = _snapshot(db) + assert after["rows"] == before["rows"], "a column other than content_hash changed" + assert (after["fts"], after["vectors"], after["indexes"]) == (before["fts"], before["vectors"], before["indexes"]) + assert after["integrity"] == "ok" + conn = apsw.Connection(str(db), flags=apsw.SQLITE_OPEN_READONLY) + try: + for chunk_id, content, stored in conn.execute("SELECT id, content, content_hash FROM chunks"): + if chunk_id.startswith(("stale-", "fresh-")): + assert stored == _sha(content), chunk_id + else: # NULL merged and every ordinary row: exactly as before. + assert stored == before["hashes"][chunk_id], chunk_id + finally: + conn.close() + assert repair_merged_content_hashes(db)["stale"] == 0 + + +def test_mid_batch_fault_rolls_back_whole_batch_and_restart_finishes(big_copy): + db, stale = big_copy + conn = apsw.Connection(str(db)) + conn.execute( # Fault on the 3rd row of the 2nd batch (batch_size=40, rowid order). + f"""CREATE TRIGGER inject_fault BEFORE UPDATE OF content_hash ON chunks WHEN NEW.id = '{stale[42]}' + BEGIN SELECT RAISE(ABORT, 'injected mid-batch fault'); END""" + ) + conn.close() + with pytest.raises(apsw.ConstraintError, match="injected mid-batch fault"): + repair_merged_content_hashes(db, apply=True, batch_size=40, max_rows=200) + hashes = _snapshot(db)["hashes"] + reader = apsw.Connection(str(db), flags=apsw.SQLITE_OPEN_READONLY) + content = dict(reader.execute("SELECT id, content FROM chunks").fetchall()) + reader.close() + assert all(hashes[c] == _sha(content[c]) for c in stale[:40]), "committed batch 1 lost" + assert all(hashes[c] != _sha(content[c]) for c in stale[40:]), "faulted batch partially committed" + + conn = apsw.Connection(str(db)) + conn.execute("DROP TRIGGER inject_fault") + conn.close() + result = repair_merged_content_hashes(db, apply=True, batch_size=40, max_rows=5000) + assert (result["stale"], result["repaired"]) == (len(stale) - 40, len(stale) - 40) + assert repair_merged_content_hashes(db, apply=True)["repaired"] == 0 + + +def test_row_merged_by_fixed_writer_between_scan_and_write_is_skipped_and_already_correct(big_copy, monkeypatch): + db, stale = big_copy + real_find = content_hash_repair._find_candidates + + def scan_then_concurrent_merge(conn, **kwargs): + found = real_find(conn, **kwargs) + writer = VectorStore(db) # A second writer lands after the scan, before BEGIN. + try: + assert _merge(writer.conn, "content", stale[0], "a concurrent revision of the copper lantern", 0) + finally: + writer.close() + return found + + monkeypatch.setattr(content_hash_repair, "_find_candidates", scan_then_concurrent_merge) + result = repair_merged_content_hashes(db, apply=True, batch_size=100, max_rows=5000) + assert (result["repaired"], result["skipped_identity_changed"]) == (len(stale) - 1, 1) + reader = apsw.Connection(str(db), flags=apsw.SQLITE_OPEN_READONLY) + content, stored = reader.execute("SELECT content, content_hash FROM chunks WHERE id=?", (stale[0],)).fetchone() + reader.close() + assert "concurrent revision" in content and stored == _sha(content) + monkeypatch.setattr(content_hash_repair, "_find_candidates", real_find) + assert repair_merged_content_hashes(db)["stale"] == 0 + + +def test_apply_subprocess_waits_for_a_held_writer_lock_then_completes(big_copy): + db, stale = big_copy + holder = apsw.Connection(str(db)) + holder.execute("BEGIN IMMEDIATE") + env = {**os.environ, "BRAINLAYER_DB": str(db.parent / "not-the-copy.db")} + proc = subprocess.Popen( + [sys.executable, "-m", "brainlayer.content_hash_repair", "--db", str(db), "--apply", "--max-rows", "5000"], + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + env=env, + ) + try: + time.sleep(1.5) + assert proc.poll() is None, f"repair did not wait on the lock: {proc.communicate()}" + dry = subprocess.run( # A read-only dry-run is not blocked by the writer (WAL). + [sys.executable, "-m", "brainlayer.content_hash_repair", "--db", str(db)], + capture_output=True, + text=True, + env=env, + timeout=60, + ) + assert dry.returncode == 0, dry.stderr + assert json.loads(dry.stdout)["stale"] == len(stale) + finally: + holder.execute("ROLLBACK") + holder.close() + out, err = proc.communicate(timeout=60) + assert proc.returncode == 0, err + assert json.loads(out)["repaired"] == len(stale) + + +def test_copy_named_by_brainlayer_db_or_hardlinked_to_live_is_refused(big_copy, monkeypatch, tmp_path): + db, _ = big_copy + before = _snapshot(db) + monkeypatch.setenv("BRAINLAYER_DB", str(db)) + with pytest.raises(RuntimeError, match="refusing to write the live"): + repair_merged_content_hashes(db, apply=True) + monkeypatch.setenv("BRAINLAYER_DB", str(tmp_path / "elsewhere.db")) + live = tmp_path / "pretend-live.db" + os.link(db, live) + monkeypatch.setattr("brainlayer.chunk_origin_wipe.live_canonical_db_path", lambda: live) + with pytest.raises(RuntimeError, match="refusing to write the live"): + repair_merged_content_hashes(db, apply=True) + assert _snapshot(db) == before From 55f8a882b60679fa0f0bd2448a59fdb9fcbad4c5 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 5 Nov 2026 04:52:01 +0000 Subject: [PATCH 3/5] docs(ops): DB-copy acceptance runbook for the #623 content_hash repair Read-only precheck (trigger header scan, merged/NULL counts, a fingerprint of every non-content_hash column plus FTS/vector counts), dry-run, explicit bounded apply, output checks, receipt and stop conditions. States that a pass does not authorise production and that the real-copy run is still UNRUN. Refs #623 Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01L3NoZE5vtwWCuuptxjhLbS --- docs/operations/content-hash-repair-dbcopy.md | 125 ++++++++++++++++++ 1 file changed, 125 insertions(+) create mode 100644 docs/operations/content-hash-repair-dbcopy.md diff --git a/docs/operations/content-hash-repair-dbcopy.md b/docs/operations/content-hash-repair-dbcopy.md new file mode 100644 index 000000000..aee7cc0c0 --- /dev/null +++ b/docs/operations/content-hash-repair-dbcopy.md @@ -0,0 +1,125 @@ +# content_hash repair: DB-copy acceptance runbook (#623) + +**Status: LOCAL / UNRUN.** Nobody has run this against a copy of the real database yet. +The synthetic proof is `tests/test_content_hash_repair_dbcopy.py`. + +This runbook checks `python -m brainlayer.content_hash_repair` against a **copy**. That tool +ships with #1637. Passing it does **not** authorise touching the live DB: + +- The tool refuses the live DB by inode. +- It also refuses whatever `BRAINLAYER_DB` names. +- It has no `--allow-live` flag. + +## 0. Preconditions + +- **Interpreter.** Use a worktree venv that contains #1637, or the keg python once a release + contains it (`scripts/release_tag_contains.py`). Never use bare `python3`: on the M4 it + imports the repo root's live working tree through `_brainlayer.pth`. +- **Copy method.** Take a consistent snapshot of the live DB: + - preferably with BrainBar's `brain_backup_vacuum_into`; + - or with `sqlite3 "file:$LIVE?mode=ro" ".backup '$COPY'"`. + + Never `cp` only the main file: uncheckpointed WAL frames would be missing from the copy. +- **Copy location.** Keep `$COPY` outside `~/.local/share/brainlayer`, and do **not** export + `BRAINLAYER_DB=$COPY`. The tool treats that path as live and refuses it, by design. +- **Do not upload the copy anywhere.** It is personal data. + +```bash +COPY=/path/outside/brainlayer/brainlayer-623-copy.db +PY=/path/to/worktree/.venv/bin/python # or the keg python, once the release contains #1637 +``` + +## 1. Read-only precheck: record every output + +```bash +sqlite3 -readonly "$COPY" "PRAGMA quick_check;" # expect: ok +sqlite3 -readonly "$COPY" "SELECT COUNT(*) FROM pragma_table_info('chunks') WHERE name='content_hash';" # expect: 1 +# Any trigger that could fire on a content_hash UPDATE. Expect no rows. Only the header (before +# BEGIN) is read: INSERT triggers' bodies contain "ON CONFLICT ... DO UPDATE" and are not UPDATE triggers. +sqlite3 -readonly "$COPY" "SELECT name FROM (SELECT name, substr(sql, 1, instr(upper(sql), 'BEGIN')) AS h + FROM sqlite_master WHERE type='trigger' AND tbl_name='chunks') + WHERE h LIKE '%UPDATE%' AND (h NOT LIKE '%UPDATE OF%' OR h LIKE '%content_hash%');" +sqlite3 -readonly "$COPY" "SELECT COUNT(*), SUM(content_hash IS NULL) FROM chunks + WHERE substr(content,1,28)='[Merged duplicate originals]';" # merged rows, merged NULL-hash rows +``` + +Fingerprint everything the repair must **not** change. That is every `chunks` column except +`content_hash`, the FTS row count, and the vector row count: + +```bash +cat > /tmp/fp623.py <<'EOF' +import hashlib, sqlite3, sys +c = sqlite3.connect(f"file:{sys.argv[1]}?mode=ro", uri=True) +cols = [r[1] for r in c.execute("PRAGMA table_info(chunks)") if r[1] != "content_hash"] +h = hashlib.sha256() +for row in c.execute(f"SELECT {', '.join(cols)} FROM chunks ORDER BY id"): + h.update(repr(row).encode()) +fts = c.execute("SELECT COUNT(*) FROM chunks_fts").fetchone()[0] +vec = c.execute("SELECT COUNT(*) FROM chunk_vectors_rowids").fetchone()[0] +print(h.hexdigest(), "fts", fts, "vectors", vec) +EOF +"$PY" -I /tmp/fp623.py "$COPY" | tee /tmp/fp623.before +``` + +## 2. Dry-run: read-only + +```bash +"$PY" -m brainlayer.content_hash_repair --db "$COPY" | tee /tmp/623.dry1.json +``` + +Record `stale`, `merged_rows_scanned` and `more_remaining`. The repair (e) census counted +2,256 stale rows across **all** content changes. Merged-marker rows are a subset, so +`stale` should be at most 2,256. A larger number is a finding: stop and report it. + +## 3. Explicit apply: bounded runs until nothing is stale + +```bash +for i in 1 2 3 4 5 6 7 8 9 10; do + "$PY" -m brainlayer.content_hash_repair --db "$COPY" --apply --batch-size 500 --max-rows 5000 \ + | tee "/tmp/623.apply$i.json" + grep -q '"stale": 0' "/tmp/623.apply$i.json" && break +done +``` + +On a quiescent copy, every run must report `skipped_identity_changed: 0`. A non-zero count +means something else is writing to the copy: stop. + +## 4. Output checks + +```bash +"$PY" -m brainlayer.content_hash_repair --db "$COPY" | tee /tmp/623.dry2.json # expect "stale": 0 +"$PY" -I /tmp/fp623.py "$COPY" | diff /tmp/fp623.before - # expect: no diff +sqlite3 -readonly "$COPY" "PRAGMA integrity_check;" # expect: ok +``` + +Spot-check the `sample_ids` from step 2. For each one, `content_hash` must equal +`sha256(content.strip())`: + +```bash +"$PY" -I -c 'import hashlib,sqlite3,sys,json +c=sqlite3.connect(f"file:{sys.argv[1]}?mode=ro",uri=True) +for i in json.load(open(sys.argv[2]))["sample_ids"]: + t,h=c.execute("SELECT content,content_hash FROM chunks WHERE id=?",(i,)).fetchone() + print(i, h==hashlib.sha256(t.strip().encode()).hexdigest())' "$COPY" /tmp/623.dry1.json +``` + +## 5. Receipt and stop conditions + +- **Receipt.** Put the JSON outputs, the two fingerprints and the counts in a small text + receipt under `docs.local/`. Never include the copy or any row content. +- **Stop and report**, without retrying around the check, on any of these: + - a precheck trigger row; + - `stale` above 2,256; + - `skipped_identity_changed` above 0; + - a fingerprint diff; + - an integrity result other than `ok`; + - any refusal. + +## What a pass does NOT cover + +- **Production.** Applying this to the live DB is a separate lead decision. The tool cannot + do it as written, because it has no live mode. +- **Order of operations.** The fixed merge writer must be **deployed** before any production + application. Until then, the old writer keeps leaving new stale rows behind. +- **The other content changes.** Stale hashes left by non-merge content changes (the rest of + the repair (e) census) are out of scope. This tool leaves them alone. From 69c41b26dcc2ac32a5a5abbd4b78764a59218c85 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 5 Nov 2026 04:58:21 +0000 Subject: [PATCH 4/5] test+docs: tighten #623 DB-copy proof and runbook to the line budget Retained-text check is now exact: a content-changing merge must yield the prior originals verbatim plus the incoming text verbatim. The runbook's trigger precheck reads only the trigger header, so INSERT triggers whose bodies contain ON CONFLICT ... DO UPDATE are no longer false positives. Refs #623 Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01L3NoZE5vtwWCuuptxjhLbS --- docs/operations/content-hash-repair-dbcopy.md | 65 +++++++------------ tests/test_content_hash_repair_dbcopy.py | 55 +++++----------- 2 files changed, 39 insertions(+), 81 deletions(-) diff --git a/docs/operations/content-hash-repair-dbcopy.md b/docs/operations/content-hash-repair-dbcopy.md index aee7cc0c0..cb411f271 100644 --- a/docs/operations/content-hash-repair-dbcopy.md +++ b/docs/operations/content-hash-repair-dbcopy.md @@ -1,28 +1,21 @@ # content_hash repair: DB-copy acceptance runbook (#623) -**Status: LOCAL / UNRUN.** Nobody has run this against a copy of the real database yet. -The synthetic proof is `tests/test_content_hash_repair_dbcopy.py`. - -This runbook checks `python -m brainlayer.content_hash_repair` against a **copy**. That tool -ships with #1637. Passing it does **not** authorise touching the live DB: - -- The tool refuses the live DB by inode. -- It also refuses whatever `BRAINLAYER_DB` names. -- It has no `--allow-live` flag. +**Status: LOCAL / UNRUN** against any copy of the real database. Synthetic proof: +`tests/test_content_hash_repair_dbcopy.py`. This checks `python -m brainlayer.content_hash_repair` +(#1637) against a **copy**. A pass does **not** authorise touching the live DB: the tool refuses +it by inode, refuses whatever `BRAINLAYER_DB` names, and has no `--allow-live` flag. ## 0. Preconditions -- **Interpreter.** Use a worktree venv that contains #1637, or the keg python once a release - contains it (`scripts/release_tag_contains.py`). Never use bare `python3`: on the M4 it - imports the repo root's live working tree through `_brainlayer.pth`. -- **Copy method.** Take a consistent snapshot of the live DB: - - preferably with BrainBar's `brain_backup_vacuum_into`; - - or with `sqlite3 "file:$LIVE?mode=ro" ".backup '$COPY'"`. - - Never `cp` only the main file: uncheckpointed WAL frames would be missing from the copy. -- **Copy location.** Keep `$COPY` outside `~/.local/share/brainlayer`, and do **not** export - `BRAINLAYER_DB=$COPY`. The tool treats that path as live and refuses it, by design. -- **Do not upload the copy anywhere.** It is personal data. +- **Interpreter:** a worktree venv containing #1637, or the keg python once a release contains it + (`scripts/release_tag_contains.py`). Never bare `python3` (on the M4 it imports the repo root's + working tree through `_brainlayer.pth`). +- **Copy:** a consistent snapshot, preferably BrainBar's `brain_backup_vacuum_into`, else + `sqlite3 "file:$LIVE?mode=ro" ".backup '$COPY'"`. Never `cp` only the main file: uncheckpointed + WAL frames would be missing. +- **Location:** keep `$COPY` outside `~/.local/share/brainlayer` and do **not** export + `BRAINLAYER_DB=$COPY`; the tool treats that path as live and refuses it, by design. Never upload + the copy anywhere; it is personal data. ```bash COPY=/path/outside/brainlayer/brainlayer-623-copy.db @@ -34,8 +27,7 @@ PY=/path/to/worktree/.venv/bin/python # or the keg python, once the release co ```bash sqlite3 -readonly "$COPY" "PRAGMA quick_check;" # expect: ok sqlite3 -readonly "$COPY" "SELECT COUNT(*) FROM pragma_table_info('chunks') WHERE name='content_hash';" # expect: 1 -# Any trigger that could fire on a content_hash UPDATE. Expect no rows. Only the header (before -# BEGIN) is read: INSERT triggers' bodies contain "ON CONFLICT ... DO UPDATE" and are not UPDATE triggers. +# Triggers that could fire on a content_hash UPDATE (header only: INSERT bodies say DO UPDATE). Expect none. sqlite3 -readonly "$COPY" "SELECT name FROM (SELECT name, substr(sql, 1, instr(upper(sql), 'BEGIN')) AS h FROM sqlite_master WHERE type='trigger' AND tbl_name='chunks') WHERE h LIKE '%UPDATE%' AND (h NOT LIKE '%UPDATE OF%' OR h LIKE '%content_hash%');" @@ -43,8 +35,7 @@ sqlite3 -readonly "$COPY" "SELECT COUNT(*), SUM(content_hash IS NULL) FROM chunk WHERE substr(content,1,28)='[Merged duplicate originals]';" # merged rows, merged NULL-hash rows ``` -Fingerprint everything the repair must **not** change. That is every `chunks` column except -`content_hash`, the FTS row count, and the vector row count: +Fingerprint what the repair must **not** change (every other `chunks` column, FTS and vector counts): ```bash cat > /tmp/fp623.py <<'EOF' @@ -92,8 +83,7 @@ means something else is writing to the copy: stop. sqlite3 -readonly "$COPY" "PRAGMA integrity_check;" # expect: ok ``` -Spot-check the `sample_ids` from step 2. For each one, `content_hash` must equal -`sha256(content.strip())`: +Spot-check step 2's `sample_ids`: each must print `True` (hash == sha256 of trimmed content): ```bash "$PY" -I -c 'import hashlib,sqlite3,sys,json @@ -105,21 +95,14 @@ for i in json.load(open(sys.argv[2]))["sample_ids"]: ## 5. Receipt and stop conditions -- **Receipt.** Put the JSON outputs, the two fingerprints and the counts in a small text - receipt under `docs.local/`. Never include the copy or any row content. -- **Stop and report**, without retrying around the check, on any of these: - - a precheck trigger row; - - `stale` above 2,256; - - `skipped_identity_changed` above 0; - - a fingerprint diff; - - an integrity result other than `ok`; - - any refusal. +Put the JSON outputs, both fingerprints and the counts in a small text receipt under `docs.local/`, +never the copy or any row content. **Stop and report** (do not retry around the check) on: a +precheck trigger row, `stale` above 2,256, `skipped_identity_changed` above 0, a fingerprint diff, +an integrity result other than `ok`, or any refusal. ## What a pass does NOT cover -- **Production.** Applying this to the live DB is a separate lead decision. The tool cannot - do it as written, because it has no live mode. -- **Order of operations.** The fixed merge writer must be **deployed** before any production - application. Until then, the old writer keeps leaving new stale rows behind. -- **The other content changes.** Stale hashes left by non-merge content changes (the rest of - the repair (e) census) are out of scope. This tool leaves them alone. +- **Production.** Applying this to the live DB is a separate lead decision; the tool has no live mode. +- **Order.** The fixed merge writer must be **deployed** first, or the old writer keeps leaving new + stale rows behind. +- **Non-merge content changes** (the rest of the repair (e) census) are out of scope and left alone. diff --git a/tests/test_content_hash_repair_dbcopy.py b/tests/test_content_hash_repair_dbcopy.py index 4be31c149..e871b0770 100644 --- a/tests/test_content_hash_repair_dbcopy.py +++ b/tests/test_content_hash_repair_dbcopy.py @@ -1,12 +1,7 @@ -"""Adversarial DB-copy proof for #623 (stacked on #1637): both merge writers and the repair. +"""Adversarial DB-copy proof for #623 (stacked on #1637). Synthetic production-schema copies only. -Every fixture is a private synthetic copy built with the production ``VectorStore`` -schema (FTS triggers, vec0 tables, the content_hash index). Nothing here opens, -copies or names a real database. - -The invariant under attack: after a content-changing merge, and after the repair, -``content_hash == sha256(content.strip())``; every other column, FTS row and -vector row is left byte-identical. +Invariant under attack: after a content-changing merge, and after the repair, content_hash == +sha256(content.strip()); every other column, FTS row and vector row stays byte-identical. """ from __future__ import annotations @@ -28,7 +23,6 @@ from brainlayer.dedupe import ( _MERGED_MARKER, _merged_originals, - _whitespace_normalized, compute_dedupe_fields, merge_duplicate_chunk, merge_existing_chunk_content, @@ -42,11 +36,11 @@ def _sha(content: str) -> str: return hashlib.sha256(content.strip().encode("utf-8")).hexdigest() -def _insert(conn, chunk_id, content, content_hash, *, source_file="synthetic.jsonl", project="synthetic"): +def _insert(conn, chunk_id, content, content_hash): conn.execute( """INSERT INTO chunks(id, content, metadata, source_file, project, content_hash, created_at, seen_count) - VALUES (?, ?, ?, ?, ?, ?, ?, 1)""", - (chunk_id, content, json.dumps({"origin": chunk_id}), source_file, project, content_hash, CREATED), + VALUES (?, ?, ?, 'synthetic.jsonl', 'synthetic', ?, ?, 1)""", + (chunk_id, content, json.dumps({"origin": chunk_id}), content_hash, CREATED), ) @@ -94,39 +88,26 @@ def test_merge_sequences_keep_hash_equal_to_current_trimmed_content(tmp_path, ki store = VectorStore(tmp_path / "seq.db") conn = store.conn try: - start = rng.choice( - [ - "plain starting body for the ledger", - _MERGED_MARKER + "\n\n1. legacy first original\n\n2. legacy second original", - ] - ) + start = rng.choice(["plain starting body", _MERGED_MARKER + "\n\n1. legacy first\n\n2. legacy second"]) _insert(conn, "c", start, _sha(start)) sent = [] for n in range(12): incoming = rng.choice(TRICKY + sent + [f"fresh original {seed}-{n} about the amber footnote"]) - before_content, before_hash = conn.execute( - "SELECT content, content_hash FROM chunks WHERE id='c'" - ).fetchone() + sent.append(incoming) + before, before_hash = conn.execute("SELECT content, content_hash FROM chunks WHERE id='c'").fetchone() changed = _merge(conn, kind, "c", incoming, n) content, stored, dedupe_hash, simhash = conn.execute( "SELECT content, content_hash, dedupe_hash, simhash FROM chunks WHERE id='c'" ).fetchone() - assert changed == (content != before_content) + assert changed == (content != before) if changed: assert stored == _sha(content), f"seed={seed} step={n}: stale hash after content-changing merge" + # Retained original text: every prior original survives verbatim, incoming appended verbatim. + assert _merged_originals(content) == [*_merged_originals(before), incoming] else: - assert (content, stored) == (before_content, before_hash), "no-op merge touched content/hash" + assert (content, stored) == (before, before_hash), "no-op merge touched content/hash" fields = compute_dedupe_fields(content, CREATED) assert (dedupe_hash, simhash) == (fields.dedupe_hash, fields.simhash) - if incoming.strip(): - sent.append(incoming) - # Retained original text: every non-empty original is still recoverable. - normalized = {_whitespace_normalized(o) for o in _merged_originals(content)} - assert ( - _whitespace_normalized(incoming) in normalized - or _whitespace_normalized(content) == (_whitespace_normalized(incoming)) - or start.startswith(_MERGED_MARKER) - ) finally: store.close() @@ -302,16 +283,10 @@ def test_apply_subprocess_waits_for_a_held_writer_lock_then_completes(big_copy): assert json.loads(out)["repaired"] == len(stale) -def test_copy_named_by_brainlayer_db_or_hardlinked_to_live_is_refused(big_copy, monkeypatch, tmp_path): +def test_copy_named_by_brainlayer_db_is_refused_as_live(big_copy, monkeypatch): db, _ = big_copy before = _snapshot(db) - monkeypatch.setenv("BRAINLAYER_DB", str(db)) - with pytest.raises(RuntimeError, match="refusing to write the live"): - repair_merged_content_hashes(db, apply=True) - monkeypatch.setenv("BRAINLAYER_DB", str(tmp_path / "elsewhere.db")) - live = tmp_path / "pretend-live.db" - os.link(db, live) - monkeypatch.setattr("brainlayer.chunk_origin_wipe.live_canonical_db_path", lambda: live) + monkeypatch.setenv("BRAINLAYER_DB", str(db)) # The runbook forbids exporting this for the copy. with pytest.raises(RuntimeError, match="refusing to write the live"): repair_merged_content_hashes(db, apply=True) assert _snapshot(db) == before From f6038eaf443f847c4b1819a6c81bafa0999d1186 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 5 Nov 2026 02:26:21 +0000 Subject: [PATCH 5/5] fix(dedupe): rewrite content_hash with every content-changing merge (#623) merge_duplicate_chunk and merge_existing_chunk_content rewrote chunks.content into a merged-originals body and refreshed dedupe/simhash fields, but left content_hash holding the hash of the pre-merge body. The drain prefers a stored hash, so it rejected every fresh enrichment event for the merged row. The content-changing UPDATE now also sets content_hash = canonical_content_hash(merged_content), inside the existing merge savepoint. No-op merges still omit the column, so replays stay idempotent. Schemas without a content_hash column are unaffected. Adds python -m brainlayer.content_hash_repair: an explicit, dry-run-first, bounded, identity-checked repair for rows the old writer left stale. It refuses the live DB and never creates one; nothing runs it on startup. Refs #623 Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01L3NoZHSCNJgDBeHiBoPu42 --- src/brainlayer/content_hash_repair.py | 152 ++++++++++++++++++++++++++ src/brainlayer/dedupe.py | 19 +++- tests/test_content_hash_repair.py | 141 ++++++++++++++++++++++++ tests/test_dedupe_content_hash.py | 134 +++++++++++++++++++++++ 4 files changed, 445 insertions(+), 1 deletion(-) create mode 100644 src/brainlayer/content_hash_repair.py create mode 100644 tests/test_content_hash_repair.py create mode 100644 tests/test_dedupe_content_hash.py diff --git a/src/brainlayer/content_hash_repair.py b/src/brainlayer/content_hash_repair.py new file mode 100644 index 000000000..3032158af --- /dev/null +++ b/src/brainlayer/content_hash_repair.py @@ -0,0 +1,152 @@ +"""Bounded repair of merged rows whose stored content_hash went stale (#623). + +Before #623 was fixed, ``merge_duplicate_chunk`` and ``merge_existing_chunk_content`` +rewrote ``chunks.content`` into a ``[Merged duplicate originals]`` body without +rewriting ``content_hash``. The drain prefers a stored hash, so every fresh +enrichment event for such a row was rejected as stale, forever. + +This tool rewrites ONLY that column, ONLY on those rows: + +* **Explicit DB.** ``--db`` is required and must already exist. The live + canonical DB (and whatever this process resolves as live) is refused outright; + run it on a copy. Nothing calls this on startup. +* **Dry-run first.** Without ``--apply`` the DB is opened read-only and only the + candidate count and ids are reported. +* **Narrow scope.** A row qualifies only when its content starts with the merge + marker, its ``content_hash`` is non-NULL, and that hash differs from + ``sha256(content.strip())``. NULL hashes and legacy-scheme hashes on ordinary + rows are not this defect and are left alone. +* **Bounded.** At most ``--max-rows`` rows per run, written in ``--batch-size`` + transactions; the report says when more remain. +* **Identity-checked.** Each write is ``WHERE id = ? AND content = ? AND + content_hash = ?`` with the values read at scan time, so a row changed in + between is skipped and counted, never stamped with a hash of other text. +* **Idempotent.** A second run finds nothing to do. + +Content, provenance, FTS rows and vectors are untouched: ``content_hash`` is not +an FTS trigger column, and the embedded text does not change. +""" + +from __future__ import annotations + +import argparse +import json +from pathlib import Path +from typing import Any + +import apsw + +from .chunk_origin_wipe import assert_not_live_db +from .chunk_write import canonical_content_hash +from .dedupe import _MERGED_MARKER + +DEFAULT_BATCH_SIZE = 500 +DEFAULT_MAX_ROWS = 5_000 +SAMPLE_IDS = 10 + + +def _find_candidates(conn: apsw.Connection, *, scan_batch: int, max_rows: int) -> tuple[list[tuple], int, bool]: + candidates: list[tuple[str, str, str, str]] = [] + scanned = 0 + last_rowid = -1 + while len(candidates) < max_rows: + rows = conn.execute( + """ + SELECT rowid, id, content, content_hash FROM chunks + WHERE rowid > ? AND content_hash IS NOT NULL AND substr(content, 1, ?) = ? + ORDER BY rowid LIMIT ? + """, + (last_rowid, len(_MERGED_MARKER), _MERGED_MARKER, scan_batch), + ).fetchall() + if not rows: + return candidates, scanned, False + for rowid, chunk_id, content, stored_hash in rows: + last_rowid = rowid + scanned += 1 + expected = canonical_content_hash(content) + if stored_hash != expected: + candidates.append((str(chunk_id), content, stored_hash, expected)) + if len(candidates) == max_rows: + break + more = ( + conn.execute( + "SELECT 1 FROM chunks WHERE rowid > ? AND content_hash IS NOT NULL AND substr(content, 1, ?) = ? LIMIT 1", + (last_rowid, len(_MERGED_MARKER), _MERGED_MARKER), + ).fetchone() + is not None + ) + return candidates, scanned, more + + +def repair_merged_content_hashes( + db_path: str | Path, + *, + apply: bool = False, + batch_size: int = DEFAULT_BATCH_SIZE, + max_rows: int = DEFAULT_MAX_ROWS, +) -> dict[str, Any]: + if batch_size < 1 or max_rows < 1: + raise ValueError("batch_size and max_rows must be positive") + resolved = assert_not_live_db(Path(db_path)) + if not resolved.is_file(): + raise FileNotFoundError(f"no database at {resolved}; this tool never creates one") + flags = apsw.SQLITE_OPEN_READWRITE if apply else apsw.SQLITE_OPEN_READONLY + conn = apsw.Connection(str(resolved), flags=flags) + try: + conn.setbusytimeout(30_000) + if "content_hash" not in {row[1] for row in conn.execute("PRAGMA table_info(chunks)")}: + raise RuntimeError("chunks.content_hash does not exist; nothing to repair") + candidates, scanned, more = _find_candidates(conn, scan_batch=batch_size, max_rows=max_rows) + repaired = skipped = batches = 0 + if apply: + for start in range(0, len(candidates), batch_size): + conn.execute("BEGIN IMMEDIATE") + try: + for chunk_id, content, stored_hash, expected in candidates[start : start + batch_size]: + conn.execute( + "UPDATE chunks SET content_hash = ? WHERE id = ? AND content = ? AND content_hash = ?", + (expected, chunk_id, content, stored_hash), + ) + if conn.changes(): + repaired += 1 + else: + skipped += 1 + conn.execute("COMMIT") + except BaseException: + conn.execute("ROLLBACK") + raise + batches += 1 + finally: + conn.close() + return { + "mode": "apply" if apply else "dry-run", + "merged_rows_scanned": scanned, + "stale": len(candidates), + "repaired": repaired, + "skipped_identity_changed": skipped, + "batches": batches, + "more_remaining": more, + "sample_ids": [candidate[0] for candidate in candidates[:SAMPLE_IDS]], + } + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser( + prog="python -m brainlayer.content_hash_repair", + description="Repair stale content_hash on merged-duplicate rows (#623). Dry-run unless --apply; " + "refuses the live DB.", + ) + parser.add_argument("--db", type=Path, required=True, help="Path to a COPY of the database") + parser.add_argument("--apply", action="store_true", help="Write the repair (default: dry-run)") + parser.add_argument("--batch-size", type=int, default=DEFAULT_BATCH_SIZE) + parser.add_argument("--max-rows", type=int, default=DEFAULT_MAX_ROWS) + args = parser.parse_args(argv) + result = repair_merged_content_hashes( + args.db.expanduser(), apply=args.apply, batch_size=args.batch_size, max_rows=args.max_rows + ) + print(json.dumps(result, sort_keys=True)) + return 0 + + +if __name__ == "__main__": # pragma: no cover + raise SystemExit(main()) diff --git a/src/brainlayer/dedupe.py b/src/brainlayer/dedupe.py index 29331f21d..2110732f2 100644 --- a/src/brainlayer/dedupe.py +++ b/src/brainlayer/dedupe.py @@ -521,8 +521,14 @@ def _changed_merge_fields( merged_tags: list[str], created_at: str | None, existing_fingerprints: tuple[Any, ...], + has_content_hash: bool, ) -> dict[str, Any]: - """Omit unchanged indexed columns: SQLite UPDATE OF fires even for equal values.""" + """Omit unchanged indexed columns: SQLite UPDATE OF fires even for equal values. + + Every content change also rewrites ``content_hash`` in the same UPDATE (#623): + a stale stored hash made the drain reject fresh enrichment for the merged + body forever. A no-op merge leaves ``content_hash`` exactly as it was. + """ fields = compute_dedupe_fields(merged_content, created_at) fingerprints = { "dedupe_hash": fields.dedupe_hash, @@ -543,6 +549,11 @@ def _changed_merge_fields( content=merged_content, char_count=len(merged_content), ) + if has_content_hash: + # Lazy: chunk_write imports this module. One hash implementation only. + from .chunk_write import canonical_content_hash + + updates["content_hash"] = canonical_content_hash(merged_content) merged_tags_json = json.dumps(merged_tags) if merged_tags else None # Legacy CSV must be rewritten: only valid JSON populates chunk_tags. # Exact representation equality also preserves canonical ordering/formatting. @@ -637,6 +648,10 @@ def _rollback_failed_merge(cursor: Any, savepoint: str, original: Exception) -> return True +def _has_content_hash_column(cursor: Any) -> bool: + return "content_hash" in {row[1] for row in cursor.execute("PRAGMA table_info(chunks)")} + + def _invalidate_merged_vectors(cursor: Any, chunk_id: str) -> None: """Remove stale vectors within the merge savepoint.""" for table in ("chunk_vectors", "chunk_vectors_binary"): @@ -702,6 +717,7 @@ def merge_duplicate_chunk( merged_tags=merged_tags, created_at=existing_created, existing_fingerprints=row[7:], + has_content_hash=_has_content_hash_column(cursor), ) ) if merged_importance is not None: @@ -903,6 +919,7 @@ def merge_existing_chunk_content( merged_tags=merged_tags, created_at=existing_created, existing_fingerprints=row[7:], + has_content_hash=_has_content_hash_column(cursor), ) ) if merged_importance is not None: diff --git a/tests/test_content_hash_repair.py b/tests/test_content_hash_repair.py new file mode 100644 index 000000000..5f22a8066 --- /dev/null +++ b/tests/test_content_hash_repair.py @@ -0,0 +1,141 @@ +"""#623 bounded repair: only stale merged rows, dry-run first, never the live DB.""" + +from __future__ import annotations + +import hashlib + +import apsw +import pytest + +from brainlayer import content_hash_repair +from brainlayer.content_hash_repair import repair_merged_content_hashes +from brainlayer.dedupe import merge_existing_chunk_content +from brainlayer.vector_store import VectorStore + + +def _sha(content: str) -> str: + return hashlib.sha256(content.strip().encode("utf-8")).hexdigest() + + +def _insert(conn, chunk_id: str, content: str, content_hash: str | None) -> None: + conn.execute( + """INSERT INTO chunks(id, content, metadata, source_file, content_hash, created_at, seen_count) + VALUES (?, ?, '{}', 'test.jsonl', ?, '2026-10-22T01:00:00Z', 1)""", + (chunk_id, content, content_hash), + ) + + +def _stale_merge(conn, chunk_id: str, original: str) -> str: + """Leave the exact pre-fix state: merged body, hash of the pre-merge body.""" + _insert(conn, chunk_id, original, _sha(original)) + merge_existing_chunk_content( + conn, chunk_id=chunk_id, incoming={"content": original + " plus a later revision"}, ensure_schema=False + ) + conn.execute("UPDATE chunks SET content_hash = ? WHERE id = ?", (_sha(original), chunk_id)) + return conn.execute("SELECT content FROM chunks WHERE id = ?", (chunk_id,)).fetchone()[0] + + +def _hashes(db) -> dict[str, str | None]: + conn = apsw.Connection(str(db), flags=apsw.SQLITE_OPEN_READONLY) + try: + return dict(conn.execute("SELECT id, content_hash FROM chunks ORDER BY id").fetchall()) + finally: + conn.close() + + +@pytest.fixture +def copy_db(tmp_path): + db = tmp_path / "copy.db" + store = VectorStore(db) + conn = store.conn + merged = { + cid: _stale_merge(conn, cid, f"Original note {cid} about the copper lantern") for cid in ("m1", "m2", "m3") + } + _insert(conn, "plain-legacy", "Ordinary row with a legacy hash scheme", "abcd1234abcd1234") + _insert(conn, "merged-null", "[Merged duplicate originals]\n\n␞ 1. a\n\n␞ 2. b", None) + fts_before = conn.execute("SELECT COUNT(*) FROM chunks_fts").fetchone()[0] + store.close() + return db, merged, fts_before + + +def test_dry_run_reports_and_writes_nothing(copy_db): + db, merged, _ = copy_db + before = _hashes(db) + result = repair_merged_content_hashes(db) + assert result["mode"] == "dry-run" + assert result["stale"] == 3 and result["repaired"] == 0 + assert sorted(result["sample_ids"]) == ["m1", "m2", "m3"] + assert _hashes(db) == before + + +def test_apply_repairs_only_stale_merged_rows_and_is_idempotent(copy_db): + db, merged, fts_before = copy_db + result = repair_merged_content_hashes(db, apply=True, batch_size=2) + assert (result["stale"], result["repaired"], result["batches"]) == (3, 3, 2) + after = _hashes(db) + for chunk_id, content in merged.items(): + assert after[chunk_id] == _sha(content) + assert after["plain-legacy"] == "abcd1234abcd1234" + assert after["merged-null"] is None + + again = repair_merged_content_hashes(db, apply=True) + assert (again["stale"], again["repaired"]) == (0, 0) + assert _hashes(db) == after + + store = VectorStore(db) + try: + assert store.conn.execute("SELECT COUNT(*) FROM chunks_fts").fetchone()[0] == fts_before + from brainlayer.drain import _apply_enrichment + + _apply_enrichment( + store.conn, {"chunk_id": "m1", "content_hash": _sha(merged["m1"]), "enrichment": {"summary": "fresh"}} + ) + assert store.conn.execute("SELECT summary FROM chunks WHERE id = 'm1'").fetchone()[0] == "fresh" + finally: + store.close() + + +def test_apply_is_bounded_per_run(copy_db): + db, _, _ = copy_db + first = repair_merged_content_hashes(db, apply=True, max_rows=2) + assert (first["repaired"], first["more_remaining"]) == (2, True) + second = repair_merged_content_hashes(db, apply=True, max_rows=2) + assert (second["repaired"], second["more_remaining"]) == (1, False) + + +def test_row_changed_after_scan_is_skipped_not_stamped(copy_db, monkeypatch): + db, _, _ = copy_db + real_find = content_hash_repair._find_candidates + + def find_then_concurrent_edit(conn, **kwargs): + found = real_find(conn, **kwargs) + other = apsw.Connection(str(db)) + other.execute("UPDATE chunks SET content = content || ' edited' WHERE id = 'm2'") + other.close() + return found + + monkeypatch.setattr(content_hash_repair, "_find_candidates", find_then_concurrent_edit) + result = repair_merged_content_hashes(db, apply=True) + assert (result["repaired"], result["skipped_identity_changed"]) == (2, 1) + + +def test_refuses_live_db_and_missing_db(copy_db, monkeypatch, tmp_path): + db, _, _ = copy_db + before = _hashes(db) + monkeypatch.setattr("brainlayer.chunk_origin_wipe._live_db_candidates", lambda: [db]) + with pytest.raises(RuntimeError, match="refusing to write the live BrainLayer DB"): + repair_merged_content_hashes(db, apply=True) + assert _hashes(db) == before + + missing = tmp_path / "absent.db" + with pytest.raises(FileNotFoundError): + repair_merged_content_hashes(missing, apply=True) + assert not missing.exists() + + +def test_cli_defaults_to_dry_run(copy_db, capsys): + db, _, _ = copy_db + before = _hashes(db) + assert content_hash_repair.main(["--db", str(db)]) == 0 + assert '"mode": "dry-run"' in capsys.readouterr().out + assert _hashes(db) == before diff --git a/tests/test_dedupe_content_hash.py b/tests/test_dedupe_content_hash.py new file mode 100644 index 000000000..06cc8c216 --- /dev/null +++ b/tests/test_dedupe_content_hash.py @@ -0,0 +1,134 @@ +"""#623: a content-changing dedupe merge must keep content_hash == sha256(content.strip()). + +Both merge writers run against the production ``VectorStore`` schema (FTS +triggers, vector tables, content_hash index) on a private tmp DB. +""" + +from __future__ import annotations + +import hashlib + +import apsw +import pytest + +from brainlayer.dedupe import merge_duplicate_chunk, merge_existing_chunk_content +from brainlayer.vector_store import VectorStore + +ORIGINAL = "Quartz retrieval original about the lantern ledger" +INCOMING = "Quartz retrieval original about the lantern ledger, now with the amber footnote" + + +def _sha(content: str) -> str: + return hashlib.sha256(content.strip().encode("utf-8")).hexdigest() + + +@pytest.fixture +def store(tmp_path): + vs = VectorStore(tmp_path / "content-hash.db") + vs.conn.execute( + """INSERT INTO chunks(id, content, metadata, source_file, content_hash, created_at, seen_count) + VALUES ('canonical', ?, '{}', 'test.jsonl', ?, '2026-10-22T01:00:00Z', 1)""", + (ORIGINAL, _sha(ORIGINAL)), + ) + try: + yield vs + finally: + vs.close() + + +def _merge(conn, merge_kind: str, content: str) -> bool: + incoming = {"content": content, "created_at": "2026-10-22T02:00:00Z"} + conn.execute("BEGIN IMMEDIATE") + try: + if merge_kind == "duplicate": + changed = merge_duplicate_chunk( + conn, + canonical_id="canonical", + duplicate_id="duplicate", + incoming=incoming, + mechanism="simhash", + hamming_distance_value=3, + ensure_schema=False, + ) + else: + changed = merge_existing_chunk_content(conn, chunk_id="canonical", incoming=incoming, ensure_schema=False) + conn.execute("COMMIT") + return changed + except Exception: + conn.execute("ROLLBACK") + raise + + +def _row(conn): + return conn.execute( + "SELECT content, content_hash, dedupe_hash, simhash FROM chunks WHERE id = 'canonical'" + ).fetchone() + + +@pytest.mark.parametrize("merge_kind", ["duplicate", "content"]) +def test_content_changing_merge_rewrites_content_hash(store, merge_kind): + assert _merge(store.conn, merge_kind, INCOMING) is True + + content, content_hash, dedupe_hash, simhash = _row(store.conn) + assert content.startswith("[Merged duplicate originals]") + assert ORIGINAL in content and INCOMING in content # retained originals + assert dedupe_hash and simhash + assert content_hash == _sha(content) + + +@pytest.mark.parametrize("merge_kind", ["duplicate", "content"]) +def test_replayed_merge_is_idempotent_and_leaves_hash_alone(store, merge_kind): + _merge(store.conn, merge_kind, INCOMING) + first = _row(store.conn) + + assert _merge(store.conn, merge_kind, INCOMING) is False + assert _merge(store.conn, merge_kind, ORIGINAL) is False + assert _row(store.conn) == first + + +@pytest.mark.parametrize("merge_kind", ["duplicate", "content"]) +def test_no_op_merge_does_not_touch_content_hash(store, merge_kind): + # A no-op merge is not a repair path: whatever content_hash holds stays. + store.conn.execute("UPDATE chunks SET content_hash = 'legacy-16charhash' WHERE id = 'canonical'") + assert _merge(store.conn, merge_kind, ORIGINAL) is False + assert _row(store.conn)[:2] == (ORIGINAL, "legacy-16charhash") + + +@pytest.mark.parametrize("merge_kind", ["duplicate", "content"]) +def test_failed_merge_rolls_back_content_and_hash_together(store, merge_kind): + before = _row(store.conn) + store.conn.execute( + """CREATE TRIGGER fail_after_chunk_update BEFORE INSERT ON dedupe_audit BEGIN + SELECT RAISE(ABORT, 'injected audit failure'); + END""" + ) + with pytest.raises(apsw.ConstraintError, match="injected audit failure"): + _merge(store.conn, merge_kind, INCOMING) + assert _row(store.conn) == before + + +def test_drain_same_id_merge_lets_fresh_enrichment_apply(store): + """The issue's symptom on the drain writer: a fresh event for the merged body was rejected.""" + from brainlayer.drain import _apply_enrichment, _insert_or_merge_chunk + + conn = store.conn + conn.execute("BEGIN IMMEDIATE") + _insert_or_merge_chunk( + conn, + { + "id": "canonical", + "content": INCOMING, + "metadata": "{}", + "source_file": "test.jsonl", + "created_at": "2026-10-22T02:00:00Z", + }, + ) + conn.execute("COMMIT") + content = _row(conn)[0] + assert content.startswith("[Merged duplicate originals]") + + event = {"chunk_id": "canonical", "content_hash": _sha(content), "enrichment": {"summary": "fresh"}} + _apply_enrichment(conn, event) + summary, stored_hash = conn.execute("SELECT summary, content_hash FROM chunks WHERE id = 'canonical'").fetchone() + assert summary == "fresh" + assert stored_hash == _sha(content)