From 2b46fd58e7c10f3fe23706cb335410e95b9bca9a Mon Sep 17 00:00:00 2001
From: Oranje AI <293843428+oranjeai@users.noreply.github.com>
Date: Tue, 4 Aug 2026 04:32:42 +1200
Subject: [PATCH 1/2] memory: merge the content-hash reinforce's extra write
 into the live row

memu-py 1.4.0's SQLiteMemoryItemRepo.create_item_reinforce reinforces an
existing item by reading `extra`, mutating a dict copy and flushing the whole
JSON column back. A plain SELECT takes no SQLite write reservation, so every
key another connection commits between the read and the flush is silently
replaced. The event-date sweep's `mentioned_at` is such a key, and 145,995 of
149,964 rows in a live store carry it while 17,935 have reinforcement_count > 1,
so the two writers genuinely co-occur.

Compute the merge server-side with json_set and read it back inside the same
still-open write transaction. json_extract on the same column in the same
statement expresses count = count + 1 without a snapshot, and the UPDATE
promotes the session to a write transaction so the read-back cannot observe an
interleaved write.

Two details are load-bearing rather than defensive:

- content_hash has no uniqueness index (the only indexes on the table are the
  id PK autoindex and ix_memu_memory_items_id), so a bare hash-filtered UPDATE
  would bump every duplicate where upstream's .first() reinforces exactly one.
  The arm resolves one target id with limit(1) and updates by id. Measured on a
  3-row duplicate set: 1 row bumped, and it is the row .first() picks.
- NULLIF is reachable, not paranoia: a writer can blank the target's `extra`
  between the id resolve and the UPDATE, and coalesce alone would then feed ''
  to json_set, which raises "malformed JSON".

The wrapper is installed before the existing semantic-dedup patch binds the
original, so the order stays semantic -> this arm -> upstream create. A
hash-first arm would change dedup precedence: with a fixture holding the query's
text under an orthogonal embedding and unrelated text under the query's
embedding, semantic-first reinforces the semantic match while hash-first
reinforces the other row.

`update_item` carries the same defective shape and is deliberately left alone:
its only extra-writing caller is gated behind enable_item_references, which
defaults False and has no environment or nerve config source, nerve's own
memory_update path passes extra=None so the column is never written at all, and
0 of 149,964 live rows carry any key that path would write.

Measured both directions with the concurrent write injected inside the window:
unpatched loses `mentioned_at`, patched keeps it, and both reach
reinforcement_count 2. Nine new tests in tests/test_memu_bridge.py cover the
window (row, returned item and cache), the NULLIF window, an increment from a
seeded count of 7, a row deleted inside the window, dedup precedence, the
duplicate-hash set, the genuine-miss create, the absence of double counting on a
semantic hit, and the patch order.
---
 nerve/memory/memu_bridge.py |  97 ++++++++
 tests/test_memu_bridge.py   | 434 ++++++++++++++++++++++++++++++++++++
 2 files changed, 531 insertions(+)

diff --git a/nerve/memory/memu_bridge.py b/nerve/memory/memu_bridge.py
index 9642175b..a725c486 100644
--- a/nerve/memory/memu_bridge.py
+++ b/nerve/memory/memu_bridge.py
@@ -1177,6 +1177,103 @@ def _set_embedding(self, value):
 
             logger.info("Patched embeddings to use numpy float32 (saves ~170 MB on 4K items)")
 
+            # Fix 7a: upstream's content-hash reinforce arm replaces the whole
+            # `extra` column from a value read before the write, so every key
+            # another connection commits in that window is silently dropped.
+            # Merge server-side instead. Installed BEFORE Fix 7 binds the
+            # original, so the order stays semantic -> this arm -> upstream
+            # create; a hash-first arm would change dedup precedence.
+            from memu.database.models import compute_content_hash as _content_hash
+
+            _upstream_hash_reinforce = SQLiteMemoryItemRepo.create_item_reinforce
+
+            def _atomic_hash_reinforce(
+                self, *, resource_id, memory_type, summary, embedding, user_data,
+            ):
+                from sqlalchemy import func, select, update
+
+                model = self._memory_item_model
+                now = self._now()
+                filters = [
+                    func.json_extract(model.extra, "$.content_hash")
+                    == _content_hash(summary, memory_type)
+                ]
+                filters.extend(self._build_filters(model, user_data))
+                # NULLIF is required, not defensive: coalesce alone lets an
+                # empty-string extra reach json_set, which raises "malformed JSON".
+                live = func.coalesce(func.nullif(model.extra, ""), "{}")
+                extra_expr = func.json_set(
+                    func.json_set(
+                        live,
+                        "$.reinforcement_count",
+                        # json_extract on the same column in the same statement,
+                        # so the increment is atomic too.
+                        func.coalesce(
+                            func.json_extract(live, "$.reinforcement_count"), 1
+                        )
+                        + 1,
+                    ),
+                    "$.last_reinforced_at",
+                    now.isoformat(),
+                )
+                with self._sessions.session() as session:
+                    # content_hash has no uniqueness index, so resolve ONE target
+                    # id: a bare hash-filtered UPDATE would bump every duplicate
+                    # where upstream's .first() reinforces exactly one.
+                    target_id = session.execute(
+                        select(model.id).where(*filters).limit(1)
+                    ).scalars().first()
+                    if target_id is None:
+                        session.rollback()
+                        return _upstream_hash_reinforce(
+                            self,
+                            resource_id=resource_id,
+                            memory_type=memory_type,
+                            summary=summary,
+                            embedding=embedding,
+                            user_data=user_data,
+                        )
+                    # The UPDATE promotes this to a write transaction, so the
+                    # SELECT that follows cannot observe an interleaved write.
+                    result = session.execute(
+                        update(model)
+                        .where(model.id == target_id)
+                        .values(extra=extra_expr, updated_at=now)
+                    )
+                    if not result.rowcount:
+                        # The row was deleted between the two statements.
+                        session.rollback()
+                        return _upstream_hash_reinforce(
+                            self,
+                            resource_id=resource_id,
+                            memory_type=memory_type,
+                            summary=summary,
+                            embedding=embedding,
+                            user_data=user_data,
+                        )
+                    row = session.execute(
+                        select(model).where(model.id == target_id)
+                    ).scalars().first()
+                    raw = row.extra
+                    item = MemoryItem(
+                        id=row.id,
+                        resource_id=row.resource_id,
+                        memory_type=row.memory_type,
+                        summary=row.summary,
+                        embedding=self._normalize_embedding(row.embedding_json),
+                        created_at=row.created_at,
+                        updated_at=row.updated_at,
+                        # Rebuild from the MERGED value, never from a
+                        # reconstructed dict.
+                        extra=json.loads(raw) if isinstance(raw, str) else dict(raw),
+                        **self._scope_kwargs_from(row),
+                    )
+                    session.commit()
+                self.items[item.id] = item
+                return item
+
+            SQLiteMemoryItemRepo.create_item_reinforce = _atomic_hash_reinforce
+
             # Fix 7: Semantic deduplication in create_item_reinforce.
             # The default dedup is content-hash only (exact text after normalization).
             # This adds cosine similarity check against the in-memory item cache so
diff --git a/tests/test_memu_bridge.py b/tests/test_memu_bridge.py
index 92037666..76f21b80 100644
--- a/tests/test_memu_bridge.py
+++ b/tests/test_memu_bridge.py
@@ -1134,3 +1134,437 @@ async def test_transient_llm_error_still_raises_backend_unavailable(self, tmp_pa
             await bridge.memorize_file(str(target))
 
         assert bridge._service.memorize.await_count == 1
+
+
+def _build_hash_reinforce_env():
+    """Patch memU once per process and build the SQLA models once.
+
+    _patch_sqlite_bugs() MUST run before get_sqlite_sqlalchemy_models(), or the
+    MRO fix has not been applied and model construction raises TypeError.
+    get_sqlite_sqlalchemy_models() is NOT re-entrant: a second call in the same
+    interpreter raises "Column object 'url' already assigned to Table", so it is
+    module-scoped, not per test.
+    """
+    import importlib.util
+
+    import memu.app.service  # noqa: F401  initialize the package graph first
+    import memu.database.sqlite.repositories.memory_item_repo as repo_mod
+    import memu.database.sqlite.schema as schema_mod
+    from memu.database.sqlite.repositories.memory_item_repo import (
+        SQLiteMemoryItemRepo as Repo,
+    )
+
+    MemUBridge._patch_sqlite_bugs()
+    models = schema_mod.get_sqlite_sqlalchemy_models()
+    from memu.database.sqlite.sqlite import SQLiteStore
+
+    # A pristine copy of the upstream module, so a test can compare our arm
+    # against unpatched memu without reimplementing either.
+    spec = importlib.util.spec_from_file_location("_pristine_repo", repo_mod.__file__)
+    pristine = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(pristine)
+
+    return {
+        "Repo": Repo,
+        "module": repo_mod,
+        "models": models,
+        "store_cls": SQLiteStore,
+        # Held in a dict, not on a class: a plain function stored as a class
+        # attribute becomes a bound method on attribute access.
+        "shipped": Repo.__dict__["create_item_reinforce"],
+        "upstream": pristine.SQLiteMemoryItemRepo.create_item_reinforce,
+    }
+
+
+@pytest.fixture(scope="module")
+def hash_reinforce_env():
+    return _build_hash_reinforce_env()
+
+
+@pytest.fixture
+def hash_reinforce_store(tmp_path, hash_reinforce_env):
+    """A real SQLiteStore with the full _patch_sqlite_bugs() stack installed."""
+    env = hash_reinforce_env
+    Repo = env["Repo"]
+    repo_mod = env["module"]
+    models = env["models"]
+    SQLiteStore = env["store_cls"]
+    arms = {"shipped": env["shipped"], "upstream": env["upstream"]}
+
+    names = (
+        "update_item", "delete_item", "clear_items", "list_items",
+        "create_item", "create_item_reinforce", "vector_search_items",
+    )
+    saved = {n: Repo.__dict__.get(n) for n in names}
+    saved_now = Repo.__dict__.get("_now")
+
+    db_path = str(tmp_path / "memu.sqlite")
+    sqlite3.connect(db_path).execute("PRAGMA journal_mode=WAL").fetchone()
+
+    class Harness:
+        path = db_path
+        repo_cls = Repo
+        module = repo_mod
+
+        @property
+        def shipped(self):
+            return arms["shipped"]
+
+        def install(self, *, fixed):
+            fn = arms["shipped"] if fixed else arms["upstream"]
+            self.module.SQLiteMemoryItemRepo.create_item_reinforce = fn
+            self.repo_cls.create_item_reinforce = fn
+
+        def open(self):
+            return SQLiteStore(dsn=f"sqlite:///{db_path}", sqla_models=models)
+
+        def extra(self, item_id):
+            raw = sqlite3.connect(db_path).execute(
+                "SELECT extra FROM memu_memory_items WHERE id=?", (item_id,),
+            ).fetchone()[0]
+            return json.loads(raw or "{}")
+
+        def reinforce(self, repo, summary, embedding):
+            return repo.create_item_reinforce(
+                resource_id=None, memory_type="knowledge",
+                summary=summary, embedding=embedding, user_data={},
+            )
+
+    try:
+        yield Harness()
+    finally:
+        for name, fn in saved.items():
+            if fn is None:
+                if name in Repo.__dict__:
+                    delattr(Repo, name)
+            else:
+                setattr(Repo, name, fn)
+        if saved_now is None:
+            if "_now" in Repo.__dict__:
+                delattr(Repo, "_now")
+        else:
+            Repo._now = saved_now
+
+
+class TestAtomicHashReinforce:
+    """memu-py 1.4.0's content-hash reinforce arm read `extra`, mutated a dict
+    copy and flushed the whole column, so any key another connection committed
+    in that window was silently dropped. Our arm computes the merge server-side
+    with json_set and reads it back inside the same write transaction.
+
+    Every test opens a SECOND store so the item cache is cold, which is what
+    forces the semantic arm to miss and the hash arm to run.
+    """
+
+    def test_preserves_a_concurrent_extra_write(self, hash_reinforce_store):
+        """The window test: unpatched memu LOSES the concurrent key, ours keeps it."""
+        results = {}
+        for label, fixed in (("base", False), ("fixed", True)):
+            hash_reinforce_store.install(fixed=fixed)
+            store = hash_reinforce_store.open()
+            seeded = hash_reinforce_store.reinforce(
+                store.memory_item_repo, "same text", [1.0, 0.0, 0.0],
+            )
+            store.close()
+
+            second = hash_reinforce_store.open()
+            repo = second.memory_item_repo
+            assert not repo.items, "cache must be cold so the hash arm runs"
+
+            # _now() is called after the row lookup and before the commit, so a
+            # write issued from here lands inside the read -> write window.
+            real_now = hash_reinforce_store.repo_cls._now
+            fired = []
+
+            def inject(self, _real=real_now, _path=hash_reinforce_store.path,
+                       _id=seeded.id):
+                if not fired:
+                    fired.append(True)
+                    writer = sqlite3.connect(_path, timeout=30)
+                    writer.execute(
+                        "UPDATE memu_memory_items SET extra=json_set("
+                        "COALESCE(NULLIF(extra,''),'{}'),'$.mentioned_at','2026-08-03')"
+                        " WHERE id=?", (_id,),
+                    )
+                    writer.commit()
+                    writer.close()
+                return _real(self)
+
+            hash_reinforce_store.repo_cls._now = inject
+            try:
+                out = hash_reinforce_store.reinforce(repo, "same text", [1.0, 0.0, 0.0])
+            finally:
+                hash_reinforce_store.repo_cls._now = real_now
+                second.close()
+
+            assert fired, f"{label}: the concurrent write never fired"
+            assert out.id == seeded.id, f"{label}: expected a reinforce, not a create"
+            results[label] = {
+                "row": hash_reinforce_store.extra(seeded.id),
+                # The RETURNED item feeds the caller and the item cache, so a
+                # reconstructed dict here would put the lost view back one layer
+                # up even with a correct row.
+                "returned": dict(out.extra or {}),
+                "cached": dict((repo.items.get(seeded.id) or out).extra or {}),
+            }
+            sqlite3.connect(hash_reinforce_store.path).execute(
+                "DELETE FROM memu_memory_items",
+            ).connection.commit()
+
+        # Both arms must actually reinforce, so the only difference measured is
+        # whether the concurrent key survived.
+        assert results["base"]["row"]["reinforcement_count"] == 2
+        assert results["fixed"]["row"]["reinforcement_count"] == 2
+        assert "mentioned_at" not in results["base"]["row"], (
+            "the defect is gone upstream: revisit whether this patch is still needed"
+        )
+        assert results["fixed"]["row"]["mentioned_at"] == "2026-08-03"
+        # The row, the returned item and the cache must all agree.
+        assert results["fixed"]["returned"] == results["fixed"]["row"]
+        assert results["fixed"]["cached"] == results["fixed"]["row"]
+
+    def test_tolerates_an_empty_extra_written_into_the_window(self, hash_reinforce_store):
+        """NULLIF is required, not defensive.
+
+        Our arm resolves a target id, then UPDATEs it. A writer can blank that
+        row's `extra` in between, and coalesce alone would then feed '' to
+        json_set, which raises "malformed JSON".
+        """
+        from sqlalchemy.orm import Session
+
+        hash_reinforce_store.install(fixed=True)
+        store = hash_reinforce_store.open()
+        seeded = hash_reinforce_store.reinforce(
+            store.memory_item_repo, "same text", [1.0, 0.0, 0.0],
+        )
+        store.close()
+
+        second = hash_reinforce_store.open()
+        repo = second.memory_item_repo
+        assert not repo.items
+
+        real_execute = Session.execute
+        fired = []
+
+        def blank_before_update(self, statement, *args, **kwargs):
+            if not fired and str(statement).lstrip().upper().startswith("UPDATE"):
+                fired.append(True)
+                writer = sqlite3.connect(hash_reinforce_store.path, timeout=30)
+                writer.execute(
+                    "UPDATE memu_memory_items SET extra='' WHERE id=?", (seeded.id,),
+                )
+                writer.commit()
+                writer.close()
+            return real_execute(self, statement, *args, **kwargs)
+
+        Session.execute = blank_before_update
+        try:
+            hash_reinforce_store.reinforce(repo, "same text", [1.0, 0.0, 0.0])
+        finally:
+            Session.execute = real_execute
+            second.close()
+
+        assert fired, "the blanking write never fired"
+        merged = hash_reinforce_store.extra(seeded.id)
+        assert merged["reinforcement_count"] == 2
+        assert "last_reinforced_at" in merged
+
+    def test_increments_from_the_live_count_not_a_constant(self, hash_reinforce_store):
+        """The count must come from json_extract on the live column.
+
+        A fixture starting at 1 cannot tell "increment" from "set to 2", so seed
+        a higher count first.
+        """
+        hash_reinforce_store.install(fixed=True)
+        store = hash_reinforce_store.open()
+        seeded = hash_reinforce_store.reinforce(
+            store.memory_item_repo, "same text", [1.0, 0.0, 0.0],
+        )
+        store.close()
+
+        con = sqlite3.connect(hash_reinforce_store.path)
+        con.execute(
+            "UPDATE memu_memory_items SET extra=json_set("
+            "extra,'$.reinforcement_count',7) WHERE id=?", (seeded.id,),
+        )
+        con.commit()
+        assert hash_reinforce_store.extra(seeded.id)["reinforcement_count"] == 7
+
+        second = hash_reinforce_store.open()
+        assert not second.memory_item_repo.items
+        out = hash_reinforce_store.reinforce(
+            second.memory_item_repo, "same text", [1.0, 0.0, 0.0],
+        )
+        second.close()
+        assert out.id == seeded.id
+        assert hash_reinforce_store.extra(seeded.id)["reinforcement_count"] == 8
+        assert (out.extra or {})["reinforcement_count"] == 8
+
+    def test_creates_when_the_row_vanishes_inside_the_window(self, hash_reinforce_store):
+        """rowcount 0 after a successful id read is a real state.
+
+        The row can be deleted between the id resolve and the UPDATE, and that
+        must fall through to a create rather than silently no-op.
+        """
+        from sqlalchemy.orm import Session
+
+        hash_reinforce_store.install(fixed=True)
+        store = hash_reinforce_store.open()
+        seeded = hash_reinforce_store.reinforce(
+            store.memory_item_repo, "same text", [1.0, 0.0, 0.0],
+        )
+        store.close()
+
+        second = hash_reinforce_store.open()
+        repo = second.memory_item_repo
+        assert not repo.items
+
+        real_execute = Session.execute
+        fired = []
+
+        def delete_before_update(self, statement, *args, **kwargs):
+            if not fired and str(statement).lstrip().upper().startswith("UPDATE"):
+                fired.append(True)
+                writer = sqlite3.connect(hash_reinforce_store.path, timeout=30)
+                writer.execute(
+                    "DELETE FROM memu_memory_items WHERE id=?", (seeded.id,),
+                )
+                writer.commit()
+                writer.close()
+            return real_execute(self, statement, *args, **kwargs)
+
+        Session.execute = delete_before_update
+        try:
+            out = hash_reinforce_store.reinforce(repo, "same text", [1.0, 0.0, 0.0])
+        finally:
+            Session.execute = real_execute
+            second.close()
+
+        assert fired, "the deleting write never fired"
+        rows = dict(sqlite3.connect(hash_reinforce_store.path).execute(
+            "SELECT id, json_extract(extra,'$.reinforcement_count') "
+            "FROM memu_memory_items",
+        ).fetchall())
+        assert seeded.id not in rows, "the deleted row came back"
+        assert out.id in rows, "the vanished row did not fall through to a create"
+        assert rows[out.id] == 1, "a fresh create must start at 1"
+
+    def test_does_not_change_dedup_precedence(self, hash_reinforce_store):
+        """Our arm must stay BEHIND the semantic arm.
+
+        Fixture: row A carries the query's text with an orthogonal embedding,
+        row B unrelated text with the query's embedding. Semantic-first picks B;
+        a hash-first arm would pick A.
+        """
+        hash_reinforce_store.install(fixed=True)
+        store = hash_reinforce_store.open()
+        repo = store.memory_item_repo
+        row_a = hash_reinforce_store.reinforce(repo, "same text", [0.0, 0.0, 1.0])
+        row_b = hash_reinforce_store.reinforce(
+            repo, "an entirely unrelated sentence", [1.0, 0.0, 0.0],
+        )
+        store.close()
+
+        # Assert the fixture: without this the two rows dedup into one at build
+        # time and the test passes vacuously.
+        rows = sqlite3.connect(hash_reinforce_store.path).execute(
+            "SELECT count() FROM memu_memory_items",
+        ).fetchone()[0]
+        assert row_a.id != row_b.id and rows == 2, "fixture degenerate"
+
+        second = hash_reinforce_store.open()
+        second.memory_item_repo.list_items()  # warm the cache so semantic can fire
+        out = hash_reinforce_store.reinforce(
+            second.memory_item_repo, "same text", [1.0, 0.0, 0.0],
+        )
+        second.close()
+        assert out.id == row_b.id, "dedup precedence changed: hash won over semantic"
+
+    def test_bumps_exactly_one_of_several_hash_duplicates(self, hash_reinforce_store):
+        """`content_hash` has no uniqueness index, so a bare hash-filtered
+        UPDATE would bump every duplicate where upstream reinforces exactly one.
+        """
+        import uuid
+
+        hash_reinforce_store.install(fixed=True)
+        store = hash_reinforce_store.open()
+        seeded = hash_reinforce_store.reinforce(
+            store.memory_item_repo, "same text", [1.0, 0.0, 0.0],
+        )
+        store.close()
+
+        con = sqlite3.connect(hash_reinforce_store.path)
+        # Derive the column list from the live schema; a hardcoded one drifts.
+        cols = [c[1] for c in con.execute("PRAGMA table_info(memu_memory_items)")]
+        others = ", ".join(c for c in cols if c != "id")
+        for _ in range(2):
+            con.execute(
+                f"INSERT INTO memu_memory_items (id, {others}) "
+                f"SELECT ?, {others} FROM memu_memory_items WHERE id=?",
+                (str(uuid.uuid4()), seeded.id),
+            )
+        con.execute(
+            "UPDATE memu_memory_items SET extra=json_set(extra,'$.reinforcement_count',1)",
+        )
+        con.commit()
+        before = dict(con.execute(
+            "SELECT id, json_extract(extra,'$.reinforcement_count') FROM memu_memory_items",
+        ).fetchall())
+        assert len(before) == 3
+
+        second = hash_reinforce_store.open()
+        assert not second.memory_item_repo.items
+        hash_reinforce_store.reinforce(
+            second.memory_item_repo, "same text", [1.0, 0.0, 0.0],
+        )
+        second.close()
+
+        after = dict(con.execute(
+            "SELECT id, json_extract(extra,'$.reinforcement_count') FROM memu_memory_items",
+        ).fetchall())
+        bumped = [i for i in after if after[i] != before[i]]
+        assert bumped == [seeded.id], f"expected only the first match bumped, got {bumped}"
+
+    def test_a_genuine_miss_still_creates(self, hash_reinforce_store):
+        """The fall-through to upstream's create arm must stay intact."""
+        hash_reinforce_store.install(fixed=True)
+        store = hash_reinforce_store.open()
+        repo = store.memory_item_repo
+        first = hash_reinforce_store.reinforce(repo, "alpha one", [1.0, 0.0, 0.0])
+        second = hash_reinforce_store.reinforce(repo, "beta two three", [0.0, 1.0, 0.0])
+        store.close()
+
+        rows = sqlite3.connect(hash_reinforce_store.path).execute(
+            "SELECT count() FROM memu_memory_items",
+        ).fetchone()[0]
+        assert first.id != second.id
+        assert rows == 2
+
+    def test_semantic_hit_does_not_double_count(self, hash_reinforce_store):
+        """Fix 7 returns inside its own hit branch, so a semantic hit must move
+        `reinforcement_count` by exactly one and never reach our arm.
+        """
+        hash_reinforce_store.install(fixed=True)
+        store = hash_reinforce_store.open()
+        repo = store.memory_item_repo
+        seeded = hash_reinforce_store.reinforce(repo, "same text", [1.0, 0.0, 0.0])
+        repo.list_items()  # warm the cache so the semantic arm can hit
+        before = hash_reinforce_store.extra(seeded.id)["reinforcement_count"]
+        hash_reinforce_store.reinforce(repo, "same text", [1.0, 0.0, 0.0])
+        after = hash_reinforce_store.extra(seeded.id)["reinforcement_count"]
+        store.close()
+        assert after == before + 1
+
+    def test_installed_before_the_semantic_arm(self, hash_reinforce_store):
+        """Patch-order guard: the semantic wrapper must close over our arm.
+
+        If ours landed after Fix 7 instead, semantic dedup would be shadowed and
+        silently stop working.
+        """
+        outer = hash_reinforce_store.shipped
+        assert outer.__qualname__.endswith("_semantic_sqlite_reinforce")
+        inner = dict(zip(
+            outer.__code__.co_freevars,
+            (cell.cell_contents for cell in outer.__closure__),
+        ))["_original_sqlite_reinforce"]
+        assert inner.__qualname__.endswith("_atomic_hash_reinforce")

From d013575a861a5531d66f38a221b1cbca752201a8 Mon Sep 17 00:00:00 2001
From: Oranje AI <293843428+oranjeai@users.noreply.github.com>
Date: Tue, 4 Aug 2026 06:28:19 +1200
Subject: [PATCH 2/2] memory: carry content_hash through the reinforce merge,
 and straddle the window

Review round 1 on the previous commit. Two defects, both in the same two places.

1. The json_set chain set only $.reinforcement_count and $.last_reinforced_at,
   so when a writer blanks the already-chosen target's `extra` between the
   limit(1) id resolve and the UPDATE, `live` degrades to '{}' and the row is
   rewritten WITHOUT content_hash. Measured consequence: the row permanently
   stops satisfying the arm's own hash filter, so the next reinforce of the same
   text creates a duplicate (rows 1 -> 2). This is introduced by the previous
   commit, not pre-existing: upstream rewrote content_hash as a side effect of
   the whole-column flush that commit removes, so the value has to be asserted
   explicitly. Fixed by one more innermost json_set. The increment still reads
   the ORIGINAL live column, so it stays live-valued rather than reading the
   rewritten expression.

2. test_preserves_a_concurrent_extra_write injected from a patched Repo._now().
   That is correct for upstream, which calls _now() after its entity SELECT, but
   this arm calls _now() before it opens its session and before the id resolve,
   so the injected write committed BEFORE the arm's first statement. The test
   therefore only proved the arm does not clobber a key written before it
   started, which a plain Python read-modify-write also satisfies: a mutant that
   keeps the id resolve, the rowcount fall-through and the read-back but merges
   in Python left the suite green. The fixed arm now injects immediately before
   its own UPDATE. The base arm keeps its _now() hook, because upstream flushes
   via session.add/commit and never issues an explicit session.execute("UPDATE"),
   so the pre-UPDATE hook cannot fire there. That asymmetry is the finding, so it
   is stated in the test rather than papered over by asserting one hook for both.
   The in-test comment that asserted the wrong arm's ordering is replaced with
   the measured one.

The load-bearing details are now three, not the two the previous commit's
message enumerates: the content_hash carry joins limit(1) and NULLIF.

Test count is unchanged at nine: the content_hash assertion and its
reachability check (a third cold reinforce must still dedup, rows == 1) extend
the existing NULLIF window test rather than adding a near-duplicate.

The mutation matrix grows from six mutants to eight. M7 removes only the inner
content_hash json_set; M8 is the Python read-modify-write above. Both are
killed, M1-M6 stay killed, and the no-op control still survives at both ends.
---
 nerve/memory/memu_bridge.py | 11 ++++-
 tests/test_memu_bridge.py   | 83 +++++++++++++++++++++++++++++--------
 2 files changed, 75 insertions(+), 19 deletions(-)

diff --git a/nerve/memory/memu_bridge.py b/nerve/memory/memu_bridge.py
index a725c486..f39dc08a 100644
--- a/nerve/memory/memu_bridge.py
+++ b/nerve/memory/memu_bridge.py
@@ -1204,7 +1204,16 @@ def _atomic_hash_reinforce(
                 live = func.coalesce(func.nullif(model.extra, ""), "{}")
                 extra_expr = func.json_set(
                     func.json_set(
-                        live,
+                        # Re-assert the hash. Upstream rewrote it as a side
+                        # effect of the whole-column flush this arm removes, so
+                        # a writer blanking the chosen target inside the window
+                        # would leave the row unmatchable by the filter above
+                        # and the next reinforce would create a duplicate.
+                        func.json_set(
+                            live,
+                            "$.content_hash",
+                            _content_hash(summary, memory_type),
+                        ),
                         "$.reinforcement_count",
                         # json_extract on the same column in the same statement,
                         # so the increment is atomic too.
diff --git a/tests/test_memu_bridge.py b/tests/test_memu_bridge.py
index 76f21b80..eab86dff 100644
--- a/tests/test_memu_bridge.py
+++ b/tests/test_memu_bridge.py
@@ -1257,7 +1257,22 @@ class TestAtomicHashReinforce:
     """
 
     def test_preserves_a_concurrent_extra_write(self, hash_reinforce_store):
-        """The window test: unpatched memu LOSES the concurrent key, ours keeps it."""
+        """The window test: unpatched memu LOSES the concurrent key, ours keeps it.
+
+        The two arms need DIFFERENT injection points, and that asymmetry is the
+        whole reason a single hook is wrong here. Upstream calls `_now()` AFTER
+        its entity SELECT (`memory_item_repo.py:320` then `:329`), so a `_now`
+        hook lands inside its read -> write window. Our arm calls `_now()` at
+        `memu_bridge.py:1196`, above the session open and above the id resolve,
+        so the same hook would commit BEFORE the arm's first statement and the
+        test would pass even for a plain Python read-modify-write. Hook the
+        first `UPDATE` instead, which is the point our window actually opens at
+        (and the shape the two sibling window tests in this class already use).
+        Upstream flushes via `session.add`/`commit` rather than an explicit
+        `session.execute("UPDATE ...")`, so that hook would never fire there.
+        """
+        from sqlalchemy.orm import Session
+
         results = {}
         for label, fixed in (("base", False), ("fixed", True)):
             hash_reinforce_store.install(fixed=fixed)
@@ -1271,30 +1286,41 @@ def test_preserves_a_concurrent_extra_write(self, hash_reinforce_store):
             repo = second.memory_item_repo
             assert not repo.items, "cache must be cold so the hash arm runs"
 
-            # _now() is called after the row lookup and before the commit, so a
-            # write issued from here lands inside the read -> write window.
-            real_now = hash_reinforce_store.repo_cls._now
             fired = []
 
-            def inject(self, _real=real_now, _path=hash_reinforce_store.path,
-                       _id=seeded.id):
+            def write_concurrently(_path=hash_reinforce_store.path, _id=seeded.id):
+                fired.append(True)
+                writer = sqlite3.connect(_path, timeout=30)
+                writer.execute(
+                    "UPDATE memu_memory_items SET extra=json_set("
+                    "COALESCE(NULLIF(extra,''),'{}'),'$.mentioned_at','2026-08-03')"
+                    " WHERE id=?", (_id,),
+                )
+                writer.commit()
+                writer.close()
+
+            real_now = hash_reinforce_store.repo_cls._now
+            real_execute = Session.execute
+
+            def inject_at_now(self, _real=real_now):
                 if not fired:
-                    fired.append(True)
-                    writer = sqlite3.connect(_path, timeout=30)
-                    writer.execute(
-                        "UPDATE memu_memory_items SET extra=json_set("
-                        "COALESCE(NULLIF(extra,''),'{}'),'$.mentioned_at','2026-08-03')"
-                        " WHERE id=?", (_id,),
-                    )
-                    writer.commit()
-                    writer.close()
+                    write_concurrently()
                 return _real(self)
 
-            hash_reinforce_store.repo_cls._now = inject
+            def inject_before_update(self, statement, *args, **kwargs):
+                if not fired and str(statement).lstrip().upper().startswith("UPDATE"):
+                    write_concurrently()
+                return real_execute(self, statement, *args, **kwargs)
+
+            if fixed:
+                Session.execute = inject_before_update
+            else:
+                hash_reinforce_store.repo_cls._now = inject_at_now
             try:
                 out = hash_reinforce_store.reinforce(repo, "same text", [1.0, 0.0, 0.0])
             finally:
                 hash_reinforce_store.repo_cls._now = real_now
+                Session.execute = real_execute
                 second.close()
 
             assert fired, f"{label}: the concurrent write never fired"
@@ -1324,12 +1350,17 @@ def inject(self, _real=real_now, _path=hash_reinforce_store.path,
         assert results["fixed"]["cached"] == results["fixed"]["row"]
 
     def test_tolerates_an_empty_extra_written_into_the_window(self, hash_reinforce_store):
-        """NULLIF is required, not defensive.
+        """NULLIF is required, not defensive, and `content_hash` must survive.
 
         Our arm resolves a target id, then UPDATEs it. A writer can blank that
         row's `extra` in between, and coalesce alone would then feed '' to
-        json_set, which raises "malformed JSON".
+        json_set, which raises "malformed JSON". The blanked row must also come
+        back out carrying `content_hash`: upstream re-wrote it as a side effect
+        of the whole-column flush this arm removes, so it has to be asserted
+        server-side or the row permanently stops matching the arm's own hash
+        filter and the next reinforce creates a duplicate.
         """
+        from memu.database.models import compute_content_hash
         from sqlalchemy.orm import Session
 
         hash_reinforce_store.install(fixed=True)
@@ -1368,6 +1399,22 @@ def blank_before_update(self, statement, *args, **kwargs):
         merged = hash_reinforce_store.extra(seeded.id)
         assert merged["reinforcement_count"] == 2
         assert "last_reinforced_at" in merged
+        assert merged["content_hash"] == compute_content_hash("same text", "knowledge")
+
+        # Reachability, not just a field check: without content_hash the row no
+        # longer satisfies the arm's hash filter, so a third cold reinforce of
+        # the same text creates a second row instead of dedup'ing.
+        third = hash_reinforce_store.open()
+        assert not third.memory_item_repo.items
+        again = hash_reinforce_store.reinforce(
+            third.memory_item_repo, "same text", [1.0, 0.0, 0.0],
+        )
+        third.close()
+        rows = sqlite3.connect(hash_reinforce_store.path).execute(
+            "SELECT count() FROM memu_memory_items",
+        ).fetchone()[0]
+        assert again.id == seeded.id, "the blanked row stopped dedup'ing"
+        assert rows == 1, f"a duplicate was created, rows={rows}"
 
     def test_increments_from_the_live_count_not_a_constant(self, hash_reinforce_store):
         """The count must come from json_extract on the live column.