fix(memory): prevent cross-conversation debounce starvation (#6672)

* fix(memory): prevent cross-conversation debounce starvation

* fix(config): bump schema version for memory max wait
This commit is contained in:
Jie Ji 2026-10-11 17:16:25 +08:00 committed by GitHub
parent 65e8b6591b
commit bdc2d12367
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
10 changed files with 276 additions and 38 deletions

View File

@ -2447,6 +2447,12 @@ become available in the next batch, when rejected additions can be retried.
### Long-Term Memory
DeerMem debounces pending updates independently per conversation, user, and agent.
`memory.backend_config.debounce_seconds` sets the quiet period (default 30 seconds),
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.
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
and agents. Failed extraction attempts are retryable execution errors, not safety passes.

View File

@ -87,6 +87,18 @@ fact-only import payload supplies empty summary defaults.
It captures `user_id` when it enqueues work.
This identity survives the background timer boundary.
DeerMem debounces each `(thread_id, user_id, agent_name)` independently using
monotonic deadlines. Same-key coalescing preserves the first pending arrival;
`backend_config.max_wait_seconds` (default 300) caps debounce deferral even
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;
`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.
DeerMem and mem0 exclude assistant tool-call intent in every representation:
`tool_calls`, `invalid_tool_calls`, and provider-raw `tool_calls`/`function_call`.
An empty parsed call list does not make an attempted call a final response.

View File

@ -143,7 +143,14 @@ class DeerMemConfig(BaseModel):
default=30,
ge=1,
le=300,
description="Seconds to wait before processing queued updates (debounce).",
description="Per-conversation quiet period before queued extraction becomes eligible.",
)
max_wait_seconds: int = Field(
default=300,
ge=1,
description=(
"Maximum debounce deferral from a conversation's first pending update. Continuous same-key updates cannot postpone eligibility beyond this limit. Extraction remains serial: consumer backlog and LLM latency can delay completion."
),
)
queue_max_depth: int = Field(
default=1000,

View File

@ -1,8 +1,11 @@
"""Memory update queue with debounce mechanism.
The queue collects conversation contexts and processes them after a
configurable debounce period; multiple contexts for the same
``(thread_id, user_id, agent_name)`` key are coalesced into one update.
per-key debounce period; multiple 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.
The queue is a process-local in-memory list plus a debounce
:class:`~threading.Timer`. Items still pending at process exit are lost
@ -68,14 +71,17 @@ class ConversationContext:
# so a flush cannot drop a pending normal update's un-extracted tail. See
# ``_enqueue_locked``'s match-key + backpressure handling.
bypass_watermark: bool = False
# Monotonic scheduling times; timestamp remains wall-clock metadata.
first_at: float = field(default_factory=time.monotonic)
due_at: float = 0.0
class MemoryUpdateQueue:
"""Queue for memory updates with debounce mechanism.
This queue collects conversation contexts and processes them after
a configurable debounce period. Multiple conversations received within
the debounce window are batched together.
a per-conversation debounce period. Conversations whose deadlines have
elapsed are batched together; unrelated arrivals cannot postpone them.
"""
def __init__(self, config: DeerMemConfig, updater: MemoryUpdater):
@ -85,6 +91,7 @@ class MemoryUpdateQueue:
self._items: list[ConversationContext] = []
self._lock = threading.Lock()
self._timer: threading.Timer | None = None
self._timer_due_at: float | None = None
self._processing = False
# Thread currently running ``_process_queue`` (None when idle). ``flush_sync``
# joins an in-flight worker instead of reporting a false-positive "completed"
@ -127,7 +134,7 @@ class MemoryUpdateQueue:
signals=frozenset(signals) if signals else frozenset(),
bypass_watermark=False,
)
self._reset_timer()
self._rearm_timer_locked()
logger.info("Memory update queued for thread %s, queue size: %d", thread_id, len(self._items))
@ -189,6 +196,12 @@ class MemoryUpdateQueue:
# Merge by signal union: a signal seen on any update for this key stays.
merged_signals = signals | (existing.signals if existing is not None else frozenset())
now = time.monotonic()
first_at = existing.first_at if existing is not None else now
due_at = now if bypass_watermark else min(now + self._config.debounce_seconds, first_at + self._config.max_wait_seconds)
# A forced flush's deadline must survive same-key coalescing.
if existing is not None and existing.due_at <= now:
due_at = existing.due_at
context = ConversationContext(
thread_id=thread_id,
messages=messages,
@ -197,18 +210,28 @@ class MemoryUpdateQueue:
trace_id=trace_id,
signals=merged_signals,
bypass_watermark=bypass_watermark,
first_at=first_at,
due_at=due_at,
)
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)
return context
def _reset_timer(self) -> None:
"""Reset the debounce timer."""
config = self._config
self._schedule_timer(config.debounce_seconds)
logger.debug("Memory update timer set for %ss", config.debounce_seconds)
def _rearm_timer_locked(self) -> None:
"""Wake for the earliest deadline. Caller must hold ``_lock``."""
if not self._items:
return
if self._processing:
self._reprocess_pending = True
return
now = time.monotonic()
due_at = min(c.due_at for c in self._items)
# Keep an already-armed earlier wakeup. Recreating it on every arrival
# can itself starve a due item under sufficiently frequent traffic.
if self._timer is not None and self._timer_due_at is not None and self._timer_due_at <= due_at:
return
self._schedule_timer(max(0.05, due_at - now))
def _schedule_timer(self, delay_seconds: float) -> None:
"""Schedule queue processing after the provided delay."""
@ -220,19 +243,30 @@ class MemoryUpdateQueue:
delay_seconds,
self._process_queue,
)
# Track the logical deadline, not the rounded wakeup time. Otherwise
# arrivals during the 50 ms rearm floor could keep restarting it.
self._timer_due_at = min((c.due_at for c in self._items), default=time.monotonic() + delay_seconds)
self._timer.daemon = True
self._timer.start()
def _process_queue(self, *, skip_inter_item_delay: bool = False) -> None:
"""Process all queued conversation contexts.
def _process_queue(self, *, skip_inter_item_delay: bool = False, due_only: bool = True) -> None:
"""Process due contexts, or all pending contexts for an explicit flush.
Args:
due_only: Select only eligible contexts. Explicit flushes set
this to False to drain even contexts still debouncing.
skip_inter_item_delay: When set, skip the inter-item rate-limit
``time.sleep``. Intended for the shutdown-drain path
(:meth:`flush_sync`), which races a bounded timeout and should
not waste budget sleeping between items.
"""
with self._lock:
# A canceled callback may already be waiting for the lock. Cancel
# the currently armed wakeup before selecting/rearming deadlines.
if self._timer is not None:
self._timer.cancel()
self._timer = None
self._timer_due_at = None
if self._processing:
# Another worker is already draining the queue. Instead of
# spawning a tight timer spin (repeatedly re-scheduling a
@ -245,11 +279,14 @@ class MemoryUpdateQueue:
if not self._items:
return
now = time.monotonic()
contexts_to_process = [c for c in self._items if not due_only or c.due_at <= now]
if not contexts_to_process:
self._rearm_timer_locked()
return
self._processing = True
self._processing_thread = threading.current_thread()
contexts_to_process = self._items
self._items = []
self._timer = None
self._items = [c for c in self._items if due_only and c.due_at > now]
logger.info("Processing %d queued memory updates", len(contexts_to_process))
@ -301,15 +338,12 @@ class MemoryUpdateQueue:
self._processing_thread = None
# Reschedule inside the lock: ``_schedule_timer`` read-cancels-
# reassigns ``self._timer`` non-atomically, and a concurrent
# ``add``'s ``_reset_timer`` (also under the lock) touches the
# ``add``'s ``_rearm_timer_locked`` (also under the lock) touches the
# same field. Holding the lock makes the reschedule atomic w.r.t.
# ``add``. ``_schedule_timer`` only calls ``Timer.start()`` (no
# synchronous lock acquisition), so this cannot deadlock.
if self._reprocess_pending:
self._reprocess_pending = False
if self._items:
# New work arrived mid-processing: re-run immediately.
self._schedule_timer(0)
self._reprocess_pending = False
self._rearm_timer_locked()
def flush(self, *, skip_inter_item_delay: bool = False) -> None:
"""Force immediate processing of the queue.
@ -325,8 +359,14 @@ class MemoryUpdateQueue:
if self._timer is not None:
self._timer.cancel()
self._timer = None
self._timer_due_at = None
# If another worker is busy, its follow-up must still honor this
# explicit drain even though ordinary deadlines are in the future.
now = time.monotonic()
for context in self._items:
context.due_at = min(context.due_at, now)
self._process_queue(skip_inter_item_delay=skip_inter_item_delay)
self._process_queue(skip_inter_item_delay=skip_inter_item_delay, due_only=False)
def flush_sync(self, timeout: float) -> bool:
"""Best-effort synchronous flush bounded by ``timeout`` seconds.
@ -396,11 +436,14 @@ class MemoryUpdateQueue:
if not finished:
return False
# flush() returned; only report success if no worker raced back in.
return bool(success) and not self.is_processing
return bool(success) and not self.is_processing and self.pending_count == 0
def flush_nowait(self) -> None:
"""Start queue processing immediately in a background thread."""
with self._lock:
now = time.monotonic()
for context in self._items:
context.due_at = min(context.due_at, now)
# Daemon thread: queued messages may be lost if the process exits
# before _process_queue completes. Acceptable for best-effort memory updates.
self._schedule_timer(0)

View File

@ -144,7 +144,7 @@ def test_finishing_worker_reschedules_once_when_reprocess_pending() -> None:
with patch("deerflow.agents.memory.backends.deermem.deermem.core.queue.threading.Timer", return_value=created_timer) as timer_cls:
queue._process_queue()
timer_cls.assert_called_once_with(0, queue._process_queue)
timer_cls.assert_called_once_with(0.05, queue._process_queue)
assert created_timer.daemon is True
created_timer.start.assert_called_once_with()
assert queue._reprocess_pending is False
@ -190,7 +190,7 @@ def test_flush_nowait_is_non_blocking() -> None:
def test_queue_keeps_updates_for_different_agents_in_same_thread() -> None:
queue = _queue()
with patch.object(queue, "_reset_timer"):
with patch.object(queue, "_rearm_timer_locked"):
queue.add(thread_id="thread-1", messages=["agent-a"], agent_name="agent-a")
queue.add(thread_id="thread-1", messages=["agent-b"], agent_name="agent-b")
@ -212,7 +212,7 @@ def test_queue_still_coalesces_updates_for_same_agent_in_same_thread() -> None:
def test_process_queue_updates_different_agents_in_same_thread_separately() -> None:
queue = _queue()
with patch.object(queue, "_reset_timer"):
with patch.object(queue, "_rearm_timer_locked"):
queue.add(thread_id="thread-1", messages=["agent-a"], agent_name="agent-a")
queue.add(thread_id="thread-1", messages=["agent-b"], agent_name="agent-b")

View File

@ -0,0 +1,168 @@
"""Deterministic regressions for per-conversation memory scheduling."""
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"
def test_other_users_cannot_postpone_or_accelerate_a_conversation() -> None:
updater = MagicMock(update_memory=MagicMock(return_value=True))
queue = MemoryUpdateQueue(DeerMemConfig(debounce_seconds=30), updater)
with patch(MODULE + ".time.monotonic") as clock, patch(MODULE + ".threading.Timer") as timer:
clock.return_value = 100
queue.add("shared", ["quiet"], user_id="a")
first_timer = queue._timer
for now in (110, 120, 129):
clock.return_value = now
queue.add("shared", ["chatty"], user_id="b")
# An unrelated arrival must leave the earliest wakeup intact.
assert queue._timer is first_timer
assert timer.call_count == 1
clock.return_value = 130
queue._process_queue()
assert [c.kwargs["user_id"] for c in updater.update_memory.call_args_list] == ["a"]
assert queue.pending_count == 1
assert timer.call_args.args[0] == 29
def test_continuous_same_key_refreshes_reach_max_wait() -> None:
updater = MagicMock(update_memory=MagicMock(return_value=True))
queue = MemoryUpdateQueue(DeerMemConfig(debounce_seconds=30, max_wait_seconds=60), updater)
with patch(MODULE + ".time.monotonic") as clock, patch(MODULE + ".threading.Timer"):
clock.return_value = 100
queue.add("t", ["first"], user_id="a")
for now in (120, 140, 159):
clock.return_value = now
queue.add("t", [str(now)], user_id="a")
assert queue._items[0].first_at == 100
assert queue._items[0].due_at == 160
clock.return_value = 160
queue._process_queue()
updater.update_memory.assert_called_once()
assert updater.update_memory.call_args.kwargs["messages"] == ["159"]
assert queue.pending_count == 0
@pytest.mark.parametrize("method", ["flush", "flush_sync", "flush_nowait"])
def test_explicit_flush_drains_unripe_items(method: str) -> None:
updater = MagicMock(update_memory=MagicMock(return_value=True))
queue = MemoryUpdateQueue(DeerMemConfig(), updater)
with patch(MODULE + ".threading.Timer"), patch(MODULE + ".time.sleep"):
queue.add("t", ["not due"])
if method == "flush_sync":
assert queue.flush_sync(2)
elif method == "flush_nowait":
queue.flush_nowait()
queue._process_queue()
else:
queue.flush()
updater.update_memory.assert_called_once()
assert queue.pending_count == 0
def test_early_wakeup_rearms_with_minimum_delay() -> None:
updater = MagicMock()
queue = MemoryUpdateQueue(DeerMemConfig(debounce_seconds=30), updater)
with patch(MODULE + ".time.monotonic") as clock, patch(MODULE + ".threading.Timer") as timer:
clock.return_value = 100
queue.add("t", [])
clock.return_value = 129.999
queue._process_queue()
assert timer.call_args.args[0] == pytest.approx(0.05)
assert queue.pending_count == 1
updater.update_memory.assert_not_called()
rearmed = queue._timer
timer.reset_mock()
queue.add("other", [])
assert queue._timer is rearmed
timer.assert_not_called()
def test_arrival_during_extraction_keeps_its_deadline() -> None:
updater = MagicMock()
queue = MemoryUpdateQueue(DeerMemConfig(debounce_seconds=30), updater)
with patch(MODULE + ".time.monotonic") as clock, patch(MODULE + ".threading.Timer") as timer:
clock.return_value = 100
queue.add("first", [])
clock.return_value = 130
def arrive(**_kwargs) -> bool:
queue.add("later", [])
queue._process_queue() # Simulate a callback racing the active worker.
return True
updater.update_memory.side_effect = arrive
queue._process_queue()
assert queue.pending_count == 1
assert timer.call_args.args[0] == 30
updater.update_memory.assert_called_once()
def test_emergency_is_immediate_and_cannot_be_delayed_by_normal_add() -> None:
updater = MagicMock(update_memory=MagicMock(return_value=True))
queue = MemoryUpdateQueue(DeerMemConfig(), updater)
with patch(MODULE + ".time.monotonic", return_value=100), patch(MODULE + ".threading.Timer"):
queue.add_nowait("t", ["before compaction"])
queue.add("t", ["normal"])
queue._process_queue()
updater.update_memory.assert_called_once()
assert updater.update_memory.call_args.kwargs["bypass_watermark"] is True
assert queue.pending_count == 1
@pytest.mark.parametrize("method", ["flush", "flush_nowait"])
def test_flush_during_extraction_preserves_forced_drain_on_refresh(method: str) -> None:
updater = MagicMock()
queue = MemoryUpdateQueue(DeerMemConfig(), updater)
with patch(MODULE + ".time.monotonic", return_value=100), patch(MODULE + ".threading.Timer") as timer:
queue.add_nowait("first", [])
def arrive(**_kwargs) -> bool:
queue.add("later", ["old"])
getattr(queue, method)()
if method == "flush_nowait":
queue._process_queue() # The immediate timer fires while busy.
queue.add("later", ["new"])
return True
updater.update_memory.side_effect = arrive
queue._process_queue()
assert queue._items[0].due_at == 100
assert timer.call_args.args[0] == 0.05
updater.update_memory.side_effect = None
updater.update_memory.return_value = True
queue._process_queue()
assert updater.update_memory.call_count == 2
assert updater.update_memory.call_args.kwargs["messages"] == ["new"]
assert queue.pending_count == 0
def test_processed_key_starts_a_new_max_wait_window() -> None:
queue = MemoryUpdateQueue(DeerMemConfig(debounce_seconds=30, max_wait_seconds=60), MagicMock())
with patch(MODULE + ".time.monotonic") as clock, patch(MODULE + ".threading.Timer"):
clock.return_value = 100
queue.add("t", [])
clock.return_value = 130
queue._process_queue()
clock.return_value = 150
queue.add("t", [])
assert queue._items[0].first_at == 150
assert queue._items[0].due_at == 180
@pytest.mark.parametrize("value", [0, -1])
def test_max_wait_must_be_positive(value: int) -> None:
with pytest.raises(ValueError, match="max_wait_seconds"):
DeerMemConfig(max_wait_seconds=value)
def test_max_wait_can_cap_a_longer_debounce_period() -> None:
queue = MemoryUpdateQueue(DeerMemConfig(debounce_seconds=30, max_wait_seconds=10), MagicMock())
with patch(MODULE + ".time.monotonic", return_value=100), patch(MODULE + ".threading.Timer"):
queue.add("t", [])
assert queue._items[0].due_at == 110

View File

@ -22,7 +22,7 @@ def test_conversation_context_user_id_default_none():
def test_queue_add_stores_user_id():
q = _queue()
with patch.object(q, "_reset_timer"):
with patch.object(q, "_rearm_timer_locked"):
q.add(thread_id="t1", messages=["msg"], user_id="alice")
assert len(q._items) == 1
assert q._items[0].user_id == "alice"
@ -33,10 +33,10 @@ def test_queue_process_passes_user_id_to_updater():
mock_updater = MagicMock()
mock_updater.update_memory.return_value = True
q = _queue(mock_updater)
with patch.object(q, "_reset_timer"):
with patch.object(q, "_rearm_timer_locked"):
q.add(thread_id="t1", messages=["msg"], user_id="alice")
q._process_queue()
q.flush()
mock_updater.update_memory.assert_called_once()
assert mock_updater.update_memory.call_args.kwargs["user_id"] == "alice"
@ -44,7 +44,7 @@ def test_queue_process_passes_user_id_to_updater():
def test_queue_keeps_updates_for_different_users_in_same_thread_and_agent():
q = _queue()
with patch.object(q, "_reset_timer"):
with patch.object(q, "_rearm_timer_locked"):
q.add(thread_id="main", messages=["alice update"], agent_name="researcher", user_id="alice")
q.add(thread_id="main", messages=["bob update"], agent_name="researcher", user_id="bob")
@ -55,7 +55,7 @@ def test_queue_keeps_updates_for_different_users_in_same_thread_and_agent():
def test_queue_still_coalesces_updates_for_same_user_thread_and_agent():
q = _queue()
with patch.object(q, "_reset_timer"):
with patch.object(q, "_rearm_timer_locked"):
q.add(thread_id="main", messages=["first"], agent_name="researcher", user_id="alice")
q.add(thread_id="main", messages=["second"], agent_name="researcher", user_id="alice")

View File

@ -20,7 +20,7 @@
# ============================================================================
# Bump this number when the config schema changes.
# Run `make config-upgrade` to merge new fields into your local config.yaml.
config_version: 58
config_version: 59
# Optional operator-owned literal prompt extensions (see CONFIGURATION.md):
# lead_prompt_overlay:
@ -2276,7 +2276,8 @@ summarization:
# filesystems; peers would rebuild or delete each other's index).
# model - LLM config for memory extraction: {provider, model, api_key, base_url,
# temperature}. Omit all fields = no extraction (non-LLM ops still work).
# debounce_seconds - Debounce wait before processing queued updates (default: 30)
# debounce_seconds - Per-conversation quiet period (default: 30)
# max_wait_seconds - Maximum debounce deferral; not a completion timeout (default: 300)
# max_facts - Maximum facts to store (default: 100)
# fact_confidence_threshold - Minimum confidence for storing facts (default: 0.7)
# max_injection_tokens - Token budget for memory injection (default: 2000)
@ -2427,7 +2428,8 @@ memory:
file_lock_timeout_seconds: 10 # per-scope cross-process advisory lock timeout (single-machine local filesystem)
retrieval_adapter: fts5 # fts5 (default), empty to disable, or a dotted RetrievalPort factory(config)
retrieval_index_path: "" # empty = {storage_path}/.retrieval; relative = below storage_path; absolute as-is. Rebuildable derived data: keep it instance-local when Gateway instances share storage_path (the Helm chart uses an emptyDir)
debounce_seconds: 30 # Wait time before processing queued updates
debounce_seconds: 30 # Per-conversation quiet period before extraction is eligible
max_wait_seconds: 300 # Cap debounce deferral from first pending update; not an LLM/completion timeout
# Backpressure cap on pending items. 0 = unlimited. At the cap, new
# non-signal updates are rejected (QueueFull); signal updates are always
# admitted so important memories are never shed.

View File

@ -135,7 +135,7 @@ they resolve from the `secrets` map):
```yaml
config: |
config_version: 58
config_version: 59
models:
- name: gpt-4
use: langchain_openai:ChatOpenAI

View File

@ -364,7 +364,7 @@ ingress:
# -- DeerFlow config.yaml content. Secrets MUST stay as $VAR references — never
# inline literal secret values here. The default enables provisioner sandbox.
config: |
config_version: 58
config_version: 59
log_level: info
recursion_limit: 100
max_recursion_limit: 1000