From 470ec64c96c6d8b606a71c0b72622d17f562d71d Mon Sep 17 00:00:00 2001 From: Oranje AI <293843428+oranjeai@users.noreply.github.com> Date: Mon, 3 Aug 2026 21:31:37 +1200 Subject: [PATCH] memory: make the event-date sweep's writes atomic so concurrent writers are not clobbered `MemUBridge._resolve_event_dates_sync` snapshotted each item's whole `extra` JSON blob in its pre-LLM SELECT, awaited an Anthropic call to resolve event dates, then wrote that pre-call snapshot back merely to add `mentioned_at`. The blob is the unit of write, so the update was last-writer-wins over the whole dict instead of over the one key the sweep owns: every key another writer set inside that window was silently reverted. The sweep runs in `_blocking_pool` on its own raw `sqlite3` connection, so it genuinely races the normal write paths rather than being serialised behind them. Measured, snapshot then concurrent write then writeback: every key the other writer set reverts to its stale value, and a key it added is deleted outright. The live victims are the reinforce counters `reinforcement_count` and `last_reinforced_at`, rewritten on existing rows by two SQLite paths (`memu_bridge.py:1211` and memU's own `create_item_reinforce` branch), so salience silently rolls back. The same loop carried a second, independent lost update: the `happened_at` UPDATE did not re-assert the SELECT's own `happened_at IS NULL` predicate, so a concurrent backfill (`cli.py`) was overwritten. Each write is now a function of the current row, evaluated inside the write statement: - the `extra` write uses `json_set` against the live column, so a concurrent writer's keys survive. `COALESCE(NULLIF(extra, ''), '{}')` normalises NULL/empty, because `json_set(NULL, ...)` returns NULL and `json_extract('', ...)` raises; without it a NULL-extra row would stay NULL and be re-swept on every future conversation forever. An inner `COALESCE(json_extract(...))` preserves the documented stamp-once behaviour against a second concurrent sweep, which is the only other writer of that key. - the `happened_at` UPDATE gains `AND happened_at IS NULL`, strictly narrowing what the SELECT already restricted. - `extra` is dropped from the SELECT column list, since no Python code reads it any more. The `instr(extra, ...)` scoping guard lives in the WHERE clause and is unchanged. Still exactly one UPDATE per row, so the small-batch commit property is preserved. `json1` is not a new requirement: memU's own `create_item` dedup path already uses `json_extract(extra, '$.content_hash')`, so no store that reaches this sweep can lack it. Three tests drive the concurrent write from inside the real LLM await window, and each fails on the previous code with the exact defects above and passes with this change. A fourth test (two cases) guards the empty-value normalisation and passes both ways by design, which its docstring states. All four use offset-aware timestamps, so none depends on the host timezone. Two of the three also assert the sweep's own write landed on an unraced control row, so they cannot pass by the write never running. A ten-arm mutation matrix kills all eight real mutants with the no-op control surviving at both ends; every mutant leaves the eleven pre-existing sweep tests green, so that suite had no visibility into this defect class. Full suite on both arms from independent clean exports: 4 failed / 2941 passed before, 1 failed / 2944 passed after, failure sets diffed by name with zero new failures. On a live store `json_set` is semantically identical to the old Python round-trip on all 140,778 rows. Note that a later concurrent sweep no longer overwrites an earlier `mentioned_at` stamp. This restores the documented contract, but it is an observable behaviour change. The eleven pre-existing sweep tests need `TZ=America/New_York`: `NerveConfig().timezone` defaults to that zone while those tests use naive timestamps, so on a host in another timezone six of them already fail at unmodified main. That is pre-existing and out of scope here; the tests added here pass in any zone. --- nerve/memory/memu_bridge.py | 19 +++-- tests/test_memu_bridge.py | 144 ++++++++++++++++++++++++++++++++++++ 2 files changed, 157 insertions(+), 6 deletions(-) diff --git a/nerve/memory/memu_bridge.py b/nerve/memory/memu_bridge.py index 9642175b..1c4c0e3d 100644 --- a/nerve/memory/memu_bridge.py +++ b/nerve/memory/memu_bridge.py @@ -2498,7 +2498,7 @@ def _resolve_event_dates_sync(self, conversation_ts: str) -> None: - timedelta(hours=self._DATE_SWEEP_WINDOW_HOURS) ).strftime("%Y-%m-%d %H:%M:%S") rows = db.execute( - "SELECT id, memory_type, summary, extra " + "SELECT id, memory_type, summary " "FROM memu_memory_items " "WHERE happened_at IS NULL " " AND (created_at IS NULL OR created_at >= ?) " @@ -2549,8 +2549,11 @@ def _commit_batch(force: bool = False) -> None: for item_id, summary in event_items: happened_at = resolved_dates.get(item_id) or conv_date + # Re-assert the SELECT's predicate: a concurrent backfill may + # have set happened_at while we were awaiting the LLM. db.execute( - "UPDATE memu_memory_items SET happened_at = ? WHERE id = ?", + "UPDATE memu_memory_items SET happened_at = ? " + "WHERE id = ? AND happened_at IS NULL", (happened_at, item_id), ) pending += 1 @@ -2559,11 +2562,15 @@ def _commit_batch(force: bool = False) -> None: # Set mentioned_at on ALL swept items (events + non-events) for row in rows: item_id = row["id"] - extra = json.loads(row["extra"]) if row["extra"] else {} - extra["mentioned_at"] = conv_date + # json_set against the LIVE column, so keys another writer added + # during the LLM await survive; inner COALESCE keeps the first stamp. db.execute( - "UPDATE memu_memory_items SET extra = ? WHERE id = ?", - (json.dumps(extra, ensure_ascii=False), item_id), + "UPDATE memu_memory_items SET extra = json_set(" + " COALESCE(NULLIF(extra, ''), '{}'), '$.mentioned_at'," + " COALESCE(json_extract(COALESCE(NULLIF(extra, ''), '{}')," + " '$.mentioned_at'), ?)" + ") WHERE id = ?", + (conv_date, item_id), ) pending += 1 _commit_batch() diff --git a/tests/test_memu_bridge.py b/tests/test_memu_bridge.py index 92037666..9f3dd137 100644 --- a/tests/test_memu_bridge.py +++ b/tests/test_memu_bridge.py @@ -317,6 +317,150 @@ def test_sweep_row_cap_takes_newest(self, tmp_path, monkeypatch): assert "mentioned_at" in json.loads(items["newer"]["extra"]) assert "mentioned_at" not in json.loads(items["oldest"]["extra"]) + def test_sweep_preserves_a_concurrent_extra_write(self, tmp_path): + """Keys another writer sets in ``extra`` during the LLM await survive + the sweep's own writeback.""" + config = _make_config(tmp_path) + db_path = config.memory.sqlite_dsn.replace("sqlite:///", "") + _create_memu_schema(db_path) + _insert_items(db_path, [ + {"id": "evt-1", "memory_type": "event", "summary": "Some event", + "extra": {"content_hash": "OLDHASH", "reinforcement_count": 3}}, + ]) + + def _llm_then_concurrent_write(items, conv_date): + writer = sqlite3.connect(db_path, timeout=30) + writer.execute( + "UPDATE memu_memory_items SET extra = ? WHERE id = 'evt-1'", + (json.dumps({ + "content_hash": "NEWHASH", + "reinforcement_count": 9, + "last_reinforced_at": "2026-02-27T00:00:00", + "ref_id": "r7", + }),), + ) + writer.commit() + writer.close() + return {"evt-1": "2026-02-05"} + + bridge = MemUBridge(config) + with patch.object(bridge, "_resolve_dates_via_llm", + side_effect=_llm_then_concurrent_write): + bridge._resolve_event_dates_sync("2026-02-27T10:00:00+00:00") + + item = _read_items(db_path)["evt-1"] + extra = json.loads(item["extra"]) + assert extra["content_hash"] == "NEWHASH" + assert extra["reinforcement_count"] == 9 + assert extra["last_reinforced_at"] == "2026-02-27T00:00:00" + assert extra["ref_id"] == "r7" + # ... and the sweep still does its own job. + assert extra["mentioned_at"] == "2026-02-27" + assert item["happened_at"] == "2026-02-05" + + def test_sweep_does_not_overwrite_a_concurrent_happened_at(self, tmp_path): + """The ``happened_at IS NULL`` predicate is re-asserted at write time, + so a concurrent backfill during the LLM await is not overwritten.""" + config = _make_config(tmp_path) + db_path = config.memory.sqlite_dsn.replace("sqlite:///", "") + _create_memu_schema(db_path) + _insert_items(db_path, [ + {"id": "evt-1", "memory_type": "event", "summary": "Some event"}, + {"id": "evt-2", "memory_type": "event", "summary": "Other event"}, + ]) + + def _llm_then_concurrent_backfill(items, conv_date): + writer = sqlite3.connect(db_path, timeout=30) + writer.execute( + "UPDATE memu_memory_items SET happened_at = '2020-01-01' " + "WHERE id = 'evt-1'" + ) + writer.commit() + writer.close() + return {"evt-1": "2026-02-05", "evt-2": "2026-02-06"} + + bridge = MemUBridge(config) + with patch.object(bridge, "_resolve_dates_via_llm", + side_effect=_llm_then_concurrent_backfill): + bridge._resolve_event_dates_sync("2026-02-27T10:00:00+00:00") + + items = _read_items(db_path) + assert items["evt-1"]["happened_at"] == "2020-01-01" + # Unraced control: the guarded UPDATE must still execute, so a test + # that only checks evt-1 cannot pass by the write never running. + assert items["evt-2"]["happened_at"] == "2026-02-06" + + def test_sweep_keeps_the_first_mentioned_at_stamp_under_a_concurrent_sweep( + self, tmp_path + ): + """The first ``mentioned_at`` stamp wins. The racing writer is a SECOND + sweep, since the sweep is the only writer of that key.""" + config = _make_config(tmp_path) + db_path = config.memory.sqlite_dsn.replace("sqlite:///", "") + _create_memu_schema(db_path) + _insert_items(db_path, [ + {"id": "evt-1", "memory_type": "event", "summary": "Some event"}, + {"id": "prof-2", "memory_type": "profile", "summary": "A fact"}, + ]) + + def _llm_then_concurrent_sweep_stamp(items, conv_date): + writer = sqlite3.connect(db_path, timeout=30) + writer.execute( + "UPDATE memu_memory_items SET extra = ? WHERE id = 'evt-1'", + (json.dumps({"mentioned_at": "2026-02-20"}),), + ) + writer.commit() + writer.close() + return {} + + bridge = MemUBridge(config) + with patch.object(bridge, "_resolve_dates_via_llm", + side_effect=_llm_then_concurrent_sweep_stamp): + bridge._resolve_event_dates_sync("2026-02-27T10:00:00+00:00") + + items = _read_items(db_path) + assert json.loads(items["evt-1"]["extra"])["mentioned_at"] == "2026-02-20" + # Unraced control: the extra UPDATE must still execute, so a test that + # only checks evt-1 cannot pass by the write never running. + assert json.loads(items["prof-2"]["extra"])["mentioned_at"] == "2026-02-27" + + @pytest.mark.parametrize("raw_extra", [None, ""]) + def test_sweep_stamps_an_empty_extra_row_once(self, tmp_path, raw_extra): + """An empty ``extra`` (SQL NULL or ``''``) becomes a real JSON object + carrying the stamp, so a later sweep no longer selects the row. + + Passes at base by design; it guards the fix's own empty-value + normalisation, where ``''`` -- not NULL -- discriminates ``NULLIF``.""" + config = _make_config(tmp_path) + db_path = config.memory.sqlite_dsn.replace("sqlite:///", "") + _create_memu_schema(db_path) + db = sqlite3.connect(db_path) + db.execute( + "INSERT INTO memu_memory_items " + "(id, resource_id, memory_type, summary, extra) " + "VALUES ('prof-1', 'res-1', 'profile', 'A fact', ?)", + (raw_extra,), + ) + db.commit() + db.close() + + bridge = MemUBridge(config) + bridge._resolve_event_dates_sync("2026-02-27T10:00:00+00:00") + + raw = _read_items(db_path)["prof-1"]["extra"] + assert raw is not None + assert json.loads(raw)["mentioned_at"] == "2026-02-27" + + # The row must not be picked up by a later sweep. + db = sqlite3.connect(db_path) + resweep = db.execute( + "SELECT count(*) FROM memu_memory_items " + "WHERE happened_at IS NULL " + " AND (extra IS NULL OR instr(extra, '\"mentioned_at\"') = 0)" + ).fetchone()[0] + db.close() + assert resweep == 0 + def _mock_anthropic(response_text: str) -> tuple[MagicMock, MagicMock]: """Create a mock anthropic module and client that returns the given text.