fix(gateway): widen force-exit to SystemExit paths + os._exit regression tests (#53107)

Builds on the salvaged force-exit fix:
- Route the start_gateway() SystemExit paths (clean-fatal-config #51228,
  planned-restart, service-restart) through the same os._exit backstop. Those
  paths previously fell through to normal interpreter finalization, leaving
  them vulnerable to the SAME wedged-non-daemon-thread hang the boolean-return
  paths now avoid. main() catches SystemExit and converts its code (None->0,
  int->code, str->1) to os._exit. Every exit path is now wedge-proof.
- Document in the helper why bypassing atexit is safe (remove_pid_file +
  release_gateway_runtime_lock are performed explicitly in start_gateway
  teardown) and why logging is not flushed (synchronous RotatingFileHandlers).
- Tests: assert termination via os._exit not SystemExit (adapted from
  @AgenticSpark's PR #53122, a duplicate of #53121), plus SystemExit(78) is
  routed through os._exit(78) and SystemExit(None) maps to os._exit(0).

gateway/run.py

@@ -19297,21 +19297,61 @@ def main():
             data = yaml.safe_load(f) or {}
             config = GatewayConfig.from_dict(data)
     
-    # start_gateway() already performs graceful teardown before returning.
-    # Force-exit afterwards so a wedged non-daemon worker thread cannot block
-    # interpreter finalization and strand the gateway half-shut down.
-    success = asyncio.run(start_gateway(config))
-    _exit_after_graceful_shutdown(success)
+    # start_gateway() performs the full graceful teardown (adapters
+    # disconnected, sessions saved + flushed, SQLite closed, cron/MCP stopped,
+    # PID file + runtime lock released) before it returns OR raises SystemExit
+    # with an explicit code. Force-exit afterwards so a wedged non-daemon worker
+    # thread (e.g. a ThreadPoolExecutor tool/LLM call blocked with no timeout)
+    # cannot block interpreter finalization (Py_FinalizeEx joins all non-daemon
+    # threads, incl. concurrent.futures' _python_exit) and strand the gateway
+    # half-shut down with the supervisor unable to restart it (#53107).
+    #
+    # SystemExit is caught explicitly: start_gateway raises it on the
+    # clean-fatal-config (#51228), planned-restart, and service-restart paths,
+    # all of which complete teardown first. Routing those codes through the
+    # same os._exit backstop means EVERY exit path is wedge-proof, not just the
+    # boolean-return ones.
+    try:
+        success = asyncio.run(start_gateway(config))
+        exit_code = 0 if success else 1
+    except SystemExit as e:
+        # e.code may be None (→ 0), an int, or a str (→ 1, like CPython).
+        if e.code is None:
+            exit_code = 0
+        elif isinstance(e.code, int):
+            exit_code = e.code
+        else:
+            exit_code = 1
+    _exit_after_graceful_shutdown(exit_code)
 
 
-def _exit_after_graceful_shutdown(success: bool) -> None:
-    """Flush stdio and terminate immediately after graceful shutdown."""
+def _exit_after_graceful_shutdown(exit_code: int) -> None:
+    """Flush stdio + logging, then hard-exit with ``exit_code``.
+
+    Graceful teardown is already complete by the time this runs, so there is
+    nothing left that needs a clean interpreter shutdown. We deliberately use
+    ``os._exit`` (not ``sys.exit``): ``sys.exit`` raises ``SystemExit``, which
+    triggers ``Py_FinalizeEx`` → ``wait_for_thread_shutdown`` and joins every
+    non-daemon thread — exactly the hang (#53107) a wedged tool-worker causes.
+
+    ``os._exit`` also bypasses ``atexit`` handlers. That is safe here: the
+    gateway's atexit-registered cleanup (``remove_pid_file``,
+    ``release_gateway_runtime_lock``) is also performed explicitly inside
+    ``start_gateway``'s teardown, so nothing is leaked. Any NEW atexit handler
+    added to this process would be skipped — perform such cleanup in the
+    teardown path, not via atexit.
+
+    Logging is not flushed here: the gateway's handlers are synchronous
+    ``RotatingFileHandler``s that write each record immediately (no
+    ``MemoryHandler``/``QueueHandler`` buffering), so there is nothing pending.
+    Only stdio is buffered, so only stdio is flushed.
+    """
     for stream in (sys.stdout, sys.stderr):
         try:
             stream.flush()
         except Exception:
             pass
-    os._exit(0 if success else 1)
+    os._exit(exit_code)
 
 
 if __name__ == "__main__":

tests/gateway/test_gateway_process_exit.py

@@ -56,3 +56,72 @@ def test_main_force_exits_one_after_failed_shutdown(monkeypatch):
     assert exc_info.value.code == 1
     stdout.flush.assert_called_once_with()
     stderr.flush.assert_called_once_with()
+
+
+def test_main_terminates_via_os_exit_not_systemexit(monkeypatch):
+    """The terminating call must be os._exit, NOT sys.exit — SystemExit is
+    exactly what triggers the Py_FinalizeEx non-daemon-thread join hang this
+    fixes (#53107). If main() ever regresses to sys.exit(), SystemExit would
+    propagate instead of our os._exit sentinel and this test would fail.
+
+    Test contributed by @AgenticSpark (PR #53122, duplicate of #53121)."""
+    async def fake_start_gateway(config=None):
+        return False
+
+    stdout = SimpleNamespace(flush=Mock())
+    stderr = SimpleNamespace(flush=Mock())
+
+    monkeypatch.setattr(gateway_run, "start_gateway", fake_start_gateway)
+    monkeypatch.setattr(gateway_run.os, "_exit", _raise_exit)
+    monkeypatch.setattr(gateway_run.sys, "argv", ["gateway.run"])
+    monkeypatch.setattr(gateway_run.sys, "stdout", stdout)
+    monkeypatch.setattr(gateway_run.sys, "stderr", stderr)
+
+    # Our os._exit sentinel must be what terminates main() — not SystemExit.
+    with pytest.raises(_ExitCalled):
+        gateway_run.main()
+
+
+def test_main_routes_systemexit_through_os_exit(monkeypatch):
+    """start_gateway raises SystemExit on the clean-fatal-config (#51228),
+    planned-restart, and service-restart paths. main() must catch it and route
+    the carried code through os._exit too, so those paths are equally wedge-proof
+    (#53107) — a SystemExit propagating to interpreter finalization would join a
+    stuck non-daemon worker and hang. Verifies the explicit code (e.g. 78) is
+    preserved through the os._exit backstop."""
+    async def fake_start_gateway(config=None):
+        raise SystemExit(78)
+
+    stdout = SimpleNamespace(flush=Mock())
+    stderr = SimpleNamespace(flush=Mock())
+
+    monkeypatch.setattr(gateway_run, "start_gateway", fake_start_gateway)
+    monkeypatch.setattr(gateway_run.os, "_exit", _raise_exit)
+    monkeypatch.setattr(gateway_run.sys, "argv", ["gateway.run"])
+    monkeypatch.setattr(gateway_run.sys, "stdout", stdout)
+    monkeypatch.setattr(gateway_run.sys, "stderr", stderr)
+
+    with pytest.raises(_ExitCalled) as exc_info:
+        gateway_run.main()
+
+    # The SystemExit(78) must be converted to os._exit(78), not propagated.
+    assert exc_info.value.code == 78
+    stdout.flush.assert_called_once_with()
+    stderr.flush.assert_called_once_with()
+
+
+def test_main_systemexit_none_code_maps_to_zero(monkeypatch):
+    """SystemExit() with no code (or None) is a clean exit → os._exit(0)."""
+    async def fake_start_gateway(config=None):
+        raise SystemExit()
+
+    monkeypatch.setattr(gateway_run, "start_gateway", fake_start_gateway)
+    monkeypatch.setattr(gateway_run.os, "_exit", _raise_exit)
+    monkeypatch.setattr(gateway_run.sys, "argv", ["gateway.run"])
+    monkeypatch.setattr(gateway_run.sys, "stdout", SimpleNamespace(flush=Mock()))
+    monkeypatch.setattr(gateway_run.sys, "stderr", SimpleNamespace(flush=Mock()))
+
+    with pytest.raises(_ExitCalled) as exc_info:
+        gateway_run.main()
+
+    assert exc_info.value.code == 0