Merge remote-tracking branch 'origin/master' into paperclip/pr-5732

CHANGELOG.md

@@ -5,6 +5,10 @@
 
 ### Fixed
 
+- **The background reaper and drain daemon threads can no longer be started twice under a race.** `start_session_channel_reaper` and `start_drain_thread` used a check-then-act (`is_alive()` then create) with no lock, so two concurrent callers could both see "not running" and each spawn a thread — the loser's thread was never referenced by the module and ran forever, un-joinable. Both start paths now perform the liveness check, thread creation, and global assignment atomically under a dedicated lifecycle lock. Stop/join still run outside the lock, so there's no deadlock. Thanks @ai-ag2026. (#5783)
+
+- **A long-running server no longer slowly leaks memory from the background-task completion-dedup map.** Each session that ever had a background task complete (delegation-completion delivery, kanban events) left a permanent entry in an in-memory dedup map — including headless completions where no browser tab was ever open — so the map grew unbounded over a long-lived server's lifetime. The session-channel reaper now sweeps that map by delivery lifecycle (dropping a session's entry once its completion has been delivered and is no longer pending), and session deletion prunes any still-pending entry. Exactly-once delivery is unaffected — it's independently guaranteed by the process registry's per-task consumed marker. Thanks @ai-ag2026. (#5781, #4633)
+
 - **Session-lineage reads and the gateway session-list watcher now open the agent's `state.db` read-only.** Reading session lineage/parent-child metadata and the watcher's cheap change-fingerprint opened a *writable* SQLite handle on the live, potentially multi-GB WAL `state.db` while the agent was streaming into it — adding needless checkpoint/lock surface to a pure read. These three read paths now open a read-only (`mode=ro`) connection, falling back to a writable one (with a warning) only if the read-only open fails on an exotic filesystem, and raising rather than letting the fallback create an empty "ghost" `state.db` at a missing path. Same rationale as the #5455 session-listing read-only path. Thanks @ai-ag2026. (#5780, #5455)
 
 - **The MCP servers/tools list now reflects the active profile instead of the default one.** When viewing MCP servers or tools while a non-default profile was active, the list was read from the ambient (default) profile's config rather than the profile you were actually in, so it could show the wrong profile's MCP servers. Both the server-list and tool-list reads are now scoped to the active profile's config home, matching where the MCP write handlers save. Deployments that point config at an external path via `HERMES_CONFIG_PATH` now also read that override consistently on the active-home path. Thanks @rodboev. (#5772, #5619)

api/background_process.py

@@ -57,6 +57,14 @@ _REAPER_THREAD: Optional[threading.Thread] = None
 _REAPER_STOP = threading.Event()
 _REAPER_INTERVAL_SECS = 60.0
 
+# Serializes the check-then-start of the module's daemon threads
+# (``start_drain_thread`` / ``start_session_channel_reaper``). Without it two
+# concurrent callers can both observe ``is_alive() == False`` and each spawn a
+# thread; the loser's thread is never referenced by the module global and runs
+# forever, un-joinable. A dedicated lock (not the purpose-bound
+# ``SESSION_CHANNELS_LOCK`` / ``_EMIT_COALESCE_LOCK``) keeps this narrow.
+_THREAD_LIFECYCLE_LOCK = threading.Lock()
+
 # T3: per-session coalesce gate for the public bg_task_complete SSE emit.
 # The server-side wakeup path remains immediate; only the browser-observation
 # frame is throttled so a burst of background task completions does not flood an
@@ -377,6 +385,32 @@ def _reaper_loop() -> None:
                     for sid in collected:
                         _LAST_EMIT_TS.pop(sid, None)
                 logger.debug("SessionChannel reaper collected: %s", collected)
+            # Sweep the per-session completion-dedup map by DELIVERY lifecycle,
+            # not channel collection. ``BG_TASK_COMPLETE_EVENTS_SEEN`` gains a
+            # ``session_id -> set[process_id]`` entry the first time a bg task
+            # completes for a session — in ``_process_one``, whether or not any
+            # tab/SSE channel ever existed — and is otherwise never deleted, so it
+            # grows unbounded. Coupling the prune to channel collection (an
+            # earlier version of this fix) missed the dominant case: a headless
+            # completion (task fires, tab closed or never opened) has no channel
+            # to collect. Instead, once a completion has been drained (its
+            # ``session_id`` removed from ``PENDING_BG_TASK_COMPLETIONS``), the
+            # short ``_move_to_finished`` dedup window is closed and the entry is
+            # pure leak — so sweep every delivered (not-pending) session here,
+            # every tick. The registry's own per-``process_id``
+            # ``_completion_consumed`` gate remains the primary idempotency
+            # backstop, so sweeping a delivered session's set can never resurrect
+            # an already-delivered completion (even in the tiny window between
+            # this module's ``SEEN.add`` and ``PENDING.add`` in ``_process_one``).
+            from api import config as _cfg
+
+            with _cfg.BG_TASK_COMPLETE_EVENTS_SEEN_LOCK:
+                for sid in [
+                    s
+                    for s in _cfg.BG_TASK_COMPLETE_EVENTS_SEEN
+                    if s not in _cfg.PENDING_BG_TASK_COMPLETIONS
+                ]:
+                    _cfg.BG_TASK_COMPLETE_EVENTS_SEEN.pop(sid, None)
         except Exception:
             logger.warning("SessionChannel reaper iteration failed", exc_info=True)
         # Wait but wake up promptly on stop.
@@ -387,16 +421,17 @@ def _reaper_loop() -> None:
 def start_session_channel_reaper() -> bool:
     """Start the SessionChannel reaper thread. Idempotent; returns True on first start."""
     global _REAPER_THREAD
-    if _REAPER_THREAD is not None and _REAPER_THREAD.is_alive():
-        return False
-    _REAPER_STOP.clear()
-    _REAPER_THREAD = threading.Thread(
-        target=_reaper_loop,
-        name="hermes-webui-session-channel-reaper",
-        daemon=True,
-    )
-    _REAPER_THREAD.start()
-    return True
+    with _THREAD_LIFECYCLE_LOCK:
+        if _REAPER_THREAD is not None and _REAPER_THREAD.is_alive():
+            return False
+        _REAPER_STOP.clear()
+        _REAPER_THREAD = threading.Thread(
+            target=_reaper_loop,
+            name="hermes-webui-session-channel-reaper",
+            daemon=True,
+        )
+        _REAPER_THREAD.start()
+        return True
 
 
 def stop_session_channel_reaper(timeout: float = 2.0) -> None:
@@ -1333,19 +1368,35 @@ def unregister_process_session(session_key: str) -> None:
         _cfg.PROCESS_SESSION_INDEX.pop(str(session_key), None)
 
 
+def forget_bg_task_completion_dedup(session_id: str) -> None:
+    """Drop a session's ``BG_TASK_COMPLETE_EVENTS_SEEN`` entry.
+
+    Called on session deletion so a session deleted while a completion is still
+    pending (undelivered) — which the reaper's delivery-gated sweep deliberately
+    keeps — can't leak its dedup set forever. Safe for unknown ids (no-op).
+    """
+    if not session_id:
+        return
+    from api import config as _cfg
+
+    with _cfg.BG_TASK_COMPLETE_EVENTS_SEEN_LOCK:
+        _cfg.BG_TASK_COMPLETE_EVENTS_SEEN.pop(str(session_id), None)
+
+
 def start_drain_thread() -> bool:
     """Start the background drain thread idempotently. Returns True on first start."""
     global _DRAIN_THREAD
-    if _DRAIN_THREAD is not None and _DRAIN_THREAD.is_alive():
-        return False
-    _DRAIN_STOP.clear()
-    _DRAIN_THREAD = threading.Thread(
-        target=_drain_loop,
-        name="hermes-webui-bg-task-complete-drain",
-        daemon=True,
-    )
-    _DRAIN_THREAD.start()
-    return True
+    with _THREAD_LIFECYCLE_LOCK:
+        if _DRAIN_THREAD is not None and _DRAIN_THREAD.is_alive():
+            return False
+        _DRAIN_STOP.clear()
+        _DRAIN_THREAD = threading.Thread(
+            target=_drain_loop,
+            name="hermes-webui-bg-task-complete-drain",
+            daemon=True,
+        )
+        _DRAIN_THREAD.start()
+        return True
 
 
 def stop_drain_thread(timeout: float = 2.0) -> None:

api/routes.py

@@ -13865,6 +13865,15 @@ def handle_post(handler, parsed) -> bool:
         # Lock entries in SESSION_AGENT_LOCKS forever.
         with SESSION_AGENT_LOCKS_LOCK:
             SESSION_AGENT_LOCKS.pop(sid, None)
+        # Prune the completion-dedup entry too. The reaper sweeps it once the
+        # completion is delivered (drained from PENDING); a session deleted
+        # while a completion is still pending would otherwise keep its entry.
+        try:
+            from api.background_process import forget_bg_task_completion_dedup
+
+            forget_bg_task_completion_dedup(sid)
+        except Exception:
+            logger.debug("Failed to prune bg-task dedup entry for deleted session %s", sid)
         try:
             from api.terminal import close_terminal
             close_terminal(sid)

tests/test_bg_dedup_reaper_prune.py

@@ -0,0 +1,122 @@
+"""Regression test for the ``BG_TASK_COMPLETE_EVENTS_SEEN`` memory leak (#4633).
+
+The per-session completion-dedup map gained one ``session_id -> set[process_id]``
+entry the first time a background task completed for a session and was never
+removed anywhere in the repo, so it grew unbounded for the server lifetime.
+
+The dedup entry is created in ``_process_one`` for EVERY completion, whether or
+not any SSE channel/tab exists — so coupling the prune to channel collection
+would miss the dominant headless case (task fires, tab closed or never opened).
+Instead the reaper sweeps the map by DELIVERY lifecycle: once a completion has
+been drained (its ``session_id`` removed from ``PENDING_BG_TASK_COMPLETIONS``),
+the short dedup window is closed and the entry is swept. Session deletion prunes
+it too, covering a session deleted while a completion is still pending.
+"""
+
+from __future__ import annotations
+
+import time
+from pathlib import Path
+
+import pytest
+
+
+REPO_ROOT = Path(__file__).resolve().parents[1]
+
+
+def test_reaper_source_sweeps_dedup_map_by_delivery():
+    """The reaper cleanup must sweep BG_TASK_COMPLETE_EVENTS_SEEN under its lock,
+    gated on PENDING_BG_TASK_COMPLETIONS (delivery), not channel collection."""
+    src = (REPO_ROOT / "api" / "background_process.py").read_text(encoding="utf-8")
+    reaper = src[src.index("def _reaper_loop("):]
+    reaper = reaper[: reaper.index("\ndef ", 1)]
+    assert "BG_TASK_COMPLETE_EVENTS_SEEN_LOCK" in reaper, (
+        "reaper must take the dedup-map lock"
+    )
+    assert "BG_TASK_COMPLETE_EVENTS_SEEN.pop(" in reaper, (
+        "reaper must pop dedup entries"
+    )
+    assert "PENDING_BG_TASK_COMPLETIONS" in reaper, (
+        "the sweep must be gated on delivery (PENDING_BG_TASK_COMPLETIONS), "
+        "not on channel collection — otherwise headless completions leak"
+    )
+
+
+def _run_reaper_once(bp, cfg):
+    """Start the real reaper, wait for one sweep pass, stop it."""
+    bp.start_session_channel_reaper()
+    time.sleep(0.4)  # first pass runs immediately (interval sleep is at loop tail)
+    bp.stop_session_channel_reaper()
+
+
+def test_reaper_sweeps_delivered_dedup_entry_without_any_channel():
+    """The dominant headless case: a completed+drained task with NO SSE channel
+    ever — its dedup entry must still be swept (this is what the channel-coupled
+    version missed)."""
+    from api import background_process as bp
+    from api import config as cfg
+
+    sid = "sess-headless-4633"
+    with cfg.BG_TASK_COMPLETE_EVENTS_SEEN_LOCK:
+        cfg.BG_TASK_COMPLETE_EVENTS_SEEN[sid] = {"proc-1"}
+    cfg.PENDING_BG_TASK_COMPLETIONS.discard(sid)  # delivered/drained
+    # No SessionChannel for sid — nothing for channel collection to reap.
+    try:
+        with bp.SESSION_CHANNELS_LOCK:
+            assert sid not in bp.SESSION_CHANNELS
+        _run_reaper_once(bp, cfg)
+        with cfg.BG_TASK_COMPLETE_EVENTS_SEEN_LOCK:
+            assert sid not in cfg.BG_TASK_COMPLETE_EVENTS_SEEN, (
+                "reaper did not sweep the delivered headless dedup entry"
+            )
+    finally:
+        bp.stop_session_channel_reaper()
+        with cfg.BG_TASK_COMPLETE_EVENTS_SEEN_LOCK:
+            cfg.BG_TASK_COMPLETE_EVENTS_SEEN.pop(sid, None)
+
+
+def test_reaper_retains_dedup_entry_while_completion_pending():
+    """An UNDELIVERED completion (still in PENDING) must keep its dedup entry —
+    the _move_to_finished dedup window is still open."""
+    from api import background_process as bp
+    from api import config as cfg
+
+    sid = "sess-pending-4633"
+    with cfg.BG_TASK_COMPLETE_EVENTS_SEEN_LOCK:
+        cfg.BG_TASK_COMPLETE_EVENTS_SEEN[sid] = {"proc-2"}
+    cfg.PENDING_BG_TASK_COMPLETIONS.add(sid)  # undelivered
+    try:
+        _run_reaper_once(bp, cfg)
+        with cfg.BG_TASK_COMPLETE_EVENTS_SEEN_LOCK:
+            assert sid in cfg.BG_TASK_COMPLETE_EVENTS_SEEN, (
+                "reaper swept a still-pending (undelivered) dedup entry"
+            )
+    finally:
+        bp.stop_session_channel_reaper()
+        cfg.PENDING_BG_TASK_COMPLETIONS.discard(sid)
+        with cfg.BG_TASK_COMPLETE_EVENTS_SEEN_LOCK:
+            cfg.BG_TASK_COMPLETE_EVENTS_SEEN.pop(sid, None)
+
+
+def test_session_delete_prunes_dedup_entry():
+    """Deleting a session prunes its dedup entry even if the completion never
+    delivered (so a delete-while-pending session can't leak forever)."""
+    from api import background_process as bp
+    from api import config as cfg
+
+    sid = "sess-delete-4633"
+    with cfg.BG_TASK_COMPLETE_EVENTS_SEEN_LOCK:
+        cfg.BG_TASK_COMPLETE_EVENTS_SEEN[sid] = {"proc-3"}
+    cfg.PENDING_BG_TASK_COMPLETIONS.add(sid)  # undelivered, but being deleted
+    try:
+        bp.forget_bg_task_completion_dedup(sid)
+        with cfg.BG_TASK_COMPLETE_EVENTS_SEEN_LOCK:
+            assert sid not in cfg.BG_TASK_COMPLETE_EVENTS_SEEN
+    finally:
+        cfg.PENDING_BG_TASK_COMPLETIONS.discard(sid)
+        with cfg.BG_TASK_COMPLETE_EVENTS_SEEN_LOCK:
+            cfg.BG_TASK_COMPLETE_EVENTS_SEEN.pop(sid, None)
+
+
+if __name__ == "__main__":  # pragma: no cover - manual invocation
+    raise SystemExit(pytest.main([__file__, "-v"]))

tests/test_bg_thread_start_lock.py

@@ -0,0 +1,98 @@
+"""Regression test for the thread-start check-then-act race in background_process.
+
+``start_drain_thread`` and ``start_session_channel_reaper`` checked
+``is_alive()`` and then created + started the daemon thread WITHOUT a lock, so
+two concurrent callers could both observe "not alive" and each spawn a thread.
+The loser's thread was never stored in the module global and ran forever,
+un-joinable. Both check-then-start sequences are now serialized under
+``_THREAD_LIFECYCLE_LOCK``, so exactly one thread is ever created.
+
+The test drives N callers through a barrier so they hit the check-and-start
+simultaneously, and asserts exactly one caller reports "started" (returns True)
+and exactly one live thread exists.
+"""
+
+from __future__ import annotations
+
+import threading
+
+import pytest
+
+
+# (start_fn, stop_fn, loop_attr, thread_attr, thread_name_fragment)
+_CASES = [
+    (
+        "start_drain_thread",
+        "stop_drain_thread",
+        "_drain_loop",
+        "_DRAIN_THREAD",
+        "bg-task-complete-drain",
+    ),
+    (
+        "start_session_channel_reaper",
+        "stop_session_channel_reaper",
+        "_reaper_loop",
+        "_REAPER_THREAD",
+        "session-channel-reaper",
+    ),
+]
+
+
+@pytest.mark.parametrize(
+    "start_name,stop_name,loop_attr,thread_attr,name_frag", _CASES
+)
+def test_concurrent_start_creates_exactly_one_thread(
+    monkeypatch, start_name, stop_name, loop_attr, thread_attr, name_frag
+):
+    from api import background_process as bp
+
+    release = threading.Event()
+
+    def _blocking_loop() -> None:
+        # Keep the started thread alive during the assertions so a genuinely
+        # started thread reads as is_alive()==True deterministically.
+        release.wait(5.0)
+
+    monkeypatch.setattr(bp, loop_attr, _blocking_loop, raising=True)
+    # Fresh slate: no pre-existing thread reference.
+    monkeypatch.setattr(bp, thread_attr, None, raising=True)
+
+    start_fn = getattr(bp, start_name)
+    n = 12
+    barrier = threading.Barrier(n)
+    results: list[bool] = []
+    results_lock = threading.Lock()
+
+    def _caller() -> None:
+        barrier.wait()
+        started = start_fn()
+        with results_lock:
+            results.append(started)
+
+    callers = [threading.Thread(target=_caller) for _ in range(n)]
+    try:
+        for c in callers:
+            c.start()
+        for c in callers:
+            c.join(timeout=5.0)
+
+        # Exactly one caller won the create; the rest saw a live thread.
+        assert sum(1 for r in results if r) == 1, (
+            f"expected exactly one start to win, got {results}"
+        )
+        # Exactly one live daemon thread carrying this loop's name.
+        live = [
+            t
+            for t in threading.enumerate()
+            if name_frag in t.name and t.is_alive()
+        ]
+        assert len(live) == 1, f"expected 1 live {name_frag} thread, got {len(live)}"
+        # The module global points at that one live thread.
+        assert getattr(bp, thread_attr) is live[0]
+    finally:
+        release.set()
+        getattr(bp, stop_name)()
+
+
+if __name__ == "__main__":  # pragma: no cover - manual invocation
+    raise SystemExit(pytest.main([__file__, "-v"]))