Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
84 changes: 80 additions & 4 deletions gateway/run.py
Original file line number Diff line number Diff line change
Expand Up @@ -67,6 +67,26 @@
_AGENT_CACHE_IDLE_TTL_SECS = 3600.0 # evict agents idle for >1h
_PLATFORM_CONNECT_TIMEOUT_SECS_DEFAULT = 30.0
_ADAPTER_DISCONNECT_TIMEOUT_SECS_DEFAULT = 5.0
# How long the shutdown path waits for in-flight ThreadPoolExecutor workers
# (agent turns) to finish before giving up and letting the CLI hard-exit
# backstop finalize the process. Bounded so a wedged worker can't strand the
# gateway "down but not exited" (non-daemon workers + concurrent.futures'
# atexit-join would otherwise block interpreter finalization). Overridable via
# HERMES_GATEWAY_EXECUTOR_DRAIN_TIMEOUT for ops tuning.
_EXECUTOR_DRAIN_TIMEOUT_SECS_DEFAULT = 8.0


def _executor_drain_timeout() -> float:
"""Return the bounded executor-drain timeout (seconds) for shutdown."""
raw = os.getenv("HERMES_GATEWAY_EXECUTOR_DRAIN_TIMEOUT", "").strip()
if raw:
try:
return max(0.0, float(raw))
except ValueError:
logger.warning(
"Ignoring invalid HERMES_GATEWAY_EXECUTOR_DRAIN_TIMEOUT=%r", raw
)
return _EXECUTOR_DRAIN_TIMEOUT_SECS_DEFAULT
_TELEGRAM_COMMAND_MENTION_RE = re.compile(r"(?<![\w:/])/([A-Za-z0-9][A-Za-z0-9_-]*)")
Comment thread
Kyzcreig marked this conversation as resolved.

# Greppable marker logged when a session auto-reset notice was supposed to be
Expand Down Expand Up @@ -14772,7 +14792,19 @@ def _get_executor(self) -> concurrent.futures.ThreadPoolExecutor:
return executor

def _shutdown_executor(self) -> None:
"""Stop the gateway-owned executor without touching the loop default."""
"""Stop the gateway-owned executor without touching the loop default.

Drains with a bounded wait so a worker that finishes promptly is joined
cleanly, but a *wedged* worker (an agent turn blocked on a slow tool or
network call) cannot stall shutdown indefinitely. Because
``ThreadPoolExecutor`` workers are non-daemon and ``concurrent.futures``
registers an ``atexit`` hook that joins them, an in-flight worker left
running here would otherwise block interpreter finalization for as long
as the turn takes — stranding the gateway "down but not exited" so
launchd/systemd can't relaunch it (observed: a 149s restart blackout).
The hard-exit backstop in the CLI entrypoint covers the wedged case;
this bounded drain makes the common case clean and logs what it saw.
"""
lock = getattr(self, "_executor_lock", None)
if lock is None:
return
Expand All @@ -14785,10 +14817,54 @@ def _shutdown_executor(self) -> None:
if executor is None:
return

# Count workers still alive so a slow/wedged drain is visible in logs.
try:
inflight = sum(
1 for t in threading.enumerate()
if t.name.startswith("hermes-gateway") and t.is_alive()
)
except Exception:
inflight = -1

drain_budget = _executor_drain_timeout()
_t0 = time.monotonic()
timed_out = False
try:
executor.shutdown(wait=False, cancel_futures=True)
except TypeError:
executor.shutdown(wait=False)
# cancel queued (not-yet-started) work, then bounded-join running workers
try:
executor.shutdown(wait=False, cancel_futures=True)
except TypeError: # Python <3.9 has no cancel_futures kwarg
executor.shutdown(wait=False)
# Bounded join: poll worker liveness rather than block forever.
_deadline = _t0 + drain_budget
while time.monotonic() < _deadline:
alive = [
t for t in threading.enumerate()
if t.name.startswith("hermes-gateway") and t.is_alive()
]
if not alive:
break
time.sleep(0.05)
else:
timed_out = True
except Exception as _e:
logger.debug("executor shutdown error: %s", _e)

_elapsed = time.monotonic() - _t0
if timed_out or _elapsed > 1.0 or (inflight and inflight > 0):
remaining = [
t.name for t in threading.enumerate()
if t.name.startswith("hermes-gateway") and t.is_alive()
]
logger.warning(
"Executor drain: %d worker(s) in-flight at shutdown, drained in "
"%.2fs (budget %.1fs, timed_out=%s)%s",
inflight, _elapsed, drain_budget, timed_out,
(f"; still-running=[{', '.join(remaining)}] — the CLI hard-exit "
f"backstop will finalize the process" if remaining else ""),
)
else:
logger.info("Executor drain: clean in %.2fs (no in-flight workers)", _elapsed)

def _decide_image_input_mode(self) -> str:
"""Resolve the image-input routing for the currently active model.
Expand Down
60 changes: 59 additions & 1 deletion hermes_cli/gateway.py
Original file line number Diff line number Diff line change
Expand Up @@ -4583,8 +4583,66 @@ def _atexit_hook() -> None:
raise
if not success:
_exit_diag("gateway.exit_nonzero")
sys.exit(1)
_force_process_exit(1)
return # unreachable in production; keeps type-checkers + tests calm
_exit_diag("gateway.exit_clean")
# start_gateway() has already completed graceful teardown (adapters
# disconnected, SessionDB closed, sessions persisted, executor drained).
# Force-exit now so a wedged non-daemon worker thread (an agent turn still
# running in the gateway ThreadPoolExecutor) cannot block interpreter
# finalization via concurrent.futures' atexit join — which strands the
# gateway "down but not exited" and prevents launchd/systemd from
# relaunching it (observed: a 149s restart blackout). Nothing of value is
# lost by hard-exiting the husk; the graceful work already happened.
_force_process_exit(0)


def _force_process_exit(code: int) -> None:
"""Flush stdio and terminate the process immediately.

Wrapped in a helper (rather than calling ``os._exit`` inline) so tests can
monkeypatch it and so any lingering non-daemon threads are logged for
post-mortem before we bypass normal interpreter finalization. ``os._exit``
skips ``atexit`` handlers by design — that is the whole point here, because
the ``concurrent.futures`` atexit join is exactly what hangs.
"""
# Best-effort diagnostics: name any non-daemon threads still alive that
# WOULD have blocked a normal exit. This is what to look for next time a
# restart stalls — grep the log for "force-exit: lingering non-daemon".
try:
import threading as _threading

lingering = [
t.name
for t in _threading.enumerate()
if t.is_alive()
and not t.daemon
and t is not _threading.main_thread()
]
if lingering:
try:
logger.warning(
"force-exit: %d lingering non-daemon thread(s) would have "
"blocked a normal exit: %s",
len(lingering), ", ".join(lingering[:20]),
)
except Exception:
pass
try:
sys.stderr.write(
"force-exit: lingering non-daemon thread(s) would have "
f"blocked a normal exit: {', '.join(lingering[:20])}\n"
)
except Exception:
pass
except Exception:
pass
for _stream in (sys.stdout, sys.stderr):
try:
_stream.flush()
except Exception:
pass
os._exit(code)


# =============================================================================
Expand Down
38 changes: 38 additions & 0 deletions tests/gateway/test_session_env.py
Original file line number Diff line number Diff line change
Expand Up @@ -402,3 +402,41 @@ async def test_gateway_executor_refuses_resurrection_after_shutdown():
finally:
runner._shutdown_executor()


@pytest.mark.asyncio
async def test_shutdown_executor_bounded_by_drain_timeout(monkeypatch):
"""A wedged worker must NOT make _shutdown_executor block indefinitely.

Non-daemon ThreadPoolExecutor workers + concurrent.futures' atexit-join
would otherwise hang interpreter finalization for as long as an in-flight
agent turn runs (observed: a 149s restart blackout). The bounded drain
caps that wait; the CLI hard-exit backstop finalizes the wedged husk.
"""
import threading
import time as _time
import gateway.run as gr

# Tiny drain budget so the test is fast; the real default is 8s.
monkeypatch.setattr(gr, "_executor_drain_timeout", lambda: 0.3)

runner = object.__new__(GatewayRunner)
release = threading.Event()

# Start a worker that blocks well past the drain budget.
fut = runner._get_executor().submit(lambda: release.wait(5.0))
# Give the worker a moment to actually start running.
_time.sleep(0.05)

t0 = _time.monotonic()
runner._shutdown_executor()
elapsed = _time.monotonic() - t0

# Bounded: returned near the budget, NOT after the worker's 5s block.
assert elapsed < 2.0, f"shutdown_executor blocked {elapsed:.2f}s (should be ~budget)"
# Let the wedged worker finish so it doesn't leak into other tests.
release.set()
try:
fut.result(timeout=5.0)
except Exception:
pass

56 changes: 53 additions & 3 deletions tests/hermes_cli/test_gateway.py
Original file line number Diff line number Diff line change
Expand Up @@ -38,6 +38,12 @@ def _install_fake_gateway_run(monkeypatch, start_gateway):
"get_gateway_runtime_snapshot",
lambda *a, **k: gateway.GatewayRuntimeSnapshot(manager="manual process"),
)
# ``run_gateway()`` now force-exits via ``os._exit`` at both the clean and
# failure tails (so a wedged non-daemon worker thread can't block interpreter
# finalization and strand the gateway). ``os._exit`` would kill the pytest
# process, so neutralize the force-exit by default; tests that assert on the
# exit behavior override this with their own recorder.
monkeypatch.setattr(gateway, "_force_process_exit", lambda code=0: None)


def test_run_gateway_exits_cleanly_on_keyboard_interrupt(monkeypatch, capsys):
Expand All @@ -63,21 +69,65 @@ def fake_asyncio_run(coro):

def test_run_gateway_exits_nonzero_when_start_gateway_reports_failure(monkeypatch):
calls = []
exits = []

def fake_start_gateway(*, replace, verbosity):
calls.append((replace, verbosity))
return object()

_install_fake_gateway_run(monkeypatch, fake_start_gateway)
monkeypatch.setattr(gateway.asyncio, "run", lambda coro: False)
# Override the helper's no-op stub with a recorder so we can assert the
# failure tail force-exits with code 1 (production uses os._exit, not
# SystemExit, so a wedged worker can't block the process teardown).
monkeypatch.setattr(gateway, "_force_process_exit", lambda code=0: exits.append(code))

with pytest.raises(SystemExit) as exc_info:
gateway.run_gateway(verbose=1, quiet=True, replace=True)
gateway.run_gateway(verbose=1, quiet=True, replace=True)

assert exc_info.value.code == 1
assert exits == [1]
assert calls == [(True, None)]


def test_run_gateway_force_exits_zero_after_clean_shutdown(monkeypatch):
"""Clean shutdown must force-exit(0), NOT fall through to normal interpreter
finalization — otherwise a wedged non-daemon executor worker blocks the
process from exiting and launchd/systemd can't relaunch it (the 149s
restart-blackout bug)."""
calls = []
exits = []

def fake_start_gateway(*, replace, verbosity):
calls.append((replace, verbosity))
return object()

_install_fake_gateway_run(monkeypatch, fake_start_gateway)
monkeypatch.setattr(gateway.asyncio, "run", lambda coro: True)
monkeypatch.setattr(gateway, "_force_process_exit", lambda code=0: exits.append(code))

gateway.run_gateway(replace=True)

assert exits == [0]
assert calls == [(True, 0)]


def test_force_process_exit_calls_os_exit_and_flushes(monkeypatch):
"""_force_process_exit must flush stdio and call os._exit with the code."""
flushed = []
exited = []
monkeypatch.setattr(
gateway.sys, "stdout", SimpleNamespace(flush=lambda: flushed.append("out"))
)
monkeypatch.setattr(
gateway.sys, "stderr", SimpleNamespace(flush=lambda: flushed.append("err"), write=lambda *a: None)
)
monkeypatch.setattr(gateway.os, "_exit", lambda code: exited.append(code))

gateway._force_process_exit(0)

assert exited == [0]
assert set(flushed) == {"out", "err"}


def test_run_gateway_refuses_root_in_official_docker(monkeypatch, tmp_path, capsys):
project_root = tmp_path / "opt" / "hermes"
(project_root / "docker").mkdir(parents=True)
Expand Down
Loading