fix(memory): preserve consecutive pre-compaction snapshots (#7152)

* fix(memory): preserve consecutive pre-compaction snapshots

* fix(memory): warn on growing emergency snapshot backlogs

---------

Co-authored-by: Willem Jiang <willem.jiang@gmail.com>

README.md

@@ -2899,6 +2899,12 @@ DeerMem debounces pending updates independently per conversation, user, and agen
 and `max_wait_seconds` caps debounce deferral from the first pending update
 (default 300 seconds). Activity in another conversation cannot extend that wait.
 Extraction remains serial, so backlog and LLM latency can still delay completion.
+Emergency snapshots of messages removed by compaction remain separate in arrival
+order, so later compactions cannot replace earlier pending snapshots.
+A backlog warning identifies the conversation, user, and agent at 32 pending emergency
+snapshots and each doubling. These snapshots remain admitted beyond `queue_max_depth`
+and stay in memory until processing. Check extraction latency when warnings appear;
+process exit can still lose pending work.
 
 The opt-in [DeerMem scope-isolation benchmark](backend/scripts/benchmark/deermem_scope_isolation/README.md)
 checks semantic safety across facts and summaries, and fact routing across users

backend/packages/harness/deerflow/agents/memory/AGENTS.md

@@ -94,10 +94,21 @@ under continuous updates. One timer wakes for the earliest deadline and one
 consumer processes eligible items serially. Other conversations cannot reset
 that deadline. Consumer backlog and LLM latency can still delay completion.
 Early wakeups rearm with a 50 ms minimum; arrivals during extraction wait for
-the finishing worker to rearm. Emergency snapshots are immediately eligible;
+the finishing worker to rearm. Emergency snapshots are immediately eligible
+and remain separate in arrival order, including within the same conversation
+key: successive compactions can remove disjoint message partitions. Only normal
+cumulative updates coalesce; neither kind replaces the other.
+Emergency snapshots bypass `queue_max_depth`. Warn at 32 pending snapshots for
+one `(thread_id, user_id, agent_name)` key and each doubling, without logging
+message contents. Count only pending emergency items; drain, cancellation, and
+clear reset the threshold naturally. Retention remains unbounded and process-local.
+Regression coverage: `tests/test_memory_emergency_backlog.py`.
 `flush`, `flush_nowait`, and `flush_sync` make all pending items eligible.
 Tests in `tests/test_memory_queue_deadlines.py` use a fake monotonic clock and
 timer to exercise scheduling without timing-sensitive sleeps.
+`tests/test_memory_emergency_snapshots.py` covers overlapping compactions through
+the real memory hook, updater, and canonical storage, plus scoped cancellation
+and drain behavior.
 
 DeerMem and mem0 exclude assistant tool-call intent in every representation:
 `tool_calls`, `invalid_tool_calls`, and provider-raw `tool_calls`/`function_call`.

backend/packages/harness/deerflow/agents/memory/backends/deermem/deermem/core/queue.py

@@ -1,20 +1,23 @@
 """Memory update queue with debounce mechanism.
 
-The queue collects conversation contexts and processes them after a
-per-key debounce period; multiple contexts for the same
+The queue collects normal conversation contexts and processes them after a
+per-key debounce period; multiple normal contexts for the same
 ``(thread_id, user_id, agent_name)`` key are coalesced into one update. A
 monotonic deadline caps debounce deferral at ``max_wait_seconds`` from the
 first pending arrival. One timer wakes for the earliest deadline; the single
-consumer processes only eligible items unless explicitly flushed.
+consumer processes only eligible items unless explicitly flushed. Emergency
+pre-compaction snapshots stay separate and are immediately eligible.
 
 The queue is a process-local in-memory list plus a debounce
 :class:`~threading.Timer`. Items still pending at process exit are lost
 (best-effort :meth:`MemoryUpdateQueue.flush_sync` drain softens this for
-graceful shutdown). Memory updates are best-effort: a failed or lost update is
-re-fed on the next conversation turn (the middleware passes the full
+graceful shutdown). Memory updates are best-effort: a failed or lost normal update
+is re-fed on the next conversation turn (the middleware passes the full
 conversation each cycle, and the updater's watermark does not advance on
 failure), so an in-memory queue covers the realistic graceful-deploy case
-without a persistence layer.
+without a persistence layer. Emergency snapshots cannot rely on that retry
+because compaction removes their messages; they are still in-memory only and
+can be lost on process exit or extraction failure.
 """
 
 from __future__ import annotations
@@ -33,6 +36,8 @@ if TYPE_CHECKING:
 
 logger = logging.getLogger(__name__)
 
+_EMERGENCY_BACKLOG_WARNING_DEPTH = 32
+
 
 class QueueFull(Exception):
     """Raised when a non-signal update is rejected under backpressure.
@@ -66,9 +71,9 @@ class ConversationContext:
     signals: frozenset[str] = field(default_factory=frozenset)
     # Emergency (summarization) flushes bypass the updater's index watermark:
     # the subset they carry is a one-shot "extract before removal" snapshot whose
-    # own length would otherwise regress the conversation watermark. Such contexts
-    # also coexist with (do not replace) a pending normal update for the same key
-    # so a flush cannot drop a pending normal update's un-extracted tail. See
+    # own length would otherwise regress the conversation watermark. Each stays
+    # separate from other emergency snapshots and a pending normal update for the
+    # same key, so no compacted partition or un-extracted normal tail is replaced. See
     # ``_enqueue_locked``'s match-key + backpressure handling.
     bypass_watermark: bool = False
     # Monotonic scheduling times; timestamp remains wall-clock metadata.
@@ -174,18 +179,19 @@ class MemoryUpdateQueue:
         bypass_watermark: bool = False,
     ) -> ConversationContext:
         key = queue_key(thread_id, user_id, agent_name)
-        # Emergency (bypass) and normal updates coexist: the match key includes
-        # ``bypass_watermark`` so a summarization flush (bypass=True) never
-        # replaces a pending normal update for the same (thread, user, agent) --
-        # replacing it would drop the normal update's un-extracted tail, which
-        # the next turn may not re-feed if the user stops. Both are processed
-        # independently instead.
-        existing = next(
-            (c for c in self._items if queue_key(c.thread_id, c.user_id, c.agent_name) == key and c.bypass_watermark == bypass_watermark),
-            None,
-        )
+        # Only normal updates carry cumulative history and can replace an older
+        # normal update. Each emergency snapshot carries messages about to be
+        # removed by one compaction: a later snapshot may be disjoint and must
+        # not replace it. Keep emergency snapshots separate, in arrival order,
+        # from each other and from the pending normal update for this key.
+        existing = None
+        if not bypass_watermark:
+            existing = next(
+                (c for c in self._items if queue_key(c.thread_id, c.user_id, c.agent_name) == key and not c.bypass_watermark),
+                None,
+            )
         # Backpressure: once depth reaches the cap, reject NEW non-signal normal
-        # items. Same-key updates merge (do not grow depth); signal-bearing items
+        # items. Same-key normal updates merge (do not grow depth); signal-bearing items
         # and emergency (bypass) flushes are always admitted. Signals capture
         # important memories, and the emergency path captures messages about to
         # be removed by summarization -- neither can be re-fed next turn, so
@@ -216,6 +222,19 @@ class MemoryUpdateQueue:
         if existing is not None:
             self._items = [c for c in self._items if not (queue_key(c.thread_id, c.user_id, c.agent_name) == key and c.bypass_watermark == bypass_watermark)]
         self._items.append(context)
+        if bypass_watermark and len(self._items) >= _EMERGENCY_BACKLOG_WARNING_DEPTH:
+            emergency_depth = sum(c.bypass_watermark and queue_key(c.thread_id, c.user_id, c.agent_name) == key for c in self._items)
+            # Warn at 32, 64, 128, ... pending snapshots for this key. Deriving
+            # the depth from pending items naturally resets after drain/cancel,
+            # without retaining per-key warning state or warning on every arrival.
+            if emergency_depth >= _EMERGENCY_BACKLOG_WARNING_DEPTH and emergency_depth & (emergency_depth - 1) == 0:
+                logger.warning(
+                    "Memory extraction backlog: %d pending emergency snapshots for thread %r (user_id=%r, agent_name=%r); retaining all snapshots in memory. Check extraction latency; pending snapshots can be lost on process exit.",
+                    emergency_depth,
+                    thread_id,
+                    user_id,
+                    agent_name,
+                )
         return context
 
     def _rearm_timer_locked(self) -> None:

backend/tests/test_memory_emergency_backlog.py

@@ -0,0 +1,115 @@
+"""Emergency snapshot retention must make a stalled extractor observable."""
+
+import logging
+import threading
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from deerflow.agents.memory.backends.deermem.deermem.config import DeerMemConfig
+from deerflow.agents.memory.backends.deermem.deermem.core.queue import MemoryUpdateQueue
+
+MODULE = "deerflow.agents.memory.backends.deermem.deermem.core.queue"
+
+
+@pytest.fixture
+def queue_factory():
+    queues = []
+    with patch(MODULE + ".threading.Timer"):
+
+        def make(max_depth=0):
+            updater = MagicMock(update_memory=MagicMock(return_value=True))
+            queue = MemoryUpdateQueue(DeerMemConfig(queue_max_depth=max_depth), updater)
+            queues.append(queue)
+            return queue, updater
+
+        try:
+            yield make
+        finally:
+            for queue in queues:
+                queue.clear()
+
+
+def _warnings(caplog):
+    return [record.getMessage() for record in caplog.records if record.name == MODULE and record.levelno == logging.WARNING]
+
+
+@pytest.mark.parametrize("max_depth", [0, 1, 1000])
+def test_stalled_extraction_warns_at_snapshot_threshold_and_doublings(queue_factory, caplog, max_depth):
+    queue, updater = queue_factory(max_depth)
+    entered, release = threading.Event(), threading.Event()
+
+    def update(**kwargs):
+        if kwargs["thread_id"] == "busy":
+            entered.set()
+            assert release.wait(10), "extraction barrier was not released"
+        return True
+
+    updater.update_memory.side_effect = update
+    queue.add_nowait("busy", ["blocked extractor"])
+    worker = threading.Thread(target=lambda: queue._process_queue(skip_inter_item_delay=True))
+    worker.start()
+    try:
+        assert entered.wait(10)
+        with caplog.at_level(logging.WARNING, logger=MODULE):
+            for index in range(1, 66):
+                queue.add_nowait("thread", [f"private snapshot payload {index}"], user_id="alice", agent_name="researcher", trace_id=f"trace-{index}")
+                if index == 31:
+                    assert _warnings(caplog) == []
+            warnings = _warnings(caplog)
+            assert len(warnings) == 2
+            assert "32 pending emergency snapshots" in warnings[0]
+            assert "64 pending emergency snapshots" in warnings[1]
+            assert all("thread" in warning and "alice" in warning and "researcher" in warning for warning in warnings)
+            assert "private snapshot payload" not in caplog.text
+            assert queue.pending_count == 65
+            assert queue.is_processing
+    finally:
+        release.set()
+        worker.join(10)
+        assert not worker.is_alive()
+
+    queue.flush(skip_inter_item_delay=True)
+    calls = [call.kwargs for call in updater.update_memory.call_args_list][1:]
+    assert [call["messages"] for call in calls] == [[f"private snapshot payload {index}"] for index in range(1, 66)]
+    assert [call["trace_id"] for call in calls] == [f"trace-{index}" for index in range(1, 66)]
+    assert all(call["bypass_watermark"] and call["user_id"] == "alice" and call["agent_name"] == "researcher" for call in calls)
+    assert queue.pending_count == 0
+
+
+def test_backlog_warnings_count_only_same_scope_emergency_snapshots(queue_factory, caplog):
+    queue, _ = queue_factory()
+    scopes = [("thread", "alice", "researcher"), ("other-thread", "alice", "researcher"), ("thread", "bob", "researcher"), ("thread", "alice", "writer"), ("thread", None, "researcher")]
+    with caplog.at_level(logging.WARNING, logger=MODULE):
+        for thread, user, agent in scopes:
+            for _ in range(31):
+                queue.add_nowait(thread, ["private snapshot"], user_id=user, agent_name=agent)
+            queue.add(thread, ["normal cumulative update"], user_id=user, agent_name=agent)
+        assert _warnings(caplog) == []
+        queue.add_nowait("thread", ["32nd snapshot"], user_id="alice", agent_name="researcher")
+        for _ in range(10):
+            queue.add("thread", ["latest normal update"], user_id="alice", agent_name="researcher")
+        warnings = _warnings(caplog)
+        assert len(warnings) == 1
+        assert "32 pending emergency snapshots" in warnings[0]
+        assert "alice" in warnings[0] and "researcher" in warnings[0]
+        assert queue.pending_count == 161
+
+
+@pytest.mark.parametrize("remove", ["drain", "cancel", "clear"])
+def test_new_backlog_warns_again_after_pending_snapshots_are_removed(queue_factory, caplog, remove):
+    queue, _ = queue_factory()
+    with caplog.at_level(logging.WARNING, logger=MODULE):
+        for _ in range(32):
+            queue.add_nowait("thread", ["snapshot"], user_id="alice", agent_name="researcher")
+        assert len(_warnings(caplog)) == 1
+        if remove == "drain":
+            queue.flush(skip_inter_item_delay=True)
+        elif remove == "cancel":
+            assert queue.cancel_by_agent("researcher", user_id="alice") == 32
+        else:
+            queue.clear()
+        assert queue.pending_count == 0
+        for _ in range(32):
+            queue.add_nowait("thread", ["later snapshot"], user_id="alice", agent_name="researcher")
+        assert len(_warnings(caplog)) == 2

backend/tests/test_memory_emergency_snapshots.py

@@ -0,0 +1,147 @@
+"""Pre-compaction snapshots must survive overlapping queued extraction."""
+
+import json
+import threading
+from pathlib import Path
+from types import SimpleNamespace
+from unittest.mock import MagicMock, patch
+
+import pytest
+from langchain_core.messages import AIMessage, HumanMessage
+
+from deerflow.agents.memory.backends.deermem.deer_mem import DeerMem
+from deerflow.agents.memory.backends.deermem.deermem.config import DeerMemConfig
+from deerflow.agents.memory.backends.deermem.deermem.core.queue import MemoryUpdateQueue
+from deerflow.agents.memory.summarization_hook import memory_flush_hook
+from deerflow.agents.middlewares.summarization_middleware import DeerFlowSummarizationMiddleware
+from deerflow.config.memory_config import MemoryConfig
+
+QUEUE_MODULE = "deerflow.agents.memory.backends.deermem.deermem.core.queue"
+HOOK_MODULE = "deerflow.agents.memory.summarization_hook"
+PREFERENCES = (
+    "My favorite beverage is oolong tea.",
+    "My preferred programming language is Rust.",
+    "My preferred editor is Neovim.",
+    "My favorite season is autumn.",
+)
+
+
+def _pair(content: str, suffix: str) -> list:
+    return [HumanMessage(content=content, id=f"human-{suffix}"), AIMessage(content="I have recorded your preference.", id=f"ai-{suffix}")]
+
+
+class _BlockingExtractor:
+    """Synthetic provider: extract every supplied preference, pausing one call."""
+
+    def __init__(self) -> None:
+        self.entered = threading.Event()
+        self.release = threading.Event()
+        self.conversations: list[str] = []
+
+    def invoke(self, messages: list, **_kwargs) -> AIMessage:
+        conversation = str(messages[-1].content).split("<conversation>", 1)[1].split("</conversation>", 1)[0]
+        self.conversations.append(conversation)
+        if "BLOCK_OTHER_THREAD" in conversation:
+            self.entered.set()
+            assert self.release.wait(10), "extraction barrier was not released"
+        facts = [{"content": content, "category": "preference", "confidence": 0.95, "scope": "user", "durability": "durable", "authority": "descriptive"} for content in PREFERENCES if content in conversation]
+        return AIMessage(content=json.dumps({"user": {}, "history": {}, "newFacts": facts, "factsToRemove": []}))
+
+
+@pytest.mark.parametrize("compaction_count", [1, 3])
+def test_disjoint_compaction_snapshots_persist_while_extraction_is_busy(tmp_path: Path, compaction_count: int) -> None:
+    extractor = _BlockingExtractor()
+    manager = DeerMem(backend_config={"storage_path": str(tmp_path), "retrieval_adapter": "", "host_llm": extractor})
+    summary_model = MagicMock()
+    summary_model.with_config.return_value = summary_model
+    summary_model.invoke.return_value = SimpleNamespace(text="A short summary omitting specific preferences.")
+    snapshots = []
+    middleware = DeerFlowSummarizationMiddleware(
+        model=summary_model,
+        trigger=("messages", 4),
+        keep=("messages", 2),
+        token_counter=len,
+        before_summarization=[snapshots.append, memory_flush_hook],
+    )
+    runtime = SimpleNamespace(context={"thread_id": "target", "user_id": "alice", "agent_name": "researcher"})
+    queue = manager._queue
+    worker = None
+    try:
+        with patch(QUEUE_MODULE + ".threading.Timer"), patch(HOOK_MODULE + ".get_memory_config", return_value=MemoryConfig(enabled=True)), patch(HOOK_MODULE + ".get_memory_manager", return_value=manager):
+            try:
+                # A real consumer holds another thread's work at the provider boundary.
+                manager.add_nowait("unrelated", _pair("BLOCK_OTHER_THREAD", "busy"), user_id="alice")
+                worker = threading.Thread(target=lambda: queue._process_queue(skip_inter_item_delay=True))
+                worker.start()
+                assert extractor.entered.wait(10)
+
+                state = {"messages": _pair(PREFERENCES[0], "0")}
+                for index in range(compaction_count):
+                    state["messages"].extend(_pair(PREFERENCES[index + 1], str(index + 1)))
+                    compacted = middleware.compact_state(state, runtime, force=True)
+                    assert compacted is not None
+                    assert [message.id for message in compacted.messages_to_summarize] == [f"human-{index}", f"ai-{index}"]
+                    state = {"messages": list(compacted.preserved_messages), "summary_text": compacted.summary_text}
+                    # After-agent capture sees only the tail retained by compaction.
+                    manager.add("target", state["messages"], agent_name="researcher", user_id="alice")
+            finally:
+                extractor.release.set()
+                if worker is not None:
+                    worker.join(10)
+                    assert not worker.is_alive()
+            assert queue.flush_sync(10)
+
+        assert len(snapshots) == compaction_count
+        expected = set(PREFERENCES[: compaction_count + 1])
+        assert {fact["content"] for fact in manager.get_memory(agent_name="researcher", user_id="alice")["facts"]} == expected
+        assert all(any(content in conversation for conversation in extractor.conversations) for content in expected)
+    finally:
+        queue.clear()
+        manager._storage.close()
+
+
+@pytest.mark.parametrize("flush_method", ["flush", "flush_nowait", "flush_sync"])
+def test_emergency_snapshots_keep_fifo_metadata_and_normal_updates_still_coalesce(flush_method: str) -> None:
+    updater = MagicMock(update_memory=MagicMock(return_value=True))
+    queue = MemoryUpdateQueue(DeerMemConfig(queue_max_depth=1), updater)
+    with patch(QUEUE_MODULE + ".threading.Timer"), patch(QUEUE_MODULE + ".time.monotonic", return_value=100), patch(QUEUE_MODULE + ".time.sleep"):
+        queue.add("thread", ["old full transcript"], user_id="alice", agent_name="researcher")
+        # Even at capacity, neither one-shot snapshot may be shed or replace another.
+        queue.add_nowait("thread", ["removed first"], user_id="alice", agent_name="researcher", trace_id="trace-first", signals=frozenset({"preference"}))
+        queue.add_nowait("thread", ["removed second"], user_id="alice", agent_name="researcher", trace_id="trace-second", signals=frozenset({"correction"}))
+        queue.add("thread", ["latest full transcript"], user_id="alice", agent_name="researcher", trace_id="trace-normal")
+        if flush_method == "flush_sync":
+            assert queue.flush_sync(2)
+        else:
+            getattr(queue, flush_method)()
+            if flush_method == "flush_nowait":
+                queue._process_queue()
+
+    calls = [call.kwargs for call in updater.update_memory.call_args_list]
+    assert [call["messages"] for call in calls] == [["removed first"], ["removed second"], ["latest full transcript"]]
+    assert [call["trace_id"] for call in calls] == ["trace-first", "trace-second", "trace-normal"]
+    assert [call["signals"] for call in calls] == [frozenset({"preference"}), frozenset({"correction"}), frozenset()]
+    assert [call["bypass_watermark"] for call in calls] == [True, True, False]
+    assert all(call["user_id"] == "alice" and call["agent_name"] == "researcher" for call in calls)
+    assert queue.pending_count == 0
+
+
+def test_scoped_cancellation_removes_all_snapshots_without_touching_other_scopes() -> None:
+    updater = MagicMock(update_memory=MagicMock(return_value=True))
+    queue = MemoryUpdateQueue(DeerMemConfig(), updater)
+    with patch(QUEUE_MODULE + ".threading.Timer"), patch(QUEUE_MODULE + ".time.sleep"):
+        queue.add_nowait("thread", ["first"], user_id="alice", agent_name="researcher")
+        queue.add_nowait("thread", ["second"], user_id="alice", agent_name="researcher")
+        queue.add("thread", ["normal"], user_id="alice", agent_name="researcher")
+        queue.add_nowait("thread", ["other user"], user_id="bob", agent_name="researcher")
+        queue.add_nowait("thread", ["other agent"], user_id="alice", agent_name="writer")
+        queue.add_nowait("thread", ["legacy scope"], agent_name="researcher")
+
+        assert queue.cancel_by_agent("researcher", user_id="alice") == 3
+        assert queue.flush_sync(2)
+
+    assert [(call.kwargs["user_id"], call.kwargs["agent_name"], call.kwargs["messages"]) for call in updater.update_memory.call_args_list] == [
+        ("bob", "researcher", ["other user"]),
+        ("alice", "writer", ["other agent"]),
+        (None, "researcher", ["legacy scope"]),
+    ]

frontend/src/content/en/harness/memory.mdx

@@ -93,7 +93,7 @@ deermem is the default backend and implements the full contract:
 
 - The factory `get_memory_manager()` reads `manager_class: deermem` and passes `backend_config` plus a set of **host hooks** (Langfuse tracing callbacks, the default extraction LLM, hidden-message filtering, extraction metrics) to `DeerMem.from_config()`, which assembles the instance.
 - Storage is bucketed per `(user, agent)`: `memory.json` holds only shared summaries and revision metadata, every fact is its own Markdown file (with YAML front matter), and a SQLite FTS5 full-text index backs `memory_search`.
-- Writes are queued by `MemoryMiddleware` calling `add()` after each turn (debounced and batched); summarization triggers an `add_nowait()` emergency flush so compacted content is never lost.
+- Writes are queued by `MemoryMiddleware` calling `add()` after each turn (debounced and batched); summarization triggers an `add_nowait()` emergency flush to extract memories from messages about to be compacted. Each pending emergency snapshot is retained separately, even when another compaction happens before extraction finishes. A warning identifies the conversation, user, and agent at 32 pending emergency snapshots and each doubling. Emergency snapshots still bypass the queue capacity limit; these warnings make extraction backlog visible without dropping compacted messages.
 - On top of that sit fact-governance capabilities, all set under `memory.backend_config`: staleness review (`staleness_review_enabled`, on by default), plus opt-in write-side near-duplicate merging (`fact_dedup_enabled`), consolidation (`consolidation_enabled`), a capacity eviction policy (`fact_eviction_policy: hybrid-v1`; the default is `confidence`), and relevance-aware retrieval (`retrieval_relevance_enabled`).
 
 ### How openviking is integrated