diff --git a/README.md b/README.md
index 2e1af56ab3..1c888e5762 100644
--- a/README.md
+++ b/README.md
@@ -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.
diff --git a/backend/packages/harness/deerflow/agents/memory/AGENTS.md b/backend/packages/harness/deerflow/agents/memory/AGENTS.md
index 46bf5c7ffd..10cdc1f12d 100644
--- a/backend/packages/harness/deerflow/agents/memory/AGENTS.md
+++ b/backend/packages/harness/deerflow/agents/memory/AGENTS.md
@@ -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.
diff --git a/backend/packages/harness/deerflow/agents/memory/backends/deermem/deermem/config.py b/backend/packages/harness/deerflow/agents/memory/backends/deermem/deermem/config.py
index 30bd6a4381..2801816979 100644
--- a/backend/packages/harness/deerflow/agents/memory/backends/deermem/deermem/config.py
+++ b/backend/packages/harness/deerflow/agents/memory/backends/deermem/deermem/config.py
@@ -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,
diff --git a/backend/packages/harness/deerflow/agents/memory/backends/deermem/deermem/core/queue.py b/backend/packages/harness/deerflow/agents/memory/backends/deermem/deermem/core/queue.py
index 49dc8d9719..16ae3266df 100644
--- a/backend/packages/harness/deerflow/agents/memory/backends/deermem/deermem/core/queue.py
+++ b/backend/packages/harness/deerflow/agents/memory/backends/deermem/deermem/core/queue.py
@@ -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)
diff --git a/backend/tests/test_memory_queue.py b/backend/tests/test_memory_queue.py
index 0597a041bf..3697133a85 100644
--- a/backend/tests/test_memory_queue.py
+++ b/backend/tests/test_memory_queue.py
@@ -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")
 
diff --git a/backend/tests/test_memory_queue_deadlines.py b/backend/tests/test_memory_queue_deadlines.py
new file mode 100644
index 0000000000..48ec738610
--- /dev/null
+++ b/backend/tests/test_memory_queue_deadlines.py
@@ -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
diff --git a/backend/tests/test_memory_queue_user_isolation.py b/backend/tests/test_memory_queue_user_isolation.py
index 033f6f19c8..1821e4b62d 100644
--- a/backend/tests/test_memory_queue_user_isolation.py
+++ b/backend/tests/test_memory_queue_user_isolation.py
@@ -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")
 
diff --git a/config.example.yaml b/config.example.yaml
index 288cf2d932..dc30df1031 100644
--- a/config.example.yaml
+++ b/config.example.yaml
@@ -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.
diff --git a/deploy/helm/deer-flow/README.md b/deploy/helm/deer-flow/README.md
index cfde3fa142..10fa36656d 100644
--- a/deploy/helm/deer-flow/README.md
+++ b/deploy/helm/deer-flow/README.md
@@ -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
diff --git a/deploy/helm/deer-flow/values.yaml b/deploy/helm/deer-flow/values.yaml
index eb6f913c91..228e283693 100644
--- a/deploy/helm/deer-flow/values.yaml
+++ b/deploy/helm/deer-flow/values.yaml
@@ -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