Gateway shutdown, restart, and hang hardening

Fixes on record

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

• #20949 fix(gateway): wait for systemd restart readiness + harden Discord slash-command sync

  merged to main

  files: gateway/platforms/discord.py, hermes_cli/gateway.py, tests/gateway/test_discord_connect.py, tests/hermes_cli/test_gateway_service.py

  Show the diff

  • gateway/platforms/discord.py

    @@ -10,6 +10,8 @@
     """
     
     import asyncio
    +import hashlib
    +import json
     import logging
     import os
     import struct
    @@ -24,6 +26,10 @@
     
     VALID_THREAD_AUTO_ARCHIVE_MINUTES = {60, 1440, 4320, 10080}
     _DISCORD_COMMAND_SYNC_POLICIES = {"safe", "bulk", "off"}
    +_DISCORD_COMMAND_SYNC_STATE_SUBDIR = "gateway"
    +_DISCORD_COMMAND_SYNC_STATE_FILENAME = "discord_command_sync_state.json"
    +_DISCORD_COMMAND_SYNC_MUTATION_INTERVAL_SECONDS = 4.5
    +_DISCORD_COMMAND_SYNC_MAX_RATE_LIMIT_SLEEP_SECONDS = 30.0
     
     try:
         import discord
    @@ -45,6 +51,7 @@
     import re
     
     from gateway.platforms.helpers import MessageDeduplicator, ThreadParticipationTracker
    +from utils import atomic_json_write
     from gateway.platforms.base import (
         BasePlatformAdapter,
         MessageEvent,
    @@ -825,6 +832,167 @@ async def disconnect(self) -> None:
     
             logger.info("[%s] Disconnected", self.name)
     
    +    def _command_sync_state_path(self) -> _Path:
    +        from hermes_constants import get_hermes_home
    +
    +        directory = get_hermes_home() / _DISCORD_COMMAND_SYNC_STATE_SUBDIR
    +        try:
    +            directory.mkdir(parents=True, exist_ok=True)
    +        except Exception:
    +            pass
    +        return directory / _DISCORD_COMMAND_SYNC_STATE_FILENAME
    +
    +    def _read_command_sync_state(self) -> dict:
    +        try:
    +            path = self._command_sync_state_path()
    +            if not path.exists():
    +                return {}
    +            data = json.loads(path.read_text(encoding="utf-8"))
    +        except Exception:
    +            return {}
    +        return data if isinstance(data, dict) else {}
    +
    +    def _write_command_sync_state(self, state: dict) -> None:
    +        atomic_json_write(
    +            self._command_sync_state_path(),
    +            state,
    +            indent=None,
    +            separators=(",", ":"),
    +        )
    +
    +    def _command_sync_state_key(self, app_id: Any) -> str:
    +        return str(app_id or "unknown")
    +
    +    def _desired_command_sync_fingerprint(self) -> str:
    +        tree = self._client.tree if self._client else None
    +        desired = []
    +        if tree is not None:
    +            desired = [
    +                self._canonicalize_app_command_payload(command.to_dict(tree))
    +                for command in tree.get_commands()
    +            ]
    +        desired.sort(key=lambda item: (item.get("type", 1), item.get("name", "")))
    +        payload = json.dumps(desired, sort_keys=True, separators=(",", ":"))
    +        return hashlib.sha256(payload.encode("utf-8")).hexdigest()
    +
    +    def _command_sync_skip_reason(self, app_id: Any, fingerprint: str) -> Optional[str]:
    +        entry = self._read_command_sync_state().get(self._command_sync_state_key(app_id))
    +        if not isinstance(entry, dict):
    +            return None
    +        now = time.time()
    +        retry_after_until = float(entry.get("retry_after_until") or 0)
    +        if retry_after_until > now:
    +            remaining = max(1, int(retry_after_until - now))
    +            return f"Discord asked us to wait before syncing slash commands; retry in {remaining}s"
    +        if entry.get("fingerprint") == fingerprint and entry.get("last_success_at"):
    +            return "same slash-command fingerprint already synced"
    +        return None
    +
    +    def _record_command_sync_attempt(self, app_id: Any, fingerprint: str) -> None:
    +        state = self._read_command_sync_state()
    +        state[self._command_sync_state_key(app_id)] = {
    +            **(
    +                state.get(self._command_sync_state_key(app_id))
    +                if isinstance(state.get(self._command_sync_state_key(app_id)), dict)
    +                else {}
    +            ),
    +            "fingerprint": fingerprint,
    +            "last_attempt_at": time.time(),
    +        }
    +        self._write_command_sync_state(state)
    +
    +    def _record_command_sync_rate_limit(self, app_id: Any, fingerprint: str, retry_after: float) -> None:
    +        retry_after = max(1.0, float(retry_after))
    +        state = self._read_command_sync_state()
    +        state[self._command_sync_state_key(app_id)] = {
    +            **(
    +                state.get(self._command_sync_state_key(app_id))
    +                if isinstance(state.get(self._command_sync_state_key(app_id)), dict)
    +                else {}
    +            ),
    +            "fingerprint": fingerprint,
    +            "last_attempt_at": time.time(),
    +            "retry_after_until": time.time() + retry_after,
    +            "retry_after": retry_after,
    +        }
    +        self._write_command_sync_state(state)
    +
    +    def _record_command_sync_success(self, app_id: Any, fingerprint: str, summary: dict) -> None:
    +        state = self._read_command_sync_state()
    +        state[self._command_sync_state_key(app_id)] = {
    +            "fingerprint": fingerprint,
    +            "last_attempt_at": time.time(),
    +            "last_success_at": time.time(),
    +            "summary": summary,
    +        }
    +        self._write_command_sync_state(state)
    +
    +    @staticmethod
    +    def _extract_discord_retry_after(exc: BaseException) -> Optional[float]:
    +        value = getattr(exc, "retry_after", None)
    +        if value is not None:
    +            try:
    +                return max(1.0, float(value))
    +            except (TypeError, ValueError):
    +                return None
    +        response = getattr(exc, "response", None)
    +        headers = getattr(response, "headers", None)
    +        if headers:
    +            for key in ("Retry-After", "X-RateLimit-Reset-After"):
    +                try:
    +                    raw = headers.get(key)
    +                except Exception:
    +                    raw = None
    +                if raw is None:
    +                    continue
    +                try:
    +                    return max(1.0, float(raw))
    +                except (TypeError, ValueError):
    +                    continue
    +        return None
    +
    +    @staticmethod
    +    def _is_discord_rate_limit(exc: BaseException) -> bool:
    +        """True only for exceptions that look like Discord 429 rate limits.
    +
    +        Narrower than ``hasattr(exc, 'retry_after')``: discord.py's own
    +        ``RateLimited`` exception and any HTTPException with status 429
    +        qualify. This prevents suppressing unrelated failures that happen
    +        to expose a ``retry_after`` attribute."""
    +        # discord.py emits RateLimited / HTTPException subclasses for 429s.
    +        # Guard with isinstance-of-class so a mocked ``discord`` module
    +        # (where attrs are MagicMocks, not types) doesn't trip isinstance.
    +        if DISCORD_AVAILABLE and discord is not None:
    +            for attr_name in ("RateLimited", "HTTPException"):
    +                cls = getattr(discord, attr_name, None)
    +                if not isinstance(cls, type):
    +                    continue
    +                if isinstance(exc, cls):
    +                    if attr_name == "RateLimited":
    +                        return True
    +                    status = getattr(exc, "status", None)
    +                    if status == 429:
    +                        return True
    +        # Fallback duck-type: something named like a rate-limit with a
    +        # numeric retry_after. Covers mocked clients in tests and exotic
    +        # transports, without swallowing arbitrary exceptions.
    +        name = type(exc).__name__.lower()
    +        if ("ratelimit" in name or "rate_limit" in name) and getattr(exc, "retry_after", None) is not None:
    +            return True
    +        response = getattr(exc, "response", None)
    +        status = getattr(response, "status", None) or getattr(response, "status_code", None)
    +        if status == 429:
    +            return True
    +        return False
    +
    +    def _command_sync_mutation_interval_seconds(self) -> float:
    +        return _DISCORD_COMMAND_SYNC_MUTATION_INTERVAL_SECONDS
    +
    +    async def _sleep_between_command_sync_mutations(self) -> None:
    +        interval = self._command_sync_mutation_interval_seconds()
    +        if interval > 0:
    +            await asyncio.sleep(interval)
    +
         async def _run_post_connect_initialization(self) -> None:
             """Finish non-critical startup work after Discord is connected."""
             if not self._client:
    @@ -840,14 +1008,46 @@ async def _run_post_connect_initialization(self) -> None:
                     logger.info("[%s] Synced %d slash command(s) via bulk tree sync", self.name, len(synced))
                     return
     
    -            # Discord's per-app command-management bucket is ~5 writes / 20 s,
    -            # so a mass-prune-plus-upsert reconcile (e.g. 77 orphans + 30
    -            # desired = 107 writes) takes several minutes of forced waits.
    -            # A flat 30 s budget blew up reliably under bucket pressure and
    -            # left slash commands broken for ~60 min until the bucket fully
    -            # recovered. Use a wide ceiling; the cap still guards against a
    -            # true hang. (#16713)
    -            summary = await asyncio.wait_for(self._safe_sync_slash_commands(), timeout=600)
    +            app_id = getattr(self._client, "application_id", None) or getattr(getattr(self._client, "user", None), "id", None)
    +            fingerprint = self._desired_command_sync_fingerprint()
    +            skip_reason = self._command_sync_skip_reason(app_id, fingerprint)
    +            if skip_reason:
    +                logger.info("[%s] Skipping Discord slash command sync: %s", self.name, skip_reason)
    +                return
    +            self._record_command_sync_attempt(app_id, fingerprint)
    +
    +            http = getattr(self._client, "http", None)
    +            has_ratelimit_timeout = http is not None and hasattr(http, "max_ratelimit_timeout")
    +            previous_ratelimit_timeout = getattr(http, "max_ratelimit_timeout", None) if has_ratelimit_timeout else None
    +            if has_ratelimit_timeout:
    +                http.max_ratelimit_timeout = _DISCORD_COMMAND_SYNC_MAX_RATE_LIMIT_SLEEP_SECONDS
    +
    +            try:
    +                # Discord's per-app command-management bucket is small, and
    +                # discord.py can otherwise sit inside one long retry sleep
    +                # before surfacing the 429. Keep the whole sync bounded and
    +                # persist Discord's retry-after when it refuses the batch.
    +                summary = await asyncio.wait_for(self._safe_sync_slash_commands(), timeout=600)
    +            except Exception as e:
    +                if not self._is_discord_rate_limit(e):
    +                    raise
    +                retry_after = self._extract_discord_retry_after(e)
    +                if retry_after is None:
    +                    # Rate-limited but no retry-after signal — back off for a
    +                    # conservative default so we don't slam the bucket again.
    +                    retry_after = _DISCORD_COMMAND_SYNC_MAX_RATE_LIMIT_SLEEP_SECONDS
    +                self._record_command_sync_rate_limit(app_id, fingerprint, retry_after)
    +                logger.warning(
    +                    "[%s] Discord rate-limited slash command sync; retrying after %.0fs",
    +                    self.name,
    +                    retry_after,
    +                )
    +                return
    +            finally:
    +                if has_ratelimit_timeout:
    +                    http.max_ratelimit_timeout = previous_ratelimit_timeout
    +
    +            self._record_command_sync_success(app_id, fingerprint, summary)
                 logger.info(
                     "[%s] Safely reconciled %d slash command(s): unchanged=%d updated=%d recreated=%d created=%d deleted=%d",
                     self.name,
    @@ -1009,11 +1209,20 @@ async def _safe_sync_slash_commands(self) -> Dict[str, int]:
             created = 0
             deleted = 0
             http = self._client.http
    +        mutation_count = 0
    +
    +        async def mutate(call, *args):
    +            nonlocal mutation_count
    +            if mutation_count:
    +                await self._sleep_between_command_sync_mutations()
    +            result = await call(*args)
    +            mutation_count += 1
    +            return result
     
             for key, desired in desired_by_key.items():
                 current = existing_by_key.pop(key, None)
                 if current is None:
    -                await http.upsert_global_command(app_id, desired)
    +                await mutate(http.upsert_global_command, app_id, desired)
                     created += 1
                     continue
     
    @@ -1025,16 +1234,16 @@ async def _safe_sync_slash_commands(self) -> Dict[str, int]:
                     continue
     
                 if self._patchable_app_command_payload(current_existing_payload) == self._patchable_app_command_payload(desired):
    -                await http.delete_global_command(app_id, current.id)
    -                await http.upsert_global_command(app_id, desired)
    +                await mutate(http.delete_global_command, app_id, current.id)
    +                await mutate(http.upsert_global_command, app_id, desired)
                     recreated += 1
                     continue
     
    -            await http.edit_global_command(app_id, current.id, desired)
    +            await mutate(http.edit_global_command, app_id, current.id, desired)
                 updated += 1
     
             for current in existing_by_key.values():
    -            await http.delete_global_command(app_id, current.id)
    +            await mutate(http.delete_global_command, app_id, current.id)
                 deleted += 1
     
             return {

  • hermes_cli/gateway.py

    @@ -505,6 +505,7 @@ def _read_systemd_unit_properties(
             "SubState",
             "Result",
             "ExecMainStatus",
    +        "MainPID",
         ),
     ) -> dict[str, str]:
         """Return selected ``systemctl show`` properties for the gateway unit."""
    @@ -538,6 +539,41 @@ def _read_systemd_unit_properties(
         return parsed
     
     
    +def _systemd_main_pid_from_props(props: dict[str, str]) -> int | None:
    +    try:
    +        pid = int(props.get("MainPID", "0") or "0")
    +    except (TypeError, ValueError):
    +        return None
    +    return pid if pid > 0 else None
    +
    +
    +def _systemd_main_pid(system: bool = False) -> int | None:
    +    return _systemd_main_pid_from_props(_read_systemd_unit_properties(system=system))
    +
    +
    +def _read_gateway_runtime_status() -> dict | None:
    +    try:
    +        from gateway.status import read_runtime_status
    +
    +        state = read_runtime_status()
    +    except Exception:
    +        return None
    +    return state if isinstance(state, dict) else None
    +
    +
    +def _gateway_runtime_status_for_pid(pid: int | None) -> dict | None:
    +    if not pid:
    +        return None
    +    state = _read_gateway_runtime_status()
    +    if not state:
    +        return None
    +    try:
    +        state_pid = int(state.get("pid", 0) or 0)
    +    except (TypeError, ValueError):
    +        return None
    +    return state if state_pid == pid else None
    +
    +
     def _wait_for_systemd_service_restart(
         *,
         system: bool = False,
    @@ -550,6 +586,7 @@ def _wait_for_systemd_service_restart(
         svc = get_service_name()
         scope_label = _service_scope_label(system).capitalize()
         deadline = time.time() + timeout
    +    printed_runtime_wait = False
     
         while time.time() < deadline:
             props = _read_systemd_unit_properties(system=system)
    @@ -562,19 +599,32 @@ def _wait_for_systemd_service_restart(
                 new_pid = get_running_pid()
             except Exception:
                 new_pid = None
    +        if not new_pid:
    +            new_pid = _systemd_main_pid_from_props(props)
     
             if active_state == "active":
                 if new_pid and (previous_pid is None or new_pid != previous_pid):
    -                print(f"✓ {scope_label} service restarted (PID {new_pid})")
    -                return True
    -            if previous_pid is None:
    -                print(f"✓ {scope_label} service restarted")
    -                return True
    +                runtime_state = _gateway_runtime_status_for_pid(new_pid)
    +                gateway_state = (runtime_state or {}).get("gateway_state")
    +                if gateway_state == "running":
    +                    print(f"✓ {scope_label} service restarted (PID {new_pid})")
    +                    return True
    +                if gateway_state == "startup_failed":
    +                    reason = (runtime_state or {}).get("exit_reason") or "startup failed"
    +                    print(f"⚠ {scope_label} service process restarted (PID {new_pid}), but gateway startup failed: {reason}")
    +                    return False
    +                if not printed_runtime_wait:
    +                    print(f"⏳ {scope_label} service process started (PID {new_pid}); waiting for gateway runtime...")
    +                    printed_runtime_wait = True
     
             if active_state == "activating" and sub_state == "auto-restart":
                 time.sleep(1)
                 continue
     
    +        if _systemd_unit_is_start_limited(props):
    +            _print_systemd_start_limit_wait(system=system)
    +            return False
    +
             time.sleep(2)
     
         print(
    @@ -585,6 +635,46 @@ def _wait_for_systemd_service_restart(
         return False
     
     
    +def _systemd_unit_is_start_limited(props: dict[str, str]) -> bool:
    +    result = props.get("Result", "").lower()
    +    sub_state = props.get("SubState", "").lower()
    +    return result == "start-limit-hit" or sub_state == "start-limit-hit"
    +
    +
    +def _systemd_error_indicates_start_limit(exc: subprocess.CalledProcessError) -> bool:
    +    parts: list[str] = []
    +    for attr in ("stderr", "stdout", "output"):
    +        value = getattr(exc, attr, None)
    +        if not value:
    +            continue
    +        if isinstance(value, bytes):
    +            value = value.decode(errors="replace")
    +        parts.append(str(value))
    +    text = "\n".join(parts).lower()
    +    return (
    +        "start-limit-hit" in text
    +        or "start request repeated too quickly" in text
    +        or "start-limit" in text
    +    )
    +
    +
    +def _systemd_service_is_start_limited(system: bool = False) -> bool:
    +    return _systemd_unit_is_start_limited(_read_systemd_unit_properties(system=system))
    +
    +
    +def _print_systemd_start_limit_wait(system: bool = False) -> None:
    +    svc = get_service_name()
    +    scope_label = _service_scope_label(system).capitalize()
    +    scope_flag = " --system" if system else ""
    +    systemctl_prefix = "systemctl " if system else "systemctl --user "
    +    journal_prefix = "journalctl " if system else "journalctl --user "
    +    print(f"⏳ {scope_label} service is temporarily rate-limited by systemd.")
    +    print("  systemd is refusing another immediate start after repeated exits.")
    +    print(f"  Wait for the start-limit window to expire, then run: {'sudo ' if system else ''}hermes gateway restart{scope_flag}")
    +    print(f"  Or clear the failed state manually: {systemctl_prefix}reset-failed {svc}")
    +    print(f"  Check logs: {journal_prefix}-u {svc} -l --since '5 min ago'")
    +
    +
     def _recover_pending_systemd_restart(system: bool = False, previous_pid: int | None = None) -> bool:
         """Recover a planned service restart that is stuck in systemd state."""
         props = _read_systemd_unit_properties(system=system)
    @@ -2135,41 +2225,52 @@ def systemd_restart(system: bool = False):
         refresh_systemd_unit_if_needed(system=system)
         from gateway.status import get_running_pid
     
    -    pid = get_running_pid()
    -    if pid is not None and _request_gateway_self_restart(pid):
    -        import time
    +    pid = get_running_pid() or _systemd_main_pid(system=system)
    +    if pid is not None:
             scope_label = _service_scope_label(system).capitalize()
             svc = get_service_name()
    +        drain_timeout = _get_restart_drain_timeout()
    +
    +        print(f"⏳ {scope_label} service restarting gracefully (PID {pid})...")
    +        if _graceful_restart_via_sigusr1(pid, drain_timeout + 5):
    +            # The gateway exits with code 75 for a planned service restart.
    +            # RestartSec can otherwise delay the relaunch even though the
    +            # operator asked for an immediate restart, so kick the unit once
    +            # the old PID has exited and then wait for the replacement PID.
    +            _run_systemctl(
    +                ["reset-failed", svc],
    +                system=system,
    +                check=False,
    +                timeout=30,
    +            )
    +            _run_systemctl(
    +                ["restart", svc],
    +                system=system,
    +                check=False,
    +                timeout=90,
    +            )
    +            if _wait_for_systemd_service_restart(system=system, previous_pid=pid):
    +                return
    +            if _systemd_service_is_start_limited(system=system):
    +                return
     
    -        # Phase 1: wait for old process to exit (drain + shutdown)
    -        print(f"⏳ {scope_label} service draining active work...")
    -        deadline = time.time() + 90
    -        while time.time() < deadline:
    -            try:
    -                os.kill(pid, 0)
    -                time.sleep(1)
    -            except (ProcessLookupError, PermissionError):
    -                break  # old process is gone
    -        else:
    -            print(f"⚠ Old process (PID {pid}) still alive after 90s")
    -
    -        # The gateway exits with code 75 for a planned service restart.
    -        # systemd can sit in the RestartSec window or even wedge itself into a
    -        # failed/rate-limited state if the operator asks for another restart in
    -        # the middle of that handoff. Clear any stale failed state and kick the
    -        # unit immediately so `hermes gateway restart` behaves idempotently.
    +        print(
    +            f"⚠ Graceful restart did not complete within {int(drain_timeout + 5)}s; "
    +            "forcing a service restart..."
    +        )
             _run_systemctl(
                 ["reset-failed", svc],
                 system=system,
                 check=False,
                 timeout=30,
             )
    -        _run_systemctl(
    -            ["start", svc],
    -            system=system,
    -            check=False,
    -            timeout=90,
    -        )
    +        try:
    +            _run_systemctl(["restart", svc], system=system, check=True, timeout=90)
    +        except subprocess.CalledProcessError as exc:
    +            if _systemd_error_indicates_start_limit(exc) or _systemd_service_is_start_limited(system=system):
    +                _print_systemd_start_limit_wait(system=system)
    +                return
    +            raise
             _wait_for_systemd_service_restart(system=system, previous_pid=pid)
             return
     
    @@ -2182,8 +2283,14 @@ def systemd_restart(system: bool = False):
             check=False,
             timeout=30,
         )
    -    _run_systemctl(["reload-or-restart", get_service_name()], system=system, check=True, timeout=90)
    -    print(f"✓ {_service_scope_label(system).capitalize()} service restarted")
    +    try:
    +        _run_systemctl(["restart", get_service_name()], system=system, check=True, timeout=90)
    +    except subprocess.CalledProcessError as exc:
    +        if _systemd_error_indicates_start_limit(exc) or _systemd_service_is_start_limited(system=system):
    +            _print_systemd_start_limit_wait(system=system)
    +            return
    +        raise
    +    _wait_for_systemd_service_restart(system=system, previous_pid=pid)
     
     
     
    @@ -2255,6 +2362,10 @@ def systemd_status(deep: bool = False, system: bool = False, full: bool = False)
         result_code = unit_props.get("Result", "")
         if active_state == "activating" and sub_state == "auto-restart":
             print("  ⏳ Restart pending: systemd is waiting to relaunch the gateway")
    +    elif _systemd_unit_is_start_limited(unit_props):
    +        print("  ⏳ Restart pending: systemd is temporarily rate-limiting starts")
    +        print(f"  Run after the start-limit window expires: {'sudo ' if system else ''}hermes gateway restart{scope_flag}")
    +        print(f"  Or clear it manually: systemctl {'--user ' if not system else ''}reset-failed {get_service_name()}")
         elif active_state == "failed" and exec_main_status == str(GATEWAY_SERVICE_RESTART_EXIT_CODE):
             print("  ⚠ Planned restart is stuck in systemd failed state (exit 75)")
             print(f"  Run: systemctl {'--user ' if not system else ''}reset-failed {get_service_name()} && {'sudo ' if system else ''}hermes gateway start{scope_flag}")

  • tests/gateway/test_discord_connect.py

    @@ -1,4 +1,5 @@
     import asyncio
    +import json
     import sys
     from types import SimpleNamespace
     from unittest.mock import AsyncMock, MagicMock
    @@ -70,6 +71,15 @@ def _ensure_discord_mock():
     from gateway.platforms.discord import DiscordAdapter  # noqa: E402
     
     
    [redacted:email](autouse=True)
    +def _speed_up_command_sync_mutation_pacing(monkeypatch):
    +    monkeypatch.setattr(
    +        DiscordAdapter,
    +        "_command_sync_mutation_interval_seconds",
    +        lambda self: 0.0,
    +    )
    +
    +
     class FakeTree:
         def __init__(self):
             self.sync = AsyncMock(return_value=[])
    @@ -536,6 +546,183 @@ async def test_post_connect_initialization_skips_sync_when_policy_off(monkeypatc
         fake_tree.sync.assert_not_called()
     
     
    [redacted:email]
    +async def test_post_connect_initialization_skips_same_fingerprint_after_success(tmp_path, monkeypatch):
    +    adapter = DiscordAdapter(PlatformConfig(enabled=True, token="[redacted:secret-assignment]"))
    +    monkeypatch.setattr("hermes_constants.get_hermes_home", lambda: tmp_path)
    +
    +    class _DesiredCommand:
    +        def to_dict(self, tree):
    +            return {
    +                "name": "status",
    +                "description": "Show Hermes status",
    +                "type": 1,
    +                "options": [],
    +            }
    +
    +    fake_tree = SimpleNamespace(
    +        get_commands=lambda: [_DesiredCommand()],
    +        fetch_commands=AsyncMock(return_value=[]),
    +    )
    +    fake_http = SimpleNamespace(
    +        upsert_global_command=AsyncMock(),
    +        edit_global_command=AsyncMock(),
    +        delete_global_command=AsyncMock(),
    +    )
    +    adapter._client = SimpleNamespace(
    +        tree=fake_tree,
    +        http=fake_http,
    +        application_id=999,
    +        user=SimpleNamespace(id=999),
    +    )
    +
    +    await adapter._run_post_connect_initialization()
    +    await adapter._run_post_connect_initialization()
    +
    +    fake_tree.fetch_commands.assert_awaited_once()
    +    fake_http.upsert_global_command.assert_awaited_once()
    +
    +
    [redacted:email]
    +async def test_post_connect_initialization_respects_discord_retry_after(tmp_path, monkeypatch):
    +    adapter = DiscordAdapter(PlatformConfig(enabled=True, token="[redacted:secret-assignment]"))
    +    monkeypatch.setattr("hermes_constants.get_hermes_home", lambda: tmp_path)
    +
    +    class _DesiredCommand:
    +        def to_dict(self, tree):
    +            return {
    +                "name": "status",
    +                "description": "Show Hermes status",
    +                "type": 1,
    +                "options": [],
    +            }
    +
    +    adapter._client = SimpleNamespace(
    +        tree=SimpleNamespace(get_commands=lambda: [_DesiredCommand()]),
    +        application_id=999,
    +        user=SimpleNamespace(id=999),
    +    )
    +    class _DiscordRateLimit(RuntimeError):
    +        retry_after = 123.0
    +
    +    sync = AsyncMock(side_effect=_DiscordRateLimit("discord rate limited"))
    +    monkeypatch.setattr(adapter, "_safe_sync_slash_commands", sync)
    +
    +    await adapter._run_post_connect_initialization()
    +    await adapter._run_post_connect_initialization()
    +
    +    sync.assert_awaited_once()
    +    state_path = (
    +        tmp_path
    +        / discord_platform._DISCORD_COMMAND_SYNC_STATE_SUBDIR
    +        / discord_platform._DISCORD_COMMAND_SYNC_STATE_FILENAME
    +    )
    +    state = json.loads(state_path.read_text())
    +    entry = state["999"]
    +    assert entry["retry_after"] == 123.0
    +    assert entry["retry_after_until"] > entry["last_attempt_at"]
    +
    +
    [redacted:email]
    +async def test_post_connect_initialization_reraises_non_rate_limit_exceptions(tmp_path, monkeypatch):
    +    """Arbitrary failures during sync must surface, not be swallowed as rate-limits."""
    +    adapter = DiscordAdapter(PlatformConfig(enabled=True, token="[redacted:secret-assignment]"))
    +    monkeypatch.setattr("hermes_constants.get_hermes_home", lambda: tmp_path)
    +
    +    class _DesiredCommand:
    +        def to_dict(self, tree):
    +            return {"name": "status", "description": "Show Hermes status", "type": 1, "options": []}
    +
    +    adapter._client = SimpleNamespace(
    +        tree=SimpleNamespace(get_commands=lambda: [_DesiredCommand()]),
    +        application_id=4242,
    +        user=SimpleNamespace(id=4242),
    +    )
    +
    +    # Unrelated failure that happens to expose retry_after. Must NOT be
    +    # caught by the rate-limit handler — it has nothing to do with 429s.
    +    class _UnrelatedError(RuntimeError):
    +        retry_after = 999.0
    +
    +    sync = AsyncMock(side_effect=_UnrelatedError("database is down"))
    +    monkeypatch.setattr(adapter, "_safe_sync_slash_commands", sync)
    +
    +    # The outer _run_post_connect_initialization has a broad except Exception
    +    # that logs defensively — so we assert on state NOT being written.
    +    await adapter._run_post_connect_initialization()
    +
    +    sync.assert_awaited_once()
    +    state_path = (
    +        tmp_path
    +        / discord_platform._DISCORD_COMMAND_SYNC_STATE_SUBDIR
    +        / discord_platform._DISCORD_COMMAND_SYNC_STATE_FILENAME
    +    )
    +    state = json.loads(state_path.read_text()) if state_path.exists() else {}
    +    entry = state.get("4242", {})
    +    # Attempt was recorded before the sync call, but no rate-limit cooldown
    +    # should have been persisted from the unrelated exception.
    +    assert "retry_after_until" not in entry
    +    assert "retry_after" not in entry
    +
    +
    [redacted:email]
    +async def test_safe_sync_slash_commands_paces_mutation_writes(monkeypatch):
    +    adapter = DiscordAdapter(PlatformConfig(enabled=True, token="[redacted:secret-assignment]"))
    +    monkeypatch.setattr(
    +        DiscordAdapter,
    +        "_command_sync_mutation_interval_seconds",
    +        lambda self: 1.25,
    +    )
    +    sleeps = []
    +
    +    async def fake_sleep(delay):
    +        sleeps.append(delay)
    +
    +    monkeypatch.setattr(discord_platform.asyncio, "sleep", fake_sleep)
    +
    +    class _DesiredCommand:
    +        def __init__(self, payload):
    +            self._payload = payload
    +
    +        def to_dict(self, tree):
    +            assert tree is not None
    +            return dict(self._payload)
    +
    +    desired_one = {
    +        "name": "status",
    +        "description": "Show Hermes status",
    +        "type": 1,
    +        "options": [],
    +    }
    +    desired_two = {
    +        "name": "debug",
    +        "description": "Generate a debug report",
    +        "type": 1,
    +        "options": [],
    +    }
    +    fake_tree = SimpleNamespace(
    +        get_commands=lambda: [_DesiredCommand(desired_one), _DesiredCommand(desired_two)],
    +        fetch_commands=AsyncMock(return_value=[]),
    +    )
    +    fake_http = SimpleNamespace(
    +        upsert_global_command=AsyncMock(),
    +        edit_global_command=AsyncMock(),
    +        delete_global_command=AsyncMock(),
    +    )
    +    adapter._client = SimpleNamespace(
    +        tree=fake_tree,
    +        http=fake_http,
    +        application_id=999,
    +        user=SimpleNamespace(id=999),
    +    )
    +
    +    summary = await adapter._safe_sync_slash_commands()
    +
    +    assert summary["created"] == 2
    +    assert fake_http.upsert_global_command.await_count == 2
    +    assert sleeps == [1.25]
    +
    +
     @pytest.mark.asyncio
     async def test_safe_sync_reads_permission_attrs_from_existing_command():
         """Regression: AppCommand.to_dict() in discord.py does NOT include

  • tests/hermes_cli/test_gateway_service.py

    @@ -2,6 +2,7 @@
     
     import os
     import pwd
    +import subprocess
     from pathlib import Path
     from types import SimpleNamespace
     
    @@ -90,6 +91,13 @@ def test_systemd_restart_refreshes_outdated_unit(self, tmp_path, monkeypatch):
             monkeypatch.setattr(gateway_cli, "generate_systemd_unit", lambda system=False, run_as_user=None: "new unit\n")
     
             calls = []
    +        monkeypatch.setattr("gateway.status.get_running_pid", lambda: None)
    +        monkeypatch.setattr(gateway_cli, "_recover_pending_systemd_restart", lambda system=False, previous_pid=None: False)
    +        monkeypatch.setattr(
    +            gateway_cli,
    +            "_wait_for_systemd_service_restart",
    +            lambda system=False, previous_pid=None: calls.append(("wait", system, previous_pid)) or True,
    +        )
     
             def fake_run(cmd, check=True, **kwargs):
                 calls.append(cmd)
    @@ -100,11 +108,12 @@ def fake_run(cmd, check=True, **kwargs):
             gateway_cli.systemd_restart()
     
             assert unit_path.read_text(encoding="utf-8") == "new unit\n"
    -        assert calls[:4] == [
    +        assert calls[:5] == [
                 ["systemctl", "--user", "daemon-reload"],
    -            ["systemctl", "--user", "show", gateway_cli.get_service_name(), "--no-pager", "--property", "ActiveState,SubState,Result,ExecMainStatus"],
    +            ["systemctl", "--user", "show", gateway_cli.get_service_name(), "--no-pager", "--property", "ActiveState,SubState,Result,ExecMainStatus,MainPID"],
                 ["systemctl", "--user", "reset-failed", gateway_cli.get_service_name()],
    -            ["systemctl", "--user", "reload-or-restart", gateway_cli.get_service_name()],
    +            ["systemctl", "--user", "restart", gateway_cli.get_service_name()],
    +            ("wait", False, None),
             ]
     
         def test_systemd_stop_marks_running_gateway_as_planned_stop(self, monkeypatch):
    @@ -611,62 +620,141 @@ def fake_run(*args, **kwargs):
             assert gateway_cli._is_service_running() is False
     
     class TestGatewaySystemServiceRouting:
    -    def test_systemd_restart_self_requests_graceful_restart_and_waits(self, monkeypatch, capsys):
    +    def test_systemd_restart_gracefully_restarts_running_service_and_waits(self, monkeypatch, capsys):
             calls = []
     
             monkeypatch.setattr(gateway_cli, "_select_systemd_scope", lambda system=False: False)
             monkeypatch.setattr(gateway_cli, "_require_service_installed", lambda action, system=False: None)
             monkeypatch.setattr(gateway_cli, "refresh_systemd_unit_if_needed", lambda system=False: calls.append(("refresh", system)))
    +        monkeypatch.setattr(gateway_cli, "_get_restart_drain_timeout", lambda: 12.0)
             monkeypatch.setattr(
                 "gateway.status.get_running_pid",
                 lambda: 654,
             )
             monkeypatch.setattr(
                 gateway_cli,
    -            "_request_gateway_self_restart",
    -            lambda pid: calls.append(("self", pid)) or True,
    +            "_graceful_restart_via_sigusr1",
    +            lambda pid, timeout: calls.append(("graceful", pid, timeout)) or True,
             )
     
    -        # Simulate: old process dies immediately, new process becomes active
    -        kill_call_count = [0]
    -        def fake_kill(pid, sig):
    -            kill_call_count[0] += 1
    -            if kill_call_count[0] >= 2:  # first call checks, second = dead
    -                raise ProcessLookupError()
    -        monkeypatch.setattr(os, "kill", fake_kill)
    -
    -        # Simulate systemctl reset-failed/start followed by an active unit
    -        new_pid = [None]
    +        # Simulate systemctl reset-failed/restart followed by an active unit.
    +        # A plain start does not break systemd's auto-restart timer once the
    +        # old gateway has exited with the planned restart code.
             def fake_subprocess_run(cmd, **kwargs):
                 if "reset-failed" in cmd:
                     calls.append(("reset-failed", cmd))
                     return SimpleNamespace(stdout="", returncode=0)
    -            if "start" in cmd:
    -                calls.append(("start", cmd))
    +            if "restart" in cmd:
    +                calls.append(("restart", cmd))
                     return SimpleNamespace(stdout="", returncode=0)
    -            if "show" in cmd:
    -                new_pid[0] = 999
    -                return SimpleNamespace(
    -                    stdout="ActiveState=active\nSubState=running\nResult=success\nExecMainStatus=0\n",
    -                    returncode=0,
    -                )
                 raise AssertionError(f"Unexpected systemctl call: {cmd}")
     
             monkeypatch.setattr(gateway_cli.subprocess, "run", fake_subprocess_run)
    -        # get_running_pid returns new PID after restart
    -        pid_calls = [0]
    -        def fake_get_pid():
    -            pid_calls[0] += 1
    -            return 999 if pid_calls[0] > 1 else 654
    -        monkeypatch.setattr("gateway.status.get_running_pid", fake_get_pid)
    +        monkeypatch.setattr(
    +            gateway_cli,
    +            "_wait_for_systemd_service_restart",
    +            lambda system=False, previous_pid=None: calls.append(("wait", system, previous_pid)) or True,
    +        )
     
             gateway_cli.systemd_restart()
     
    -        assert ("self", 654) in calls
    +        assert ("graceful", 654, 17.0) in calls
             assert any(call[0] == "reset-failed" for call in calls)
    -        assert any(call[0] == "start" for call in calls)
    +        assert any(call[0] == "restart" for call in calls)
    +        assert ("wait", False, 654) in calls
             out = capsys.readouterr().out.lower()
    -        assert "restarted" in out
    +        assert "restarting gracefully" in out
    +
    +    def test_systemd_restart_uses_systemd_main_pid_when_pid_file_is_missing(self, monkeypatch, capsys):
    +        calls = []
    +
    +        monkeypatch.setattr(gateway_cli, "_select_systemd_scope", lambda system=False: False)
    +        monkeypatch.setattr(gateway_cli, "_require_service_installed", lambda action, system=False: None)
    +        monkeypatch.setattr(gateway_cli, "refresh_systemd_unit_if_needed", lambda system=False: None)
    +        monkeypatch.setattr(gateway_cli, "_get_restart_drain_timeout", lambda: 10.0)
    +        monkeypatch.setattr("gateway.status.get_running_pid", lambda: None)
    +        monkeypatch.setattr(
    +            gateway_cli,
    +            "_read_systemd_unit_properties",
    +            lambda system=False: {
    +                "ActiveState": "active",
    +                "SubState": "running",
    +                "Result": "success",
    +                "ExecMainStatus": "0",
    +                "MainPID": "777",
    +            },
    +        )
    +        monkeypatch.setattr(
    +            gateway_cli,
    +            "_graceful_restart_via_sigusr1",
    +            lambda pid, timeout: calls.append(("graceful", pid, timeout)) or True,
    +        )
    +        monkeypatch.setattr(gateway_cli, "_run_systemctl", lambda args, **kwargs: calls.append(args) or SimpleNamespace(stdout="", returncode=0))
    +        monkeypatch.setattr(
    +            gateway_cli,
    +            "_wait_for_systemd_service_restart",
    +            lambda system=False, previous_pid=None: calls.append(("wait", system, previous_pid)) or True,
    +        )
    +
    +        gateway_cli.systemd_restart()
    +
    +        assert ("graceful", 777, 15.0) in calls
    +        assert ("wait", False, 777) in calls
    +        assert "restarting gracefully (pid 777)" in capsys.readouterr().out.lower()
    +
    +    def test_wait_for_systemd_restart_waits_for_runtime_running(self, monkeypatch, capsys):
    +        monkeypatch.setattr(
    +            gateway_cli,
    +            "_read_systemd_unit_properties",
    +            lambda system=False: {
    +                "ActiveState": "active",
    +                "SubState": "running",
    +                "Result": "success",
    +                "ExecMainStatus": "0",
    +                "MainPID": "999",
    +            },
    +        )
    +        monkeypatch.setattr("gateway.status.get_running_pid", lambda: None)
    +        monkeypatch.setattr(
    +            gateway_cli,
    +            "_gateway_runtime_status_for_pid",
    +            lambda pid: {"pid": pid, "gateway_state": "running"},
    +        )
    +
    +        assert gateway_cli._wait_for_systemd_service_restart(previous_pid=777, timeout=0.1) is True
    +        assert "restarted (pid 999)" in capsys.readouterr().out.lower()
    +
    +    def test_systemd_restart_reports_start_limit_hit(self, monkeypatch, capsys):
    +        calls = []
    +
    +        monkeypatch.setattr(gateway_cli, "_select_systemd_scope", lambda system=False: False)
    +        monkeypatch.setattr(gateway_cli, "_require_service_installed", lambda action, system=False: None)
    +        monkeypatch.setattr(gateway_cli, "refresh_systemd_unit_if_needed", lambda system=False: None)
    +        monkeypatch.setattr("gateway.status.get_running_pid", lambda: None)
    +        monkeypatch.setattr(gateway_cli, "_recover_pending_systemd_restart", lambda system=False, previous_pid=None: False)
    +
    +        def fake_run_systemctl(args, **kwargs):
    +            calls.append(args)
    +            if args[0] == "show":
    +                return SimpleNamespace(stdout="ActiveState=inactive\nSubState=dead\nResult=success\nExecMainStatus=0\nMainPID=0\n", stderr="", returncode=0)
    +            if args[0] == "reset-failed":
    +                return SimpleNamespace(stdout="", stderr="", returncode=0)
    +            if args[0] == "restart":
    +                raise subprocess.CalledProcessError(
    +                    1,
    +                    ["systemctl", "--user", *args],
    +                    stderr="Job failed. See result 'start-limit-hit'.",
    +                )
    +            raise AssertionError(f"Unexpected args: {args}")
    +
    +        monkeypatch.setattr(gateway_cli, "_run_systemctl", fake_run_systemctl)
    +
    +        gateway_cli.systemd_restart()
    +
    +        assert ["restart", gateway_cli.get_service_name()] in calls
    +        out = capsys.readouterr().out.lower()
    +        assert "rate-limited by systemd" in out
    +        assert "reset-failed" in out
     
         def test_systemd_restart_recovers_failed_planned_restart(self, monkeypatch, capsys):
             monkeypatch.setattr(gateway_cli, "_select_systemd_scope", lambda system=False: False)
    @@ -711,6 +799,11 @@ def fake_subprocess_run(cmd, **kwargs):
                 "gateway.status.get_running_pid",
                 lambda: 999 if started["value"] else None,
             )
    +        monkeypatch.setattr(
    +            gateway_cli,
    +            "_gateway_runtime_status_for_pid",
    +            lambda pid: {"pid": pid, "gateway_state": "running"},
    +        )
     
             gateway_cli.systemd_restart()
     

• #52118 fix(ddgs): bound DuckDuckGo search with a wall-clock timeout (#36776)

  merged to main

  files: plugins/web/ddgs/provider.py, tests/tools/test_web_providers_ddgs.py

• #66955 fix(gateway): thread shutdown watchdog + loop heartbeat for frozen SIGTERM drain (#66892)

  merged to main

  files: gateway/run.py, gateway/shutdown_watchdog.py, tests/gateway/test_shutdown_watchdog.py

  Show the diff

  • gateway/run.py

    @@ -1917,6 +1917,12 @@ def _platform_has_bot_credential(platform: "Platform", platform_config: "Platfor
         merge_pending_message_event,
         utf16_len,
     )
    +from gateway.shutdown_watchdog import (
    +    DEFAULT_HEARTBEAT_INTERVAL_S,
    +    arm_shutdown_watchdog,
    +    loop_heartbeat_forever,
    +    resolve_shutdown_watchdog_delay,
    +)
     from gateway.restart import (
         DEFAULT_GATEWAY_RESTART_DRAIN_TIMEOUT,
         GATEWAY_FATAL_CONFIG_EXIT_CODE,
    @@ -2972,6 +2978,12 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
         _session_model_overrides: Dict[str, Dict[str, str]] = {}
         _session_reasoning_overrides: Dict[str, Dict[str, Any]] = {}
         _startup_restore_in_progress: bool = False
    +    # Loop-liveness heartbeat / shutdown-watchdog handles (#66892). Class-level
    +    # defaults so partial construction in tests doesn't blow up on access; the
    +    # real values are set in __init__ / start() / stop().
    +    _loop_heartbeat_task: Optional["asyncio.Task"] = None
    +    _gateway_started_at: float = 0.0
    +    _shutdown_watchdog_done: Optional["threading.Event"] = None
     
         def __init__(self, config: Optional[GatewayConfig] = None):
             global _gateway_runner_ref
    @@ -3297,6 +3309,12 @@ def __init__(self, config: Optional[GatewayConfig] = None):
             # Track background tasks to prevent garbage collection mid-execution
             self._background_tasks: set = set()
     
    +        # Event-loop liveness heartbeat (#66892): rewritten every 30s while
    +        # the loop is dispatching. External supervisors use the file mtime /
    +        # updated_at to distinguish "process alive" from "loop frozen".
    +        self._gateway_started_at: float = time.time()
    +        self._loop_heartbeat_task: Optional[asyncio.Task] = None
    +
             # scale-to-zero (Phase 0, F13): gateway-scoped "last inbound seen" clock.
             # There is no such clock today (only a per-agent _last_activity_ts), so the
             # idle predicate needs this. Stamped in _handle_message (the single inbound
    @@ -7535,7 +7553,27 @@ async def start(self) -> bool:
     
             self._running = True
             self._update_runtime_status("running")
    -        
    +
    +        # Loop-liveness heartbeat (#66892): an asyncio task so a frozen loop
    +        # stops refreshing ``state/gateway.heartbeat``. Cancelled with the
    +        # other background tasks during stop(). Best-effort — a liveness probe
    +        # must never be able to abort startup.
    +        try:
    +            _existing_hb = getattr(self, "_loop_heartbeat_task", None)
    +            if _existing_hb is None or _existing_hb.done():
    +                self._loop_heartbeat_task = asyncio.create_task(
    +                    loop_heartbeat_forever(
    +                        interval_s=DEFAULT_HEARTBEAT_INTERVAL_S,
    +                        start_time=getattr(self, "_gateway_started_at", 0.0),
    +                    )
    +                )
    +                _bg = getattr(self, "_background_tasks", None)
    +                if _bg is not None:
    +                    _bg.add(self._loop_heartbeat_task)
    +                    self._loop_heartbeat_task.add_done_callback(_bg.discard)
    +        except Exception:
    +            logger.debug("Failed to start gateway loop heartbeat", exc_info=True)
    +
             # Emit gateway:startup hook
             hook_count = len(self.hooks.loaded_hooks)
             if hook_count:
    @@ -8499,11 +8537,57 @@ def _kill_tool_subprocesses(phase: str) -> None:
                     except Exception as _e:
                         logger.debug("cleanup_all_browsers (%s) error: %s", phase, _e)
     
    +            # Thread-based shutdown watchdog (#66892): asyncio timeouts cannot
    +            # recover a frozen loop. Arm a plain OS thread at the start of
    +            # stop(); if teardown never finishes within drain+grace it dumps
    +            # faulthandler stacks and os._exit so KeepAlive/systemd can revive.
    +            # Skip under pytest so stop()-driving unit tests don't get a
    +            # delayed hard-exit in the worker.
    +            _watchdog_done = threading.Event()
    +            self._shutdown_watchdog_done = _watchdog_done
    +            _stop_started_at_box: dict[str, float] = {}
    +
    +            def _shutdown_watchdog_snapshot() -> dict:
    +                started = _stop_started_at_box.get("t")
    +                return {
    +                    "restart_requested": bool(self._restart_requested),
    +                    "draining": bool(self._draining),
    +                    "running": bool(self._running),
    +                    "active_agents": self._running_agent_count(),
    +                    "active_cron_jobs": self._active_cron_job_count(),
    +                    "active_api_runs": self._active_api_run_count(),
    +                    "restart_drain_timeout": self._restart_drain_timeout,
    +                    "watchdog_delay_s": resolve_shutdown_watchdog_delay(
    +                        self._restart_drain_timeout
    +                    ),
    +                    "phase_elapsed_s": (
    +                        time.monotonic() - started if started is not None else None
    +                    ),
    +                }
    +
    +            if not os.environ.get("PYTEST_CURRENT_TEST"):
    +                arm_shutdown_watchdog(
    +                    resolve_shutdown_watchdog_delay(self._restart_drain_timeout),
    +                    done_event=_watchdog_done,
    +                    snapshot_fn=_shutdown_watchdog_snapshot,
    +                    exit_code=1,
    +                )
    +
    +            try:
    +                await _stop_impl_body(
    +                    _kill_tool_subprocesses,
    +                    _stop_started_at_box,
    +                )
    +            finally:
    +                _watchdog_done.set()
    +
    +        async def _stop_impl_body(_kill_tool_subprocesses, _stop_started_at_box) -> None:
                 logger.info(
                     "Stopping gateway%s...",
                     " for restart" if self._restart_requested else "",
                 )
                 _stop_started_at = time.monotonic()
    +            _stop_started_at_box["t"] = _stop_started_at
     
                 def _phase_elapsed() -> float:
                     return time.monotonic() - _stop_started_at

  • gateway/shutdown_watchdog.py

    @@ -0,0 +1,272 @@
    +"""Out-of-loop shutdown backstop + event-loop liveness heartbeat (#66892).
    +
    +When the asyncio loop freezes mid-drain, every asyncio-based recovery path is
    +structurally unable to fire: the drain deadline, status rewrites, and forensics
    +all need the same loop that is stuck. launchd/systemd KeepAlive only restarts a
    +*dead* process, so a wedged-but-alive gateway sits as a zombie until manual
    +SIGKILL.
    +
    +This module provides:
    +
    +1. A plain OS-thread shutdown watchdog armed at ``stop()``. If shutdown has not
    +   completed within ``restart_drain_timeout + grace``, it dumps all-thread
    +   stacks via ``faulthandler`` plus a metadata snapshot, then ``os._exit`` so
    +   the service manager can revive the process.
    +2. An event-loop heartbeat file at ``<HERMES_HOME>/state/gateway.heartbeat`` so
    +   external supervision can distinguish "process alive" from "loop frozen"
    +   (``gateway_state.json`` alone can't — it only rewrites on transitions/turns).
    +"""
    +
    +from __future__ import annotations
    +
    +import asyncio
    +import faulthandler
    +import json
    +import logging
    +import os
    +import sys
    +import threading
    +import time
    +from datetime import datetime, timezone
    +from pathlib import Path
    +from typing import Any, Callable, Dict, Optional
    +
    +from hermes_constants import get_hermes_home
    +from utils import atomic_json_write
    +
    +logger = logging.getLogger(__name__)
    +
    +# Extra leash beyond ``agent.restart_drain_timeout`` so a slow-but-progressing
    +# drain is not cut short. Matches the issue #66892 suggested hardening.
    +DEFAULT_SHUTDOWN_WATCHDOG_GRACE_S = 60.0
    +DEFAULT_HEARTBEAT_INTERVAL_S = 30.0
    +_HEARTBEAT_RELATIVE = ("state", "gateway.heartbeat")
    +_WATCHDOG_DUMP_RELATIVE = ("logs", "gateway-shutdown-watchdog.log")
    +
    +
    +def _process_hermes_home() -> Path:
    +    """HERMES_HOME for process-level identity files (ignore profile overrides)."""
    +    val = os.environ.get("HERMES_HOME", "").strip()
    +    if val:
    +        return Path(val)
    +    return get_hermes_home()
    +
    +
    +def get_loop_heartbeat_path(home: Optional[Path] = None) -> Path:
    +    """Return ``<HERMES_HOME>/state/gateway.heartbeat``."""
    +    base = home if home is not None else _process_hermes_home()
    +    return base.joinpath(*_HEARTBEAT_RELATIVE)
    +
    +
    +def get_shutdown_watchdog_dump_path(home: Optional[Path] = None) -> Path:
    +    """Return the faulthandler / metadata dump path for a fired watchdog."""
    +    base = home if home is not None else _process_hermes_home()
    +    return base.joinpath(*_WATCHDOG_DUMP_RELATIVE)
    +
    +
    +def write_loop_heartbeat(
    +    *,
    +    pid: Optional[int] = None,
    +    start_time: Optional[float] = None,
    +    home: Optional[Path] = None,
    +    extra: Optional[Dict[str, Any]] = None,
    +) -> Path:
    +    """Atomically rewrite the loop-liveness heartbeat file.
    +
    +    ``start_time`` is the gateway process start (``time.time()`` epoch seconds)
    +    so supervisors can detect PID reuse. Best-effort — never raises.
    +    """
    +    path = get_loop_heartbeat_path(home)
    +    payload: Dict[str, Any] = {
    +        "pid": int(pid if pid is not None else os.getpid()),
    +        "updated_at": datetime.now(timezone.utc).isoformat(),
    +        "monotonic": time.monotonic(),
    +    }
    +    if start_time is not None:
    +        payload["start_time"] = float(start_time)
    +    if extra:
    +        payload.update(extra)
    +    try:
    +        atomic_json_write(path, payload, indent=None)
    +    except Exception:
    +        logger.debug("Failed to write gateway loop heartbeat", exc_info=True)
    +    return path
    +
    +
    +def resolve_shutdown_watchdog_delay(
    +    drain_timeout: float,
    +    *,
    +    grace_s: float = DEFAULT_SHUTDOWN_WATCHDOG_GRACE_S,
    +) -> float:
    +    """Return the wall-clock leash for the shutdown watchdog thread."""
    +    try:
    +        drain = max(float(drain_timeout), 0.0)
    +    except (TypeError, ValueError):
    +        drain = 0.0
    +    try:
    +        grace = max(float(grace_s), 0.0)
    +    except (TypeError, ValueError):
    +        grace = DEFAULT_SHUTDOWN_WATCHDOG_GRACE_S
    +    return drain + grace
    +
    +
    +def _write_watchdog_dump(
    +    dump_path: Path,
    +    *,
    +    delay_s: float,
    +    snapshot: Optional[Dict[str, Any]],
    +) -> None:
    +    """Best-effort faulthandler + metadata dump before hard-exit."""
    +    try:
    +        dump_path.parent.mkdir(parents=True, exist_ok=True)
    +    except OSError:
    +        return
    +
    +    header = {
    +        "event": "shutdown_watchdog_fired",
    +        "pid": os.getpid(),
    +        "delay_s": delay_s,
    +        "fired_at": datetime.now(timezone.utc).isoformat(),
    +        "snapshot": snapshot or {},
    +    }
    +    try:
    +        with open(dump_path, "a", encoding="utf-8") as fh:
    +            fh.write(json.dumps(header, default=str) + "\n")
    +            fh.write("--- faulthandler dump (all threads) ---\n")
    +            fh.flush()
    +            try:
    +                faulthandler.dump_traceback(file=fh, all_threads=True)
    +            except Exception:
    +                fh.write("(faulthandler.dump_traceback failed)\n")
    +            fh.write("--- end dump ---\n")
    +            fh.flush()
    +    except Exception:
    +        pass
    +
    +    # Also dump to stderr so journald/launchd capture it even if the file
    +    # write failed (wedged disk was one of the #66892 hypotheses).
    +    try:
    +        sys.stderr.write(
    +            f"Gateway shutdown watchdog fired after {delay_s:.0f}s "
    +            f"(pid={os.getpid()}); dumping all thread stacks.\n"
    +        )
    +        sys.stderr.flush()
    +        faulthandler.dump_traceback(all_threads=True)
    +    except Exception:
    +        pass
    +
    +
    +def arm_shutdown_watchdog(
    +    delay_s: float,
    +    *,
    +    done_event: Optional[threading.Event] = None,
    +    snapshot_fn: Optional[Callable[[], Dict[str, Any]]] = None,
    +    exit_code: int = 1,
    +    dump_path: Optional[Path] = None,
    +    name: str = "gateway-shutdown-watchdog",
    +) -> threading.Event:
    +    """Arm a daemon-thread hard-exit backstop for a wedged shutdown path.
    +
    +    If ``done_event`` is set before ``delay_s`` elapses, the thread exits
    +    quietly (normal / progressing shutdown completed). Otherwise it dumps
    +    diagnostics and calls ``os._exit(exit_code)``.
    +
    +    Never raises. Returns the ``done_event`` (creating one when omitted) so
    +    the caller can disarm on successful completion.
    +    """
    +    done = done_event if done_event is not None else threading.Event()
    +    try:
    +        delay = max(float(delay_s), 0.0)
    +    except (TypeError, ValueError):
    +        delay = DEFAULT_SHUTDOWN_WATCHDOG_GRACE_S
    +
    +    if delay <= 0:
    +        return done
    +
    +    def _watchdog() -> None:
    +        # Wait with interruptible chunks so a late disarm doesn't need the
    +        # full remaining sleep to observe done_event.
    +        deadline = time.monotonic() + delay
    +        while time.monotonic() < deadline:
    +            remaining = deadline - time.monotonic()
    +            if done.wait(timeout=min(remaining, 1.0)):
    +                return
    +        if done.is_set():
    +            return
    +
    +        snapshot: Optional[Dict[str, Any]] = None
    +        if snapshot_fn is not None:
    +            try:
    +                snapshot = snapshot_fn()
    +            except Exception as exc:
    +                snapshot = {"snapshot_error": repr(exc)}
    +
    +        target = dump_path if dump_path is not None else get_shutdown_watchdog_dump_path()
    +        _write_watchdog_dump(target, delay_s=delay, snapshot=snapshot)
    +
    +        try:
    +            logger.critical(
    +                "Shutdown watchdog fired after %.0fs — forcing process exit "
    +                "(asyncio drain path appears wedged; see %s)",
    +                delay,
    +                target,
    +            )
    +        except Exception:
    +            pass
    +
    +        for stream in (sys.stdout, sys.stderr):
    +            try:
    +                stream.flush()
    +            except Exception:
    +                pass
    +        # Mirror _exit_after_graceful_shutdown: release PID file + runtime
    +        # lock BEFORE the log drain (locks must never be stranded), then
    +        # drain the async log queue so the logger.critical above actually
    +        # reaches the file before os._exit bypasses atexit. (#66892)
    +        try:
    +            from gateway.status import remove_pid_file, release_gateway_runtime_lock
    +            remove_pid_file()
    +            release_gateway_runtime_lock()
    +        except Exception:
    +            pass
    +        try:
    +            from hermes_logging import drain_log_queue
    +            drain_log_queue(timeout=1.0)
    +        except Exception:
    +            pass
    +        os._exit(exit_code)
    +
    +    try:
    +        threading.Thread(target=_watchdog, daemon=True, name=name).start()
    +    except Exception:
    +        logger.debug("Failed to arm shutdown watchdog", exc_info=True)
    +    return done
    +
    +
    +async def loop_heartbeat_forever(
    +    *,
    +    interval_s: float = DEFAULT_HEARTBEAT_INTERVAL_S,
    +    start_time: Optional[float] = None,
    +    home: Optional[Path] = None,
    +    should_continue: Optional[Callable[[], bool]] = None,
    +) -> None:
    +    """Rewrite the loop heartbeat file on a cadence until cancelled / gated off.
    +
    +    Runs as an asyncio task on the gateway loop — if the loop freezes, this
    +    task stops and the file mtime/updated_at goes stale for external monitors.
    +    """
    +    try:
    +        interval = max(float(interval_s), 1.0)
    +    except (TypeError, ValueError):
    +        interval = DEFAULT_HEARTBEAT_INTERVAL_S
    +
    +    # Immediate first write so monitors see a fresh file as soon as the
    +    # gateway is running, not after the first interval.
    +    write_loop_heartbeat(start_time=start_time, home=home)
    +    while True:
    +        if should_continue is not None and not should_continue():
    +            return
    +        await asyncio.sleep(interval)
    +        if should_continue is not None and not should_continue():
    +            return
    +        write_loop_heartbeat(start_time=start_time, home=home)

  • tests/gateway/test_shutdown_watchdog.py

    @@ -0,0 +1,150 @@
    +"""Shutdown watchdog + loop heartbeat coverage for #66892.
    +
    +The drain path is asyncio-based; a frozen loop makes every asyncio timeout
    +structurally unable to fire. These tests pin the out-of-loop backstop
    +(thread watchdog) and the loop-liveness heartbeat file contract.
    +"""
    +
    +from __future__ import annotations
    +
    +import asyncio
    +import json
    +import threading
    +import time
    +from unittest.mock import patch
    +
    +import pytest
    +
    +from gateway.shutdown_watchdog import (
    +    DEFAULT_SHUTDOWN_WATCHDOG_GRACE_S,
    +    arm_shutdown_watchdog,
    +    get_loop_heartbeat_path,
    +    get_shutdown_watchdog_dump_path,
    +    loop_heartbeat_forever,
    +    resolve_shutdown_watchdog_delay,
    +    write_loop_heartbeat,
    +)
    +
    +def test_resolve_shutdown_watchdog_delay_adds_grace():
    +    assert resolve_shutdown_watchdog_delay(180) == 180 + DEFAULT_SHUTDOWN_WATCHDOG_GRACE_S
    +    assert resolve_shutdown_watchdog_delay(0) == DEFAULT_SHUTDOWN_WATCHDOG_GRACE_S
    +    assert resolve_shutdown_watchdog_delay("bad") == DEFAULT_SHUTDOWN_WATCHDOG_GRACE_S
    +    assert resolve_shutdown_watchdog_delay(10, grace_s=5) == 15.0
    +
    +
    +def test_write_loop_heartbeat_atomic_json(tmp_path, monkeypatch):
    +    monkeypatch.setenv("HERMES_HOME", str(tmp_path))
    +    path = write_loop_heartbeat(pid=4242, start_time=100.5, home=tmp_path)
    +    assert path == tmp_path / "state" / "gateway.heartbeat"
    +    assert path.is_file()
    +    data = json.loads(path.read_text(encoding="utf-8"))
    +    assert data["pid"] == 4242
    +    assert data["start_time"] == 100.5
    +    assert "updated_at" in data
    +    assert "monotonic" in data
    +    assert get_loop_heartbeat_path(tmp_path) == path
    +
    +
    +def test_arm_shutdown_watchdog_disarm_before_fire(tmp_path):
    +    done = threading.Event()
    +    exited = []
    +
    +    def fake_exit(code):
    +        exited.append(code)
    +        raise _ExitCalled(code)
    +
    +    with patch("gateway.shutdown_watchdog.os._exit", side_effect=fake_exit):
    +        arm_shutdown_watchdog(
    +            0.4,
    +            done_event=done,
    +            dump_path=tmp_path / "dump.log",
    +            exit_code=7,
    +        )
    +        time.sleep(0.1)
    +        done.set()
    +        time.sleep(0.5)
    +
    +    assert exited == []
    +
    +
    +def test_arm_shutdown_watchdog_fires_with_dump_and_exit(tmp_path):
    +    done = threading.Event()
    +    fired = threading.Event()
    +    dump = tmp_path / "logs" / "watchdog.log"
    +    snapshot_calls = []
    +    exit_codes = []
    +
    +    def snapshot():
    +        snapshot_calls.append(1)
    +        return {"active_agents": 1, "draining": True}
    +
    +    def fake_exit(code):
    +        exit_codes.append(code)
    +        fired.set()
    +
    +    with patch("gateway.shutdown_watchdog.os._exit", side_effect=fake_exit):
    +        arm_shutdown_watchdog(
    +            0.15,
    +            done_event=done,
    +            snapshot_fn=snapshot,
    +            dump_path=dump,
    +            exit_code=9,
    +        )
    +        assert fired.wait(timeout=5.0), "watchdog did not fire"
    +
    +    assert exit_codes == [9]
    +    assert snapshot_calls == [1]
    +    assert dump.is_file()
    +    text = dump.read_text(encoding="utf-8")
    +    assert "shutdown_watchdog_fired" in text
    +    assert "faulthandler dump" in text
    +    assert get_shutdown_watchdog_dump_path(tmp_path).name == "gateway-shutdown-watchdog.log"
    +
    +
    [redacted:email]
    +async def test_loop_heartbeat_rewrites_until_cancelled(tmp_path):
    +    path = get_loop_heartbeat_path(tmp_path)
    +    task = asyncio.create_task(
    +        loop_heartbeat_forever(
    +            interval_s=0.05,
    +            start_time=12.0,
    +            home=tmp_path,
    +        )
    +    )
    +    try:
    +        # First write is immediate.
    +        for _ in range(50):
    +            if path.is_file():
    +                break
    +            await asyncio.sleep(0.02)
    +        assert path.is_file()
    +        first = path.read_text(encoding="utf-8")
    +        assert json.loads(first)["start_time"] == 12.0
    +
    +        # Poll until a refresh lands (monotonic / updated_at change).
    +        second = first
    +        for _ in range(100):
    +            await asyncio.sleep(0.03)
    +            second = path.read_text(encoding="utf-8")
    +            if second != first:
    +                break
    +        assert second != first
    +        assert json.loads(second)["start_time"] == 12.0
    +    finally:
    +        task.cancel()
    +        with pytest.raises(asyncio.CancelledError):
    +            await task
    +
    +
    +def test_gateway_runner_exposes_shutdown_watchdog_state():
    +    """Attrs used by stop()/start() exist after normal construction hooks."""
    +    from gateway.run import GatewayRunner
    +
    +    runner = object.__new__(GatewayRunner)
    +    runner._shutdown_watchdog_done = threading.Event()
    +    runner._loop_heartbeat_task = None
    +    runner._gateway_started_at = time.time()
    +    assert not runner._shutdown_watchdog_done.is_set()
    +    runner._shutdown_watchdog_done.set()
    +    assert runner._shutdown_watchdog_done.is_set()
    +    assert runner._loop_heartbeat_task is None

• #68444 fix(web/ddgs): isolate search in a process so GIL hangs cannot freeze Hermes (#68096)

  merged to main

  files: plugins/web/ddgs/_search_worker.py, plugins/web/ddgs/provider.py, tests/tools/test_web_providers_ddgs.py

  Show the diff

  • plugins/web/ddgs/_search_worker.py

    @@ -0,0 +1,113 @@
    +"""DDGS search child-process entrypoint (#68096).
    +
    +Invoked as ``python plugins/web/ddgs/_search_worker.py`` (script path from the
    +parent provider). Reads one JSON request from stdin, writes one JSON envelope
    +to stdout, then exits.
    +
    +Request::
    +    {"query": str, "safe_limit": int}
    +
    +Envelope::
    +    {"ok": true, "results": [...]}
    +    {"ok": false, "error": str}
    +
    +Optional test hooks (only when ``HERMES_DDGS_ALLOW_TEST_HOOKS=1``)::
    +    {"query": ..., "safe_limit": ..., "test_hook": "sleep"|"gil"|"success"|"error"|"empty"}
    +"""
    +
    +from __future__ import annotations
    +
    +import json
    +import os
    +import sys
    +import time
    +
    +
    +def _hold_gil(secs: int) -> None:
    +    """Block in a foreign call that keeps the GIL (ctypes.PyDLL).
    +
    +    Mirrors native ``primp`` holding the interpreter lock. ``PyDLL`` (unlike
    +    ``CDLL``/``WinDLL``) does not release the GIL around the call.
    +    """
    +    import ctypes
    +
    +    if sys.platform == "win32":
    +        lib = ctypes.PyDLL("kernel32")
    +        lib.Sleep.argtypes = [ctypes.c_uint]
    +        lib.Sleep(int(secs * 1000))
    +        return
    +
    +    lib = ctypes.PyDLL(None)
    +    try:
    +        sleep = lib.sleep
    +    except AttributeError:  # pragma: no cover — macOS libSystem fallback
    +        sleep = ctypes.PyDLL("/usr/lib/libSystem.B.dylib").sleep
    +    sleep.argtypes = [ctypes.c_uint]
    +    sleep(int(secs))
    +
    +
    +def _run_test_hook(hook: str) -> dict:
    +    if hook == "sleep":
    +        time.sleep(30)
    +        return {"ok": False, "error": "sleep hook returned unexpectedly"}
    +    if hook == "gil":
    +        _hold_gil(30)
    +        return {"ok": False, "error": "gil hook returned unexpectedly"}
    +    if hook == "success":
    +        return {
    +            "ok": True,
    +            "results": [
    +                {
    +                    "title": "Hit",
    +                    "url": "https://example.com",
    +                    "description": "body",
    +                    "position": 1,
    +                }
    +            ],
    +        }
    +    if hook == "empty":
    +        return {"ok": True, "results": []}
    +    if hook == "error":
    +        return {"ok": False, "error": "RuntimeError: boom"}
    +    return {"ok": False, "error": f"unknown test_hook: {hook!r}"}
    +
    +
    +def _write_envelope(envelope: dict) -> None:
    +    json.dump(envelope, sys.stdout)
    +    sys.stdout.flush()
    +
    +
    +def main() -> int:
    +    try:
    +        request = json.load(sys.stdin)
    +    except Exception as exc:  # noqa: BLE001
    +        _write_envelope({"ok": False, "error": f"invalid request: {exc}"})
    +        return 2
    +
    +    hook = request.get("test_hook")
    +    if hook:
    +        if os.environ.get("HERMES_DDGS_ALLOW_TEST_HOOKS") != "1":
    +            _write_envelope(
    +                {"ok": False, "error": "test_hook refused (hooks not enabled)"}
    +            )
    +            return 3
    +        envelope = _run_test_hook(str(hook))
    +        _write_envelope(envelope)
    +        return 0 if envelope.get("ok") else 1
    +
    +    query = str(request.get("query") or "")
    +    safe_limit = max(1, int(request.get("safe_limit") or 1))
    +    try:
    +        # Import inside main so script startup stays light / patchable.
    +        from plugins.web.ddgs.provider import _run_ddgs_search
    +
    +        results = _run_ddgs_search(query, safe_limit)
    +        _write_envelope({"ok": True, "results": results})
    +        return 0
    +    except Exception as exc:  # noqa: BLE001
    +        _write_envelope({"ok": False, "error": f"{type(exc).__name__}: {exc}"})
    +        return 1
    +
    +
    +if __name__ == "__main__":
    +    raise SystemExit(main())

  • plugins/web/ddgs/provider.py

    @@ -8,13 +8,24 @@
     The ``ddgs`` package is an optional dependency. ``is_available()`` reflects
     whether the package is importable; the plugin still registers either way so
     ``hermes tools`` can prompt the user to install it.
    +
    +Isolation note (#68096): ``ddgs``/``primp`` can block inside native code while
    +holding the Python GIL. A ``ThreadPoolExecutor`` + ``future.result(timeout=…)``
    +cap (see #52118) cannot fire in that state — the waiter never reacquires the
    +GIL — so the whole Hermes process freezes through Ctrl+C/SIGTERM. Each search
    +therefore runs in a disposable child process the parent can terminate/kill.
     """
     
     from __future__ import annotations
     
    -import concurrent.futures as _cf
    +import concurrent.futures as cf
    +import json
     import logging
    -from typing import Any, Dict
    +import os
    +import subprocess
    +import sys
    +import time
    +from typing import Any, Dict, Optional
     
     from agent.web_search_provider import WebSearchProvider
     
    @@ -23,18 +34,28 @@
     # Overall wall-clock cap for a single ddgs search. The DDGS constructor's
     # ``timeout`` only bounds individual HTTP requests; ddgs's multi-engine retry
     # loop has no overall cap, so a slow/rate-limited DuckDuckGo response can hang
    -# the (single, shared) agent loop indefinitely and block every platform
    -# (#36776). Enforce a hard cap here via a worker thread.
    +# the (single, shared) agent loop indefinitely (#36776). Enforce a hard cap
    +# here by killing a disposable worker process (#68096).
     _SEARCH_TIMEOUT_SECS = 30
     
    +# How often the parent polls stdout / interrupt flag while waiting.
    +_POLL_INTERVAL_SECS = 0.1
    +
    +# After terminate(), wait this long before escalating to kill().
    +_TERMINATE_GRACE_SECS = 1.0
    +
    +
    +class _SearchInterrupted(Exception):
    +    """Raised when tools.interrupt.is_interrupted() trips during a search wait."""
    +
     
     def _run_ddgs_search(query: str, safe_limit: int) -> list[dict[str, Any]]:
         """Run the blocking ddgs query and return normalized hits.
     
    -    Module-level (not a closure) so tests can patch it directly without
    -    spawning a real multi-second worker thread. ``DDGS(timeout=...)`` bounds
    +    Module-level (not a closure) so the child worker can import it and so
    +    tests can patch it for in-process unit tests. ``DDGS(timeout=…)`` bounds
         each individual HTTP request; the overall wall-clock cap is enforced by
    -    the caller via a future timeout.
    +    the parent via process timeout (#68096).
         """
         from ddgs import DDGS  # type: ignore
     
    @@ -55,6 +76,192 @@ def _run_ddgs_search(query: str, safe_limit: int) -> list[dict[str, Any]]:
         return results
     
     
    +# Optional test-only hook name forwarded to the child (see _search_worker.py).
    +# Production search() never sets this.
    +_test_hook: Optional[str] = None
    +
    +# Last worker Popen started by ``_run_ddgs_search_bounded`` (test reap checks).
    +_last_worker_proc: Optional[subprocess.Popen] = None
    +
    +
    +def _plugins_path_entry() -> str:
    +    """Return the ``sys.path`` entry that makes ``import plugins`` work.
    +
    +    Prefer the live ``plugins`` package location over counting ``dirname``s from
    +    this file — that stays correct for source checkouts and site-packages.
    +    """
    +    try:
    +        import plugins as plugins_pkg
    +
    +        pkg_file = getattr(plugins_pkg, "__file__", None)
    +        if pkg_file:
    +            return os.path.dirname(os.path.dirname(os.path.abspath(pkg_file)))
    +    except Exception:  # noqa: BLE001 — fall through to path-walk fallback
    +        pass
    +    return os.path.dirname(
    +        os.path.dirname(
    +            os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
    +        )
    +    )
    +
    +
    +def _terminate_and_reap(
    +    proc: Optional[subprocess.Popen],
    +    *,
    +    grace: float = _TERMINATE_GRACE_SECS,
    +) -> None:
    +    """Terminate a worker, escalate to kill, and wait so no orphan remains.
    +
    +    Does not close the parent's pipe ends — the caller must finish any
    +    ``communicate()``/reader first. Closing stdout while another thread is
    +    blocked in ``read()`` deadlocks on some platforms.
    +    """
    +    if proc is None:
    +        return
    +
    +    def _wait_until_dead(seconds: float) -> bool:
    +        deadline = time.monotonic() + seconds
    +        while time.monotonic() < deadline:
    +            if proc.poll() is not None:
    +                return True
    +            time.sleep(0.05)
    +        return proc.poll() is not None
    +
    +    try:
    +        if proc.poll() is None:
    +            proc.terminate()
    +            _wait_until_dead(grace)
    +        if proc.poll() is None:
    +            proc.kill()
    +            if not _wait_until_dead(grace):
    +                logger.warning("DDGS worker pid=%s did not exit after kill", proc.pid)
    +    except Exception as exc:  # noqa: BLE001 — best-effort cleanup
    +        logger.debug("DDGS worker reap error: %s", exc)
    +
    +
    +def _run_ddgs_search_bounded(query: str, safe_limit: int) -> list[dict[str, Any]]:
    +    """Run ``_run_ddgs_search`` in a disposable process with a hard deadline.
    +
    +    The parent never joins the child while it may be inside native code holding
    +    *its* GIL — it only polls a communicator thread and, on timeout/interrupt,
    +    terminates the child OS process. Raises ``TimeoutError``,
    +    ``_SearchInterrupted``, or ``RuntimeError``.
    +    """
    +    # Imported lazily so plugin import stays light for ``hermes tools`` probes.
    +    from tools.interrupt import is_interrupted
    +
    +    global _last_worker_proc
    +
    +    request: dict[str, Any] = {"query": query, "safe_limit": safe_limit}
    +    if _test_hook:
    +        request["test_hook"] = _test_hook
    +
    +    from tools.environments.local import _sanitize_subprocess_env
    +
    +    env = _sanitize_subprocess_env(dict(os.environ))
    +    if _test_hook:
    +        env["HERMES_DDGS_ALLOW_TEST_HOOKS"] = "1"
    +
    +    # Running the worker as a script puts ``plugins/web/ddgs/`` on ``sys.path[0]``,
    +    # which breaks ``import plugins...``. Prepend the path entry that makes the
    +    # live ``plugins`` package importable (source tree or site-packages).
    +    child_pythonpath = env.get("PYTHONPATH", "")
    +    path_entry = _plugins_path_entry()
    +    if path_entry and path_entry not in child_pythonpath.split(os.pathsep):
    +        env["PYTHONPATH"] = (
    +            path_entry + os.pathsep + child_pythonpath if child_pythonpath else path_entry
    +        )
    +
    +    worker_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "_search_worker.py")
    +    # Platform-only spawn knobs — stdin/stdout/stderr must stay as explicit
    +    # keyword args on the Popen call so scripts/check_subprocess_stdin.py can
    +    # see them (TUI gateway inherits stdin; #14036).
    +    extra_kwargs: dict[str, Any] = {}
    +    if sys.platform == "win32":
    +        # New process group so terminate/kill reach the worker cleanly on Windows.
    +        extra_kwargs["creationflags"] = subprocess.CREATE_NEW_PROCESS_GROUP
    +    else:
    +        # Own session so a hung primp/libcurl grandchild can be reaped with the worker.
    +        extra_kwargs["start_new_session"] = True
    +
    +    proc = subprocess.Popen(
    +        [sys.executable, worker_path],
    +        stdin=subprocess.PIPE,
    +        stdout=subprocess.PIPE,
    +        # DEVNULL avoids the classic deadlock where a chatty child fills the
    +        # stderr pipe buffer while the parent only drains stdout.
    +        stderr=subprocess.DEVNULL,
    +        env=env,
    +        text=True,
    +        **extra_kwargs,
    +    )
    +    _last_worker_proc = proc
    +
    +    # ``communicate`` runs in a side thread so the parent can poll interrupt /
    +    # deadline without blocking. Killing the child unblocks communicate.
    +    pool = cf.ThreadPoolExecutor(max_workers=1)
    +    fut = pool.submit(proc.communicate, json.dumps(request))
    +    timed_out = False
    +    interrupted = False
    +    raw = ""
    +    try:
    +        deadline = time.monotonic() + _SEARCH_TIMEOUT_SECS
    +        while True:
    +            if is_interrupted():
    +                interrupted = True
    +                break
    +            remaining = deadline - time.monotonic()
    +            if remaining <= 0:
    +                timed_out = True
    +                break
    +            try:
    +                out, _err = fut.result(timeout=min(_POLL_INTERVAL_SECS, remaining))
    +                raw = out or ""
    +                break
    +            except cf.TimeoutError:
    +                continue
    +    finally:
    +        _terminate_and_reap(proc)
    +        # After kill, communicate should return promptly; don't block forever.
    +        if not fut.done():
    +            try:
    +                out, _err = fut.result(timeout=_TERMINATE_GRACE_SECS)
    +                if not raw:
    +                    raw = out or ""
    +            except Exception:  # noqa: BLE001
    +                pass
    +        pool.shutdown(wait=False, cancel_futures=True)
    +
    +    if interrupted:
    +        raise _SearchInterrupted("DuckDuckGo search interrupted")
    +    if timed_out:
    +        raise TimeoutError(
    +            f"DuckDuckGo search timed out after {_SEARCH_TIMEOUT_SECS}s"
    +        )
    +
    +    raw = raw.strip()
    +    if not raw:
    +        raise RuntimeError(
    +            f"DDGS worker exited without a result (code={proc.poll()})"
    +        )
    +
    +    try:
    +        envelope = json.loads(raw)
    +    except json.JSONDecodeError as exc:
    +        raise RuntimeError(
    +            f"DDGS worker returned invalid JSON: {raw[:200]!r}"
    +        ) from exc
    +
    +    if not isinstance(envelope, dict):
    +        raise RuntimeError(f"DDGS worker returned an invalid envelope: {envelope!r}")
    +    if envelope.get("ok"):
    +        results = envelope.get("results") or []
    +        if not isinstance(results, list):
    +            raise RuntimeError("DDGS worker returned non-list results")
    +        return results
    +    raise RuntimeError(str(envelope.get("error") or "DDGS worker failed"))
    +
    +
     class DDGSWebSearchProvider(WebSearchProvider):
         """DuckDuckGo HTML-scrape search provider.
     
    @@ -94,9 +301,9 @@ def supports_extract(self) -> bool:
         def search(self, query: str, limit: int = 5) -> Dict[str, Any]:
             """Execute a DuckDuckGo search and return normalized results.
     
    -        The synchronous ``ddgs`` call is run in a worker thread with a hard
    -        wall-clock timeout (``_SEARCH_TIMEOUT_SECS``) so a hung search cannot
    -        block the shared agent loop indefinitely (#36776).
    +        The synchronous ``ddgs`` call runs in a disposable child process with
    +        a hard wall-clock timeout (``_SEARCH_TIMEOUT_SECS``) so a hung native
    +        ``primp`` call cannot freeze the Hermes process (#36776, #68096).
             """
             try:
                 import ddgs  # type: ignore  # noqa: F401 — availability probe
    @@ -110,40 +317,35 @@ def search(self, query: str, limit: int = 5) -> Dict[str, Any]:
             # in case the package ignores the hint.
             safe_limit = max(1, int(limit))
     
    -        # A fresh single-worker pool per call (rather than a module-level one)
    -        # is intentional: on timeout the blocking ddgs call cannot be cancelled
    -        # and keeps running, so a shared pool would serialise every later search
    -        # behind that hung worker. A per-call pool isolates each search from a
    -        # previously-hung one.
    -        pool = _cf.ThreadPoolExecutor(max_workers=1)
             try:
    -            future = pool.submit(_run_ddgs_search, query, safe_limit)
    -            try:
    -                web_results = future.result(timeout=_SEARCH_TIMEOUT_SECS)
    -            except _cf.TimeoutError:
    -                logger.warning(
    -                    "DDGS search timed out after %ds for query: %r",
    -                    _SEARCH_TIMEOUT_SECS, query,
    -                )
    -                return {
    -                    "success": False,
    -                    "error": (
    -                        f"DuckDuckGo search timed out after {_SEARCH_TIMEOUT_SECS}s — "
    -                        "DuckDuckGo may be rate-limiting or slow. Try again later "
    -                        "or switch to a different search provider."
    -                    ),
    -                }
    +            web_results = _run_ddgs_search_bounded(query, safe_limit)
    +        except TimeoutError:
    +            logger.warning(
    +                "DDGS search timed out after %ds for query: %r",
    +                _SEARCH_TIMEOUT_SECS,
    +                query,
    +            )
    +            return {
    +                "success": False,
    +                "error": (
    +                    f"DuckDuckGo search timed out after {_SEARCH_TIMEOUT_SECS}s — "
    +                    "DuckDuckGo may be rate-limiting or slow. Try again later "
    +                    "or switch to a different search provider."
    +                ),
    +            }
    +        except _SearchInterrupted:
    +            logger.info("DDGS search interrupted for query: %r", query)
    +            return {
    +                "success": False,
    +                "error": "DuckDuckGo search interrupted",
    +            }
             except Exception as exc:  # noqa: BLE001 — ddgs raises its own exceptions
                 logger.warning("DDGS search error: %s", exc)
                 return {"success": False, "error": f"DuckDuckGo search failed: {exc}"}
    -        finally:
    -            # Return immediately without joining the worker. On timeout the
    -            # already-running ddgs call can't be cancelled (cancel_futures only
    -            # affects not-yet-started work), so the worker runs to completion
    -            # on its own; it writes nothing shared, so leaking it is safe.
    -            pool.shutdown(wait=False, cancel_futures=True)
    -
    -        logger.info("DDGS search '%s': %d results (limit %d)", query, len(web_results), limit)
    +
    +        logger.info(
    +            "DDGS search '%s': %d results (limit %d)", query, len(web_results), limit
    +        )
             return {"success": True, "data": {"web": web_results}}
     
         def get_setup_schema(self) -> Dict[str, Any]:

  • tests/tools/test_web_providers_ddgs.py

    @@ -4,6 +4,7 @@
     - DDGSWebSearchProvider.is_available() — reflects package importability
     - DDGSWebSearchProvider.search() — happy path, missing package, runtime error
     - Result normalization (title, url, description, position)
    +- Process-isolated timeout / interrupt / GIL-hold / reap (#68096)
     - _is_backend_available("ddgs") / _get_backend() integration
     - web_extract returns a search-only error when ddgs is active
     """
    @@ -11,6 +12,7 @@
     
     import json
     import sys
    +import time
     import types
     
     import pytest
    @@ -52,6 +54,21 @@ def text(self, query, max_results=5):
         return fake
     
     
    +def _force_inprocess_search(monkeypatch, prov):
    +    """Route bounded search through the in-process helper.
    +
    +    Happy-path unit tests install a fake ``ddgs`` in the parent interpreter;
    +    spawn workers would not see that fake. Isolation behavior is covered by
    +    dedicated process tests below.
    +    """
    +    monkeypatch.setattr(
    +        prov,
    +        "_run_ddgs_search_bounded",
    +        lambda query, safe_limit: prov._run_ddgs_search(query, safe_limit),
    +        raising=True,
    +    )
    +
    +
     # ---------------------------------------------------------------------------
     # DDGSWebSearchProvider unit tests
     # ---------------------------------------------------------------------------
    @@ -98,9 +115,10 @@ def test_happy_path_normalizes_results(self, monkeypatch):
                 {"title": "B", "href": "https://b.example.com", "body": "desc B"},
                 {"title": "C", "href": "https://c.example.com", "body": "desc C"},
             ])
    -        from plugins.web.ddgs.provider import DDGSWebSearchProvider
    +        import plugins.web.ddgs.provider as prov
    +        _force_inprocess_search(monkeypatch, prov)
     
    -        result = DDGSWebSearchProvider().search("q", limit=5)
    +        result = prov.DDGSWebSearchProvider().search("q", limit=5)
     
             assert result["success"] is True
             web = result["data"]["web"]
    @@ -112,9 +130,10 @@ def test_accepts_url_key_as_fallback_for_href(self, monkeypatch):
             _install_fake_ddgs(monkeypatch, text_results=[
                 {"title": "A", "url": "https://a.example.com", "body": "desc A"},
             ])
    -        from plugins.web.ddgs.provider import DDGSWebSearchProvider
    +        import plugins.web.ddgs.provider as prov
    +        _force_inprocess_search(monkeypatch, prov)
     
    -        result = DDGSWebSearchProvider().search("q", limit=5)
    +        result = prov.DDGSWebSearchProvider().search("q", limit=5)
     
             assert result["success"] is True
             assert result["data"]["web"][0]["url"] == "https://a.example.com"
    @@ -124,9 +143,10 @@ def test_limit_is_respected(self, monkeypatch):
                 {"title": f"R{i}", "href": f"https://r{i}.example.com", "body": ""}
                 for i in range(10)
             ])
    -        from plugins.web.ddgs.provider import DDGSWebSearchProvider
    +        import plugins.web.ddgs.provider as prov
    +        _force_inprocess_search(monkeypatch, prov)
     
    -        result = DDGSWebSearchProvider().search("q", limit=3)
    +        result = prov.DDGSWebSearchProvider().search("q", limit=3)
     
             assert result["success"] is True
             assert len(result["data"]["web"]) == 3
    @@ -151,54 +171,42 @@ def blocked_import(name, *args, **kwargs):
     
         def test_runtime_error_returns_failure(self, monkeypatch):
             _install_fake_ddgs(monkeypatch, text_raises=RuntimeError("rate limited 202"))
    -        from plugins.web.ddgs.provider import DDGSWebSearchProvider
    +        import plugins.web.ddgs.provider as prov
    +        _force_inprocess_search(monkeypatch, prov)
     
    -        result = DDGSWebSearchProvider().search("q", limit=5)
    +        result = prov.DDGSWebSearchProvider().search("q", limit=5)
             assert result["success"] is False
             assert "rate limited" in result["error"] or "failed" in result["error"].lower()
     
         def test_empty_results(self, monkeypatch):
             _install_fake_ddgs(monkeypatch, text_results=[])
    -        from plugins.web.ddgs.provider import DDGSWebSearchProvider
    +        import plugins.web.ddgs.provider as prov
    +        _force_inprocess_search(monkeypatch, prov)
     
    -        result = DDGSWebSearchProvider().search("nothing", limit=5)
    +        result = prov.DDGSWebSearchProvider().search("nothing", limit=5)
             assert result["success"] is True
             assert result["data"]["web"] == []
     
    +    @pytest.mark.live_system_guard_bypass
         def test_hung_search_times_out_and_returns_failure(self, monkeypatch):
    -        """#36776: a ddgs call that never returns must be bounded by the
    -        wall-clock timeout and surface a failure instead of hanging the
    -        shared agent loop. We patch the blocking helper to wait on an Event
    -        (released in finally so no worker thread leaks past the test) and
    -        shrink the timeout; search() must return success=False promptly."""
    -        import threading
    -        import time
    -
    -        # ddgs must import-probe True for search() to proceed.
    +        """#36776 / #68096: a hung worker must be bounded by the wall-clock
    +        timeout and reaped — even when the child never returns to Python."""
             _install_fake_ddgs(monkeypatch)
    -        monkeypatch.delitem(sys.modules, "plugins.web.ddgs.provider", raising=False)
    -        import plugins.web.ddgs.provider as _prov
    -
    -        release = threading.Event()
    +        import plugins.web.ddgs.provider as prov
     
    -        def _blocking_search(query, safe_limit):
    -            release.wait(timeout=10)  # bounded so the worker can never truly leak
    -            return []
    +        monkeypatch.setattr(prov, "_test_hook", "sleep", raising=True)
    +        monkeypatch.setattr(prov, "_SEARCH_TIMEOUT_SECS", 0.4, raising=True)
    +        monkeypatch.setattr(prov, "_TERMINATE_GRACE_SECS", 0.5, raising=True)
    +        monkeypatch.setattr("tools.interrupt.is_interrupted", lambda: False)
     
    -        monkeypatch.setattr(_prov, "_run_ddgs_search", _blocking_search, raising=True)
    -        monkeypatch.setattr(_prov, "_SEARCH_TIMEOUT_SECS", 0.3, raising=True)
    +        start = time.monotonic()
    +        result = prov.DDGSWebSearchProvider().search("hangs forever", limit=5)
    +        elapsed = time.monotonic() - start
     
    -        try:
    -            start = time.monotonic()
    -            result = _prov.DDGSWebSearchProvider().search("hangs forever", limit=5)
    -            elapsed = time.monotonic() - start
    -
    -            assert result["success"] is False
    -            assert "timed out" in result["error"].lower()
    -            # Returned well before the worker's 10s wait — proves the cap fired.
    -            assert elapsed < 3.0, f"search did not return promptly ({elapsed:.1f}s)"
    -        finally:
    -            release.set()  # let the orphaned worker finish immediately
    +        assert result["success"] is False
    +        assert "timed out" in result["error"].lower()
    +        assert elapsed < 5.0, f"search did not return promptly ({elapsed:.1f}s)"
    +        _assert_worker_reaped(prov)
     
         def test_fast_search_not_affected_by_timeout_wrapper(self, monkeypatch):
             """Happy-path guard: the timeout wrapper must not break a normal,
    @@ -207,14 +215,115 @@ def test_fast_search_not_affected_by_timeout_wrapper(self, monkeypatch):
                 monkeypatch,
                 text_results=[{"title": "T", "href": "https://e.com", "body": "B"}],
             )
    -        from plugins.web.ddgs.provider import DDGSWebSearchProvider
    +        import plugins.web.ddgs.provider as prov
    +        _force_inprocess_search(monkeypatch, prov)
     
    -        result = DDGSWebSearchProvider().search("q", limit=5)
    +        result = prov.DDGSWebSearchProvider().search("q", limit=5)
             assert result["success"] is True
             assert result["data"]["web"][0]["url"] == "https://e.com"
             assert result["data"]["web"][0]["title"] == "T"
     
     
    +# ---------------------------------------------------------------------------
    +# Process isolation (#68096)
    +# ---------------------------------------------------------------------------
    +
    +
    +def _assert_worker_reaped(prov) -> None:
    +    """Assert the last DDGS worker process has exited."""
    +    proc = prov._last_worker_proc
    +    assert proc is not None, "expected a DDGS worker process to have been started"
    +    assert proc.poll() is not None, (
    +        f"DDGS worker still alive (pid={proc.pid}, returncode={proc.returncode})"
    +    )
    +
    +
    [redacted:email]_system_guard_bypass
    +class TestDDGSProcessIsolation:
    +    def test_gil_holding_worker_times_out_and_is_reaped(self, monkeypatch):
    +        """#68096: parent deadline still fires when the child holds its GIL."""
    +        _install_fake_ddgs(monkeypatch)
    +        import plugins.web.ddgs.provider as prov
    +
    +        monkeypatch.setattr(prov, "_test_hook", "gil", raising=True)
    +        monkeypatch.setattr(prov, "_SEARCH_TIMEOUT_SECS", 0.5, raising=True)
    +        monkeypatch.setattr(prov, "_TERMINATE_GRACE_SECS", 0.5, raising=True)
    +        monkeypatch.setattr("tools.interrupt.is_interrupted", lambda: False)
    +
    +        start = time.monotonic()
    +        result = prov.DDGSWebSearchProvider().search("gil hold", limit=5)
    +        elapsed = time.monotonic() - start
    +
    +        assert result["success"] is False
    +        assert "timed out" in result["error"].lower()
    +        assert elapsed < 5.0, f"GIL-hold search did not time out promptly ({elapsed:.1f}s)"
    +        _assert_worker_reaped(prov)
    +
    +    def test_interrupt_terminates_worker_promptly(self, monkeypatch):
    +        """TUI/gateway interrupt must kill the DDGS child before the deadline."""
    +        _install_fake_ddgs(monkeypatch)
    +        import plugins.web.ddgs.provider as prov
    +
    +        # Flip interrupt after the first poll so the wait loop observes it.
    +        calls = {"n": 0}
    +
    +        def _interrupt_after_poll():
    +            calls["n"] += 1
    +            return calls["n"] >= 2
    +
    +        monkeypatch.setattr(prov, "_test_hook", "sleep", raising=True)
    +        monkeypatch.setattr(prov, "_SEARCH_TIMEOUT_SECS", 30, raising=True)
    +        monkeypatch.setattr(prov, "_TERMINATE_GRACE_SECS", 0.5, raising=True)
    +        monkeypatch.setattr("tools.interrupt.is_interrupted", _interrupt_after_poll)
    +
    +        start = time.monotonic()
    +        result = prov.DDGSWebSearchProvider().search("interrupt me", limit=5)
    +        elapsed = time.monotonic() - start
    +
    +        assert result["success"] is False
    +        assert "interrupted" in result["error"].lower()
    +        assert elapsed < 5.0, f"interrupt did not return promptly ({elapsed:.1f}s)"
    +        _assert_worker_reaped(prov)
    +
    +    def test_spawned_worker_success_envelope(self, monkeypatch):
    +        """Real spawn path: success envelope round-trips through the pipe."""
    +        _install_fake_ddgs(monkeypatch)
    +        import plugins.web.ddgs.provider as prov
    +
    +        monkeypatch.setattr(prov, "_test_hook", "success", raising=True)
    +        monkeypatch.setattr(prov, "_SEARCH_TIMEOUT_SECS", 5, raising=True)
    +        monkeypatch.setattr("tools.interrupt.is_interrupted", lambda: False)
    +
    +        result = prov.DDGSWebSearchProvider().search("q", limit=5)
    +        assert result["success"] is True
    +        assert result["data"]["web"][0]["url"] == "https://example.com"
    +        _assert_worker_reaped(prov)
    +
    +    def test_spawned_worker_error_envelope(self, monkeypatch):
    +        """Real spawn path: error envelope becomes success=False."""
    +        _install_fake_ddgs(monkeypatch)
    +        import plugins.web.ddgs.provider as prov
    +
    +        monkeypatch.setattr(prov, "_test_hook", "error", raising=True)
    +        monkeypatch.setattr(prov, "_SEARCH_TIMEOUT_SECS", 5, raising=True)
    +        monkeypatch.setattr("tools.interrupt.is_interrupted", lambda: False)
    +
    +        result = prov.DDGSWebSearchProvider().search("q", limit=5)
    +        assert result["success"] is False
    +        assert "boom" in result["error"]
    +        _assert_worker_reaped(prov)
    +
    +    def test_no_orphan_after_successful_search(self, monkeypatch):
    +        _install_fake_ddgs(monkeypatch)
    +        import plugins.web.ddgs.provider as prov
    +
    +        monkeypatch.setattr(prov, "_test_hook", "empty", raising=True)
    +        monkeypatch.setattr("tools.interrupt.is_interrupted", lambda: False)
    +
    +        result = prov.DDGSWebSearchProvider().search("q", limit=5)
    +        assert result["success"] is True
    +        _assert_worker_reaped(prov)
    +
     # ---------------------------------------------------------------------------
     # Integration: _is_backend_available / _get_backend / check_web_api_key
     # ---------------------------------------------------------------------------