fix: flush opt-in shared metrics on the sibling graceful os._exit paths

The same skipped-atexit loss on the other graceful hard exits: the compute host's
shutdown frame and orphan exit (both via close()), a SIGTERMed kanban worker (whose
normal one-shot exit already runs shutdown_runtimes) and an orphaned slash worker
each now run the bounded flush_before_hard_exit before os._exit. One test file covers
every path with a real NeMo Relay runtime and an open session; red on main.

hermes_cli/cli_single_query.py

@@ -443,6 +443,9 @@ def _install_single_query_signal_handlers(cli):
                 # store here or the worker's turn (and its usage deltas) never become durable (#88583 /
                 # #50881 class). Best-effort under the SIGALRM deadman above.
                 _flush_one_shot_session_store(cli)
+            # The one-shot exit's shared_metrics shutdown_runtimes is skipped too: bounded inside the deadman.
+            from hermes_cli.observability.shared_metrics_process import flush_before_hard_exit
+            flush_before_hard_exit()
             _flush_logging_and_stdio()
             _kill_foreground_and_exit()
         raise KeyboardInterrupt()

tests/hermes_cli/test_shared_metrics_exit_flush.py

@@ -1,8 +1,9 @@
 """Graceful ``os._exit`` paths run the shared-metrics exit flush their skipped atexit hook would have run.
 
-The gateway (SIGTERM, update restart, idle exit all funnel through ``_exit_after_graceful_shutdown``)
-and the compute host leave via ``os._exit``, so a session still open at the stop never wrote its
-summary rows to the opt-in local store. The flush is bounded and creates nothing when collection is off.
+The gateway (SIGTERM, update restart, idle exit all funnel through ``_exit_after_graceful_shutdown``),
+the compute host, a SIGTERMed kanban worker and an orphaned slash worker leave via ``os._exit``, so a
+session still open at the stop never wrote its summary rows to the opt-in local store. The flush is
+bounded and creates nothing when collection is off.
 """
 
 from __future__ import annotations
@@ -95,6 +96,63 @@ def test_graceful_gateway_exit_writes_the_open_sessions_summary(metrics_home, mo
     assert "hermes.session.count" in _metric_names(home)
 
 
+def test_compute_host_shutdown_frame_writes_the_open_sessions_summary(metrics_home):
+    import io
+
+    from tui_gateway.compute_host import ComputeHost
+
+    home, _ = metrics_home
+    _finished_task_in_open_session()
+    _wait_for(home, "hermes.task_run.finished")
+    host = ComputeHost(stdout=io.StringIO(), heartbeat_secs=0)
+
+    host.handle_frame({"type": "shutdown"})  # the reader thread os._exit(0)s right after this
+
+    assert "hermes.session.count" in _metric_names(home)
+
+
+def test_sigtermed_kanban_worker_writes_the_open_sessions_summary(metrics_home, monkeypatch):
+    import signal
+
+    import cli
+    from hermes_cli import cli_single_query
+
+    home, _ = metrics_home
+    _finished_task_in_open_session()
+    _wait_for(home, "hermes.task_run.finished")
+    handlers = {}
+    monkeypatch.setenv("HERMES_KANBAN_TASK", "t-1")
+    monkeypatch.setattr(signal, "signal", lambda signum, handler: handlers.__setitem__(signum, handler))
+    monkeypatch.setattr(signal, "alarm", lambda seconds: 0)
+    for name in ("_arm_exit_watchdog_on_shutdown_signal", "_flush_logging_and_stdio"):
+        monkeypatch.setattr(cli, name, lambda: None)
+    monkeypatch.setattr(cli, "_interrupt_agent_for_signal", lambda agent, signum: None)
+    monkeypatch.setattr(cli, "_flush_one_shot_session_store", lambda c: None)
+    monkeypatch.setattr(cli_single_query.os, "_exit", _fake_exit)
+    cli_single_query._install_single_query_signal_handlers(object())
+
+    with pytest.raises(_Exited):
+        handlers[signal.SIGTERM](signal.SIGTERM, None)
+
+    assert "hermes.session.count" in _metric_names(home)
+
+
+def test_orphaned_slash_worker_writes_the_open_sessions_summary(metrics_home, monkeypatch):
+    from tui_gateway import slash_worker
+
+    home, _ = metrics_home
+    _finished_task_in_open_session()
+    _wait_for(home, "hermes.task_run.finished")
+    exited = threading.Event()
+    monkeypatch.setattr(slash_worker, "_is_orphaned", lambda ppid: True)
+    monkeypatch.setattr(slash_worker.os, "_exit", lambda code: exited.set())
+
+    slash_worker._start_parent_death_watchdog(1)
+
+    assert exited.wait(10)
+    assert "hermes.session.count" in _metric_names(home)
+
+
 def test_graceful_exit_with_collection_off_creates_no_store(metrics_home, monkeypatch):
     import gateway.run as gateway_run
 

tui_gateway/compute_host.py

@@ -107,6 +107,9 @@ class ComputeHost:
         with contextlib.suppress(Exception):
             from tools.environments.base import kill_live_foreground_processes
             kill_live_foreground_processes()
+        # ...and the opt-in shared-metrics rows that atexit hook would have written (bounded).
+        from hermes_cli.observability.shared_metrics_process import flush_before_hard_exit
+        flush_before_hard_exit()
 
     def shutdown(self, *, reason: str = "shutdown", wait: float = 10.0) -> None:
         """Drain in-flight turns, then finalize every session.

tui_gateway/slash_worker.py

@@ -76,6 +76,8 @@ def _start_parent_death_watchdog(original_ppid) -> None:
         deadline = time.monotonic() + _ORPHAN_GRACE_S
         while _in_flight.is_set() and time.monotonic() < deadline:
             time.sleep(0.05)  # let an in-flight command finish/flush
+        from hermes_cli.observability.shared_metrics_process import flush_before_hard_exit
+        flush_before_hard_exit()  # os._exit skips the opt-in metrics runtimes' atexit flush (bounded)
         os._exit(0)
     threading.Thread(target=_loop, daemon=True).start()