Process hangs on exit due to stuck worker threads

Fixes on record

Merged to main. A stamp counts the issues a fix is linked to.

• #54111 fix(gateway): route SystemExit exit paths through the force-exit backstop too (#53107)

  merged to main

  files: gateway/run.py, tests/gateway/test_gateway_process_exit.py

• #54122 fix(gateway): force-exit after graceful shutdown so a wedged worker can't hang the process (#53107)

  merged to main

  files: gateway/run.py, infographic/gateway-force-exit-53107/infographic.png, tests/gateway/test_gateway_process_exit.py

• #57000 fix(cli): reliable interrupts, bounded exit, and immediate shutdown feedback

  merged to main

  files: agent/memory_manager.py, agent/tool_executor.py, cli.py, tests/cli/test_cli_interrupt_ack_race.py, tests/run_agent/test_run_agent.py, tests/tools/test_daemon_pool.py, tools/async_delegation.py, tools/daemon_pool.py, tools/delegate_tool.py, tools/skills_hub.py

  Show the diff

  • agent/memory_manager.py

    @@ -651,7 +651,12 @@ def _get_sync_executor(self) -> Optional[ThreadPoolExecutor]:
             with self._sync_executor_lock:
                 if self._sync_executor is None:
                     try:
    -                    self._sync_executor = ThreadPoolExecutor(
    +                    # Daemon workers (see tools.daemon_pool): a provider wedged
    +                    # on a network call must never block interpreter exit —
    +                    # stdlib ThreadPoolExecutor's atexit hook would join it
    +                    # unconditionally even after shutdown(wait=False).
    +                    from tools.daemon_pool import DaemonThreadPoolExecutor
    +                    self._sync_executor = DaemonThreadPoolExecutor(
                             max_workers=1,
                             thread_name_prefix="mem-sync",
                         )

  • agent/tool_executor.py

    @@ -638,7 +638,13 @@ def _run_tool(index, tool_call, function_name, function_args, middleware_trace):
             deadline = time.monotonic() + timeout_s if timeout_s is not None else None
             if runnable_calls:
                 max_workers = min(len(runnable_calls), _MAX_TOOL_WORKERS)
    -            executor = concurrent.futures.ThreadPoolExecutor(max_workers=max_workers)
    +            # Daemon workers: an interrupted/timed-out batch is abandoned with
    +            # shutdown(wait=False), but stdlib ThreadPoolExecutor workers are
    +            # non-daemon and registered in concurrent.futures' atexit hook,
    +            # which joins them unconditionally — so one wedged tool thread
    +            # would block interpreter exit forever (multi-minute CLI exits).
    +            from tools.daemon_pool import DaemonThreadPoolExecutor
    +            executor = DaemonThreadPoolExecutor(max_workers=max_workers)
                 abandon_executor = False
                 try:
                     for submit_index, (i, tc, name, args) in enumerate(runnable_calls):

  • cli.py

    @@ -982,6 +982,71 @@ def _prepare_deferred_agent_startup() -> None:
                 exc_info=True,
             )
     
    +def _arm_exit_watchdog(timeout_s: float | None = None) -> None:
    +    """Guarantee the process actually exits once shutdown has begun.
    +
    +    Two hang classes have kept "dead" CLI processes alive for minutes:
    +
    +      1. A cleanup step wedged on network I/O (memory provider
    +         ``on_session_end``, MCP teardown, remote terminal cleanup).
    +      2. Interpreter teardown blocked joining non-daemon threads —
    +         stdlib ``ThreadPoolExecutor`` workers are joined unconditionally
    +         by ``concurrent.futures``' atexit hook even after
    +         ``shutdown(wait=False)``, so one tool thread wedged on a socket
    +         held the process open forever (#27563 class).
    +
    +    The shared daemon pool (``tools.daemon_pool``) removes the main cause
    +    of (2); this watchdog is the backstop for both. It arms a daemon
    +    timer when ``_run_cleanup`` starts; if the process is still alive
    +    after ``timeout_s`` it flushes logging/stdio and calls ``os._exit(0)``.
    +    Daemon threads keep running through ``Py_FinalizeEx``'s thread joins,
    +    so the timer fires even when the main thread is stuck in teardown.
    +
    +    Tune with ``HERMES_EXIT_WATCHDOG_S`` (seconds); ``0`` disables.
    +    """
    +    if timeout_s is None:
    +        try:
    +            timeout_s = float(os.getenv("HERMES_EXIT_WATCHDOG_S", "30"))
    +        except (TypeError, ValueError):
    +            timeout_s = 30.0
    +    if timeout_s <= 0:
    +        return
    +    # Never arm under pytest: tests invoke _run_cleanup() directly and a
    +    # 30s-delayed os._exit(0) would silently kill the test worker.
    +    if os.environ.get("PYTEST_CURRENT_TEST"):
    +        return
    +
    +    def _watchdog():
    +        time.sleep(timeout_s)
    +        # Still alive — cleanup or interpreter teardown is wedged.
    +        try:
    +            logger.warning(
    +                "Exit watchdog fired after %.0fs — forcing process exit "
    +                "(a cleanup step or non-daemon thread is wedged).",
    +                timeout_s,
    +            )
    +        except Exception:
    +            pass
    +        try:
    +            import logging as _lg
    +            _lg.shutdown()
    +        except Exception:
    +            pass
    +        for _stream in (sys.stdout, sys.stderr):
    +            try:
    +                _stream.flush()
    +            except Exception:
    +                pass
    +        os._exit(0)
    +
    +    try:
    +        threading.Thread(
    +            target=_watchdog, daemon=True, name="exit-watchdog"
    +        ).start()
    +    except Exception:
    +        pass  # best-effort — never block shutdown on watchdog setup
    +
    +
     def _run_cleanup(*, notify_session_finalize: bool = True):
         """Run resource cleanup exactly once."""
         global _cleanup_done
    @@ -989,6 +1054,11 @@ def _run_cleanup(*, notify_session_finalize: bool = True):
             return
         _cleanup_done = True
     
    +    # Bound total shutdown time: if cleanup (or the interpreter's
    +    # thread-join teardown after it) wedges, force-exit instead of
    +    # leaving a zombie CLI holding the terminal for minutes.
    +    _arm_exit_watchdog()
    +
         # Reset terminal input modes first, before the slower resource teardown
         # below (MCP / browser / memory shutdown can take seconds). On Ctrl+C the
         # user's terminal becomes usable immediately, and a later step raising
    @@ -12161,8 +12231,15 @@ def run_agent():
                             if interrupt_msg:
                                 # If clarify is active, the Enter handler routes
                                 # input directly; this queue shouldn't have anything.
    -                            # But if it does (race condition), don't interrupt.
    +                            # But if it does (race condition), don't interrupt —
    +                            # and don't drop the message either: park it in
    +                            # _pending_input so it runs as the next turn.
                                 if self._clarify_state or self._clarify_freetext:
    +                                try:
    +                                    self._pending_input.put(interrupt_msg)
    +                                except Exception:
    +                                    pass
    +                                interrupt_msg = None
                                     continue
                                 print("\n⚡ New message detected, interrupting...")
                                 # Signal TTS to stop on interrupt
    @@ -12334,6 +12411,33 @@ def run_agent():
                     # Add indicator that we were interrupted
                     if response and pending_message:
                         response = response + "\n\n---\n_[Interrupted - processing new message]_"
    +            elif interrupt_msg:
    +                # We fired agent.interrupt(interrupt_msg) but the turn result
    +                # doesn't acknowledge it. Two ways this happens, both racy:
    +                #   1. The agent thread had already passed its last interrupt
    +                #      check (or finished) when the interrupt landed — the turn
    +                #      completed normally and finalize_turn() never saw the flag.
    +                #   2. The 10s post-interrupt wait above expired and we
    +                #      abandoned the daemon thread; `result` is still None.
    +                # In both cases the user's message must NOT be dropped —
    +                # re-queue it as the next turn (#interrupt-vacuumed-into-void).
    +                pending_message = interrupt_msg
    +                # If the interrupt landed after finalize_turn()'s
    +                # clear_interrupt(), the stale flag would instantly abort the
    +                # NEXT turn at its first loop check. Clear it now that we've
    +                # claimed the message — but ONLY if the agent thread actually
    +                # exited. If it's still alive (abandoned after the 10s wait),
    +                # the flag is what makes the wedged tool eventually unwind;
    +                # clearing it would un-signal that thread.
    +                try:
    +                    if (
    +                        not agent_thread.is_alive()
    +                        and self.agent
    +                        and getattr(self.agent, "_interrupt_requested", False)
    +                    ):
    +                        self.agent.clear_interrupt()
    +                except Exception:
    +                    pass
     
                 response_previewed = result.get("response_previewed", False) if result else False
     
    @@ -15240,6 +15344,15 @@ def new_event_loop(self):
             finally:
                 self._should_exit = True
                 self._pet_stop_anim()
    +            # Immediate feedback: prompt_toolkit has just torn down the input
    +            # box + status bar, so without a line here the terminal sits
    +            # silent for the whole cleanup window (session flush, memory
    +            # shutdown, MCP/browser/terminal teardown) and the exit looks
    +            # hung. Print before any potentially-slow step.
    +            try:
    +                print(f"{_DIM}Shutting down… (finalizing session){_RST}", flush=True)
    +            except Exception:
    +                pass
                 # Interrupt the agent immediately so its daemon thread stops making
                 # API calls and exits promptly (agent_thread is daemon, so the
                 # process will exit once the main thread finishes, but interrupting

  • tests/cli/test_cli_interrupt_ack_race.py

    @@ -0,0 +1,194 @@
    +"""Regression tests for the CLI interrupt-acknowledgement race.
    +
    +Symptom (user report, July 2026): interrupting an active turn is
    +unreliable — the interrupt message is sometimes "vacuumed into the void".
    +
    +Root cause: ``HermesCLI.chat()`` fires ``agent.interrupt(msg)`` from its
    +monitor loop, but only re-queued the message when the turn RESULT carried
    +``interrupted=True``. Two races defeat that:
    +
    +  1. The agent thread passes its last ``_interrupt_requested`` check (or
    +     finishes entirely) just before the interrupt lands — the turn
    +     completes "normally", ``finalize_turn()`` never acknowledges the
    +     interrupt, and the user's message was silently dropped.
    +  2. Worse, when the interrupt lands *after* ``finalize_turn()``'s
    +     ``clear_interrupt()``, the stale ``_interrupt_requested`` flag
    +     survives on the agent and instantly aborts the NEXT turn at its
    +     first loop check.
    +
    +The fix: when ``chat()`` consumed an ``interrupt_msg`` but the result
    +doesn't acknowledge the interrupt, re-queue the message as the next turn
    +and clear the stale agent flag (only when the agent thread has exited).
    +"""
    +
    +from __future__ import annotations
    +
    +import importlib
    +import queue
    +import sys
    +import time
    +from unittest.mock import MagicMock, patch
    +
    +
    +def _make_cli():
    +    """Build a HermesCLI with prompt_toolkit stubbed (same pattern as
    +    test_cli_interrupt_drain_regression.py)."""
    +    _clean_config = {
    +        "model": {
    +            "default": "anthropic/claude-opus-4.6",
    +            "base_url": "https://openrouter.ai/api/v1",
    +            "provider": "auto",
    +        },
    +        "display": {"compact": False, "tool_progress": "all"},
    +        "agent": {},
    +        "terminal": {"env_type": "local"},
    +    }
    +    clean_env = {"LLM_MODEL": "", "HERMES_MAX_ITERATIONS": ""}
    +    prompt_toolkit_stubs = {
    +        "prompt_toolkit": MagicMock(),
    +        "prompt_toolkit.history": MagicMock(),
    +        "prompt_toolkit.styles": MagicMock(),
    +        "prompt_toolkit.patch_stdout": MagicMock(),
    +        "prompt_toolkit.application": MagicMock(),
    +        "prompt_toolkit.layout": MagicMock(),
    +        "prompt_toolkit.layout.processors": MagicMock(),
    +        "prompt_toolkit.filters": MagicMock(),
    +        "prompt_toolkit.layout.dimension": MagicMock(),
    +        "prompt_toolkit.layout.menus": MagicMock(),
    +        "prompt_toolkit.widgets": MagicMock(),
    +        "prompt_toolkit.key_binding": MagicMock(),
    +        "prompt_toolkit.completion": MagicMock(),
    +        "prompt_toolkit.formatted_text": MagicMock(),
    +        "prompt_toolkit.auto_suggest": MagicMock(),
    +    }
    +    with patch.dict(sys.modules, prompt_toolkit_stubs), patch.dict(
    +        "os.environ", clean_env, clear=False
    +    ):
    +        import cli as _cli_mod
    +
    +        _cli_mod = importlib.reload(_cli_mod)
    +        with patch.object(_cli_mod, "get_tool_definitions", return_value=[]), patch.dict(
    +            _cli_mod.__dict__, {"CLI_CONFIG": _clean_config}
    +        ):
    +            return _cli_mod.HermesCLI()
    +
    +
    +class _StubAgent:
    +    """Agent whose turn completes WITHOUT acknowledging the interrupt."""
    +
    +    def __init__(self, session_id, turn_seconds=0.5):
    +        self.session_id = session_id
    +        self.turn_seconds = turn_seconds
    +        self._interrupt_requested = False
    +        self._interrupt_message = None
    +        self._active_children = []
    +        self.interrupt_calls = []
    +        self.clear_calls = 0
    +        self.max_iterations = 90
    +        self.model = "test/model"
    +        self.platform = "cli"
    +
    +    def run_conversation(self, **kwargs):
    +        # Simulate a turn that finishes normally — it never observed the
    +        # interrupt flag (raced past its last check).
    +        time.sleep(self.turn_seconds)
    +        return {
    +            "final_response": "turn finished normally",
    +            "messages": [
    +                {"role": "user", "content": "original"},
    +                {"role": "assistant", "content": "turn finished normally"},
    +            ],
    +            "api_calls": 1,
    +            "completed": True,
    +            # NOTE: no "interrupted" key — the race means finalize_turn
    +            # never saw the flag (or cleared it before it was re-set).
    +            "partial": True,  # skip auto-title thread in the test
    +            # Skip the Rich Panel rendering path (crashes under the
    +            # prompt_toolkit/skin mocks; irrelevant to this regression).
    +            "response_previewed": True,
    +        }
    +
    +    def interrupt(self, message=None):
    +        self.interrupt_calls.append(message)
    +        self._interrupt_requested = True
    +        self._interrupt_message = message
    +
    +    def clear_interrupt(self):
    +        self.clear_calls += 1
    +        self._interrupt_requested = False
    +        self._interrupt_message = None
    +
    +
    +def test_unacknowledged_interrupt_message_is_requeued_not_dropped():
    +    cli = _make_cli()
    +    agent = _StubAgent(cli.session_id)
    +    cli.agent = agent
    +
    +    cli._interrupt_queue = queue.Queue()
    +    cli._pending_input = queue.Queue()
    +    cli._interrupt_queue.put("urgent new message")
    +
    +    with patch.object(cli, "_ensure_runtime_credentials", return_value=True), \
    +         patch.object(cli, "_resolve_turn_agent_config", return_value={
    +             "signature": cli._active_agent_route_signature,
    +             "model": None, "runtime": None, "request_overrides": None,
    +         }), \
    +         patch.object(cli, "_init_agent", return_value=True):
    +        cli.chat("original")
    +
    +    # The interrupt fired against the agent...
    +    assert agent.interrupt_calls == ["urgent new message"]
    +    # ...the turn result never acknowledged it, so the message must be
    +    # re-queued as the next turn instead of dropped.
    +    queued = []
    +    while not cli._pending_input.empty():
    +        queued.append(cli._pending_input.get_nowait())
    +    assert any("urgent new message" in str(q) for q in queued), (
    +        f"interrupt message was dropped; pending_input={queued!r}"
    +    )
    +    # ...and the stale flag must be cleared so the NEXT turn doesn't
    +    # instantly self-abort at its first _interrupt_requested check.
    +    assert agent._interrupt_requested is False
    +    assert agent.clear_calls >= 1
    +
    +
    +def test_acknowledged_interrupt_still_requeues_message():
    +    """The pre-existing path (result carries interrupted=True) still works."""
    +    cli = _make_cli()
    +
    +    class _AckAgent(_StubAgent):
    +        def run_conversation(self, **kwargs):
    +            # Wait until the monitor loop delivers the interrupt.
    +            for _ in range(100):
    +                if self._interrupt_requested:
    +                    break
    +                time.sleep(0.05)
    +            return {
    +                "final_response": "partial work",
    +                "messages": [{"role": "assistant", "content": "partial work"}],
    +                "api_calls": 1,
    +                "completed": False,
    +                "interrupted": True,
    +                "interrupt_message": self._interrupt_message,
    +                "partial": True,
    +            }
    +
    +    agent = _AckAgent(cli.session_id)
    +    cli.agent = agent
    +    cli._interrupt_queue = queue.Queue()
    +    cli._pending_input = queue.Queue()
    +    cli._interrupt_queue.put("redirect please")
    +
    +    with patch.object(cli, "_ensure_runtime_credentials", return_value=True), \
    +         patch.object(cli, "_resolve_turn_agent_config", return_value={
    +             "signature": cli._active_agent_route_signature,
    +             "model": None, "runtime": None, "request_overrides": None,
    +         }), \
    +         patch.object(cli, "_init_agent", return_value=True):
    +        cli.chat("original")
    +
    +    queued = []
    +    while not cli._pending_input.empty():
    +        queued.append(cli._pending_input.get_nowait())
    +    assert any("redirect please" in str(q) for q in queued)
    +    assert cli._last_turn_interrupted is True

  • tests/run_agent/test_run_agent.py

    @@ -2731,7 +2731,7 @@ def shutdown(self, *args, **kwargs):
             mock_msg = _mock_assistant_msg(content="", tool_calls=[tc1, tc2])
             messages = []
     
    -        with patch("agent.tool_executor.concurrent.futures.ThreadPoolExecutor", ShutdownExecutor):
    +        with patch("tools.daemon_pool.DaemonThreadPoolExecutor", ShutdownExecutor):
                 agent._execute_tool_calls_concurrent(mock_msg, messages, "task-1")
     
             assert len(messages) == 2

  • tests/tools/test_daemon_pool.py

    @@ -0,0 +1,89 @@
    +"""Tests for tools.daemon_pool.DaemonThreadPoolExecutor.
    +
    +The daemon pool exists so abandoned workers (interrupted/timed-out tool
    +batches, wedged memory-provider syncs) can never block interpreter exit:
    +stdlib ThreadPoolExecutor workers are non-daemon AND registered in
    +concurrent.futures.thread._threads_queues, whose atexit hook joins every
    +worker unconditionally — even after shutdown(wait=False).
    +"""
    +
    +import subprocess
    +import sys
    +import threading
    +import time
    +
    +from concurrent.futures.thread import _threads_queues
    +
    +from tools.daemon_pool import DaemonThreadPoolExecutor
    +
    +
    +def test_workers_are_daemon_threads():
    +    pool = DaemonThreadPoolExecutor(max_workers=2)
    +    try:
    +        info = pool.submit(
    +            lambda: (threading.current_thread().daemon, threading.current_thread())
    +        ).result(timeout=10)
    +        is_daemon, worker = info
    +        assert is_daemon is True
    +        # Not registered with concurrent.futures' atexit join hook.
    +        assert worker not in _threads_queues
    +    finally:
    +        pool.shutdown(wait=True)
    +
    +
    +def test_results_and_initializer_work_like_stdlib():
    +    seen = []
    +
    +    def _init(tag):
    +        seen.append(tag)
    +
    +    pool = DaemonThreadPoolExecutor(max_workers=1, initializer=_init, initargs=("t",))
    +    try:
    +        assert pool.submit(lambda: 41 + 1).result(timeout=10) == 42
    +        assert seen == ["t"]
    +    finally:
    +        pool.shutdown(wait=True)
    +
    +
    +def test_idle_worker_reuse():
    +    pool = DaemonThreadPoolExecutor(max_workers=4)
    +    try:
    +        tid1 = pool.submit(threading.get_ident).result(timeout=10)
    +        time.sleep(0.05)  # let the worker park on the idle semaphore
    +        tid2 = pool.submit(threading.get_ident).result(timeout=10)
    +        assert tid1 == tid2
    +    finally:
    +        pool.shutdown(wait=True)
    +
    +
    +def test_wedged_worker_does_not_block_interpreter_exit():
    +    """A worker stuck in a long sleep must not hold the process open.
    +
    +    With stdlib ThreadPoolExecutor this subprocess hangs until the sleep
    +    finishes (the atexit hook joins the worker); with the daemon pool it
    +    exits as soon as the main thread returns.
    +    """
    +    script = (
    +        "import sys; sys.path.insert(0, %r)\n"
    +        "from tools.daemon_pool import DaemonThreadPoolExecutor\n"
    +        "import time\n"
    +        "pool = DaemonThreadPoolExecutor(max_workers=1)\n"
    +        "pool.submit(time.sleep, 120)\n"
    +        "time.sleep(0.3)\n"
    +        "pool.shutdown(wait=False)\n"
    +        "print('main-done', flush=True)\n"
    +    ) % (str(_repo_root()),)
    +    proc = subprocess.run(
    +        [sys.executable, "-c", script],
    +        capture_output=True,
    +        text=True,
    +        timeout=30,
    +    )
    +    assert proc.returncode == 0
    +    assert "main-done" in proc.stdout
    +
    +
    +def _repo_root():
    +    import pathlib
    +
    +    return pathlib.Path(__file__).resolve().parents[2]

  • tools/async_delegation.py

    @@ -40,48 +40,18 @@
     import threading
     import time
     import uuid
    -import weakref
     from concurrent.futures import ThreadPoolExecutor
    -from concurrent.futures.thread import _worker
     from typing import Any, Callable, Dict, List, Optional
     
    +from tools.daemon_pool import DaemonThreadPoolExecutor
     from tools.thread_context import propagate_context_to_thread
     
     logger = logging.getLogger(__name__)
     
    -
    -class _DaemonThreadPoolExecutor(ThreadPoolExecutor):
    -    """ThreadPoolExecutor variant whose workers do not block process exit.
    -
    -    Stdlib ``ThreadPoolExecutor`` workers are non-daemon. Background
    -    delegation is explicitly best-effort detached work, so a long child should
    -    be interruptible by ``/stop``/shutdown but must not keep a CLI process alive
    -    after the user exits.
    -    """
    -
    -    def _adjust_thread_count(self) -> None:
    -        if self._idle_semaphore.acquire(timeout=0):
    -            return
    -
    -        def weakref_cb(_, q=self._work_queue):
    -            q.put(None)
    -
    -        num_threads = len(self._threads)
    -        if num_threads < self._max_workers:
    -            thread_name = "%s_%d" % (self._thread_name_prefix or self, num_threads)
    -            t = threading.Thread(
    -                name=thread_name,
    -                target=_worker,
    -                args=(
    -                    weakref.ref(self, weakref_cb),
    -                    self._work_queue,
    -                    self._initializer,
    -                    self._initargs,
    -                ),
    -                daemon=True,
    -            )
    -            t.start()
    -            self._threads.add(t)
    +# Back-compat alias — the daemon executor now lives in tools.daemon_pool so
    +# other subsystems (tool_executor, memory_manager, delegate_tool, skills_hub)
    +# can share it. Existing imports of ``_DaemonThreadPoolExecutor`` keep working.
    +_DaemonThreadPoolExecutor = DaemonThreadPoolExecutor
     
     
     # ---------------------------------------------------------------------------

  • tools/daemon_pool.py

    @@ -0,0 +1,64 @@
    +"""Shared daemon-thread ThreadPoolExecutor.
    +
    +Stdlib ``ThreadPoolExecutor`` workers are non-daemon AND are registered in
    +``concurrent.futures.thread._threads_queues``, whose atexit hook
    +(``_python_exit``) joins every worker unconditionally — even after
    +``shutdown(wait=False)``.  A single wedged worker (tool blocked on network
    +I/O, hung provider daemon, stuck subagent) therefore blocks interpreter
    +exit forever.  This is the root cause of multi-minute CLI exits on long
    +sessions: every abandoned concurrent-tool batch leaves workers that the
    +exit hook insists on joining.
    +
    +``DaemonThreadPoolExecutor`` spawns daemon workers and skips the
    +``_threads_queues`` registration, so:
    +
    +  - ``_python_exit`` never joins them, and
    +  - the interpreter's non-daemon thread join at shutdown skips them.
    +
    +Semantics are otherwise identical (initializer/initargs, work queue,
    +idle-thread reuse).  Use it for any pool whose work is best-effort or
    +independently interruptible and must never hold the process open:
    +concurrent tool execution, background memory sync, catalog fan-out,
    +subagent timeout wrappers.  Do NOT use it for work that must complete
    +before exit (durable writes) — those belong on foreground threads with
    +explicit bounded joins.
    +"""
    +
    +from __future__ import annotations
    +
    +import threading
    +import weakref
    +from concurrent.futures import ThreadPoolExecutor
    +from concurrent.futures.thread import _worker
    +
    +__all__ = ["DaemonThreadPoolExecutor"]
    +
    +
    +class DaemonThreadPoolExecutor(ThreadPoolExecutor):
    +    """ThreadPoolExecutor variant whose workers do not block process exit."""
    +
    +    def _adjust_thread_count(self) -> None:
    +        # Mirrors CPython's implementation (3.8–3.13) with two changes:
    +        # daemon=True and no _threads_queues registration.
    +        if self._idle_semaphore.acquire(timeout=0):
    +            return
    +
    +        def weakref_cb(_, q=self._work_queue):
    +            q.put(None)
    +
    +        num_threads = len(self._threads)
    +        if num_threads < self._max_workers:
    +            thread_name = "%s_%d" % (self._thread_name_prefix or self, num_threads)
    +            t = threading.Thread(
    +                name=thread_name,
    +                target=_worker,
    +                args=(
    +                    weakref.ref(self, weakref_cb),
    +                    self._work_queue,
    +                    self._initializer,
    +                    self._initargs,
    +                ),
    +                daemon=True,
    +            )
    +            t.start()
    +            self._threads.add(t)

  • tools/delegate_tool.py

    @@ -1887,7 +1887,11 @@ def _heartbeat_loop():
             # result(timeout=None) blocks until the child finishes). Stuck-child
             # protection comes from the heartbeat staleness monitor instead.
             child_timeout = _get_child_timeout()
    -        _timeout_executor = ThreadPoolExecutor(
    +        # Daemon worker (tools.daemon_pool): a timed-out child is abandoned
    +        # below; a stdlib non-daemon worker would then block interpreter
    +        # exit at atexit-join time if the child never unwinds.
    +        from tools.daemon_pool import DaemonThreadPoolExecutor
    +        _timeout_executor = DaemonThreadPoolExecutor(
                 max_workers=1,
                 # Install a non-interactive approval callback in the worker thread
                 # so dangerous-command prompts from the subagent don't fall back to
    @@ -2535,7 +2539,11 @@ def _execute_and_aggregate() -> dict:
                 completed_count = 0
                 spinner_ref = getattr(parent_agent, "_delegate_spinner", None)
     
    -            with ThreadPoolExecutor(max_workers=max_children) as executor:
    +            # Daemon workers (tools.daemon_pool): the `with` block still joins
    +            # normally, but if the parent is interrupted while a child is
    +            # wedged, the abandoned worker must not block interpreter exit.
    +            from tools.daemon_pool import DaemonThreadPoolExecutor
    +            with DaemonThreadPoolExecutor(max_workers=max_children) as executor:
                     futures = {}
                     for i, t, child in children:
                         future = executor.submit(

  • tools/skills_hub.py

    @@ -4005,8 +4005,11 @@ def parallel_search_sources(
         # worker finishes — so a single slow source (e.g. ClawHub) keeps the
         # caller blocked for minutes and renders ``overall_timeout`` a no-op.
         # Manage the executor manually and shut it down with ``wait=False`` so
    -    # the timeout is actually honoured.
    -    pool = ThreadPoolExecutor(max_workers=min(len(active), 8))
    +    # the timeout is actually honoured.  Daemon workers (tools.daemon_pool):
    +    # an abandoned slow source must not block interpreter exit either —
    +    # stdlib workers are joined unconditionally by the atexit hook.
    +    from tools.daemon_pool import DaemonThreadPoolExecutor
    +    pool = DaemonThreadPoolExecutor(max_workers=min(len(active), 8))
         futures = {}
         for src in active:
             lim = per_source_limits.get(src.source_id(), 50)