fix(memory): replace busy timer spin with deferred single re-run flag (#4073)

When _process_queue found another worker already processing, it called
_schedule_timer(0), spawning a fresh Timer thread immediately and looping
tightly (spawn -> busy -> reschedule -> spawn) until the active worker
finished. Replace this with a _reprocess_pending flag: a concurrent caller
sets the flag and returns, and the active worker reschedules exactly once
in its finally block when work remains. Reset the flag in clear().

backend/packages/harness/deerflow/agents/memory/queue.py

@@ -42,6 +42,7 @@ class MemoryUpdateQueue:
         self._lock = threading.Lock()
         self._timer: threading.Timer | None = None
         self._processing = False
+        self._reprocess_pending = False
 
     @staticmethod
     def _queue_key(
@@ -181,8 +182,12 @@ class MemoryUpdateQueue:
 
         with self._lock:
             if self._processing:
-                # Preserve immediate flush semantics even if another worker is active.
-                self._schedule_timer(0)
+                # Another worker is already draining the queue. Instead of
+                # spawning a tight timer spin (repeatedly re-scheduling a
+                # 0-delay Timer thread while busy), defer a single re-run: the
+                # active worker checks this flag in its finally block and
+                # reschedules once if work remains.
+                self._reprocess_pending = True
                 return
 
             if not self._queue:
@@ -232,6 +237,10 @@ class MemoryUpdateQueue:
         finally:
             with self._lock:
                 self._processing = False
+                if self._reprocess_pending:
+                    self._reprocess_pending = False
+                    if self._queue:
+                        self._schedule_timer(0)
 
     def flush(self) -> None:
         """Force immediate processing of the queue.
@@ -263,6 +272,7 @@ class MemoryUpdateQueue:
                 self._timer = None
             self._queue.clear()
             self._processing = False
+            self._reprocess_pending = False
 
     @property
     def pending_count(self) -> int:

backend/tests/test_memory_queue.py

@@ -134,17 +134,70 @@ def test_add_nowait_cancels_existing_timer_and_starts_immediate_timer() -> None:
     created_timer.start.assert_called_once_with()
 
 
-def test_process_queue_reschedules_immediately_when_already_processing() -> None:
+def test_process_queue_defers_reprocess_when_already_processing() -> None:
+    """When a timer fires while a worker is active, ``_process_queue`` must set the
+    deferred-rerun flag instead of spinning up a tight 0-delay Timer chain.
+
+    The old behavior re-scheduled a 0-delay Timer on every re-entry while busy,
+    burning a fresh thread each time. The fix defers a single re-run via
+    ``_reprocess_pending`` that the finishing worker honors once.
+    """
     queue = MemoryUpdateQueue()
     queue._processing = True
-    created_timer = MagicMock()
 
-    with patch("deerflow.agents.memory.queue.threading.Timer", return_value=created_timer) as timer_cls:
+    with patch("deerflow.agents.memory.queue.threading.Timer") as timer_cls:
+        queue._process_queue()
+
+    timer_cls.assert_not_called()
+    assert queue._reprocess_pending is True
+
+
+def test_finishing_worker_reschedules_once_when_reprocess_pending() -> None:
+    """A worker that finishes with ``_reprocess_pending`` set and work still queued
+    schedules exactly one follow-up run (not a per-arrival timer spin)."""
+    queue = MemoryUpdateQueue()
+    queue._queue = [ConversationContext(thread_id="thread-1", messages=["first"], agent_name="lead_agent")]
+    queue._reprocess_pending = True
+    created_timer = MagicMock()
+    mock_updater = MagicMock()
+
+    def _enqueue_more_while_processing(**_kwargs) -> bool:
+        # Simulate a new update arriving mid-processing so the finally block sees
+        # remaining work and reschedules exactly once.
+        queue._queue.append(ConversationContext(thread_id="thread-2", messages=["second"], agent_name="lead_agent"))
+        return True
+
+    mock_updater.update_memory.side_effect = _enqueue_more_while_processing
+
+    with (
+        patch("deerflow.agents.memory.updater.MemoryUpdater", return_value=mock_updater),
+        patch("deerflow.agents.memory.queue.threading.Timer", return_value=created_timer) as timer_cls,
+    ):
         queue._process_queue()
 
     timer_cls.assert_called_once_with(0, queue._process_queue)
     assert created_timer.daemon is True
     created_timer.start.assert_called_once_with()
+    assert queue._reprocess_pending is False
+
+
+def test_finishing_worker_does_not_reschedule_when_no_work_remains() -> None:
+    """The deferred re-run is cleared even when nothing is left to process, so a
+    stray flag never leaves a dangling ``_reprocess_pending``."""
+    queue = MemoryUpdateQueue()
+    queue._queue = [ConversationContext(thread_id="thread-1", messages=["only"], agent_name="lead_agent")]
+    queue._reprocess_pending = True
+    mock_updater = MagicMock()
+    mock_updater.update_memory.return_value = True
+
+    with (
+        patch("deerflow.agents.memory.updater.MemoryUpdater", return_value=mock_updater),
+        patch("deerflow.agents.memory.queue.threading.Timer") as timer_cls,
+    ):
+        queue._process_queue()
+
+    timer_cls.assert_not_called()
+    assert queue._reprocess_pending is False
 
 
 def test_flush_nowait_is_non_blocking() -> None: