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.