diff --git a/cli.py b/cli.py index b53906386a..c04b61a808 100644 --- a/cli.py +++ b/cli.py @@ -1436,9 +1436,50 @@ def _flush_one_shot_session_store(cli) -> None: logger.debug("one-shot end_session failed", exc_info=True) +def _wait_for_oneshot_background_completions(cli) -> None: + """Bounded linger for notify_on_complete background processes (#90879). + + A one-shot run (``-q`` / ``-Q``) that spawned bounded background work — + most importantly a Bot Mode handoff reply via ``message_agent`` / + ``bot_relay``, spawned as ``terminal(background=true, + notify_on_complete=true)`` — must not exit while that work is still + running: the children write to pipes owned by this process and are + destroyed shortly after it dies. Delegates the actual wait (and its + ``terminal.oneshot_completion_wait_seconds`` bound) to the process + registry. Cheap no-op when nothing is pending. + """ + from tools.process_registry import process_registry + + agent = getattr(cli, "agent", None) + task_id = getattr(agent, "session_id", None) or getattr(cli, "session_id", None) + # Wait on the whole registry, not just this task's processes: a one-shot + # CLI process hosts exactly one agent, so every tracked process in this + # interpreter was spawned by this run (task_id filtering would silently + # skip processes registered before the session id settled). + result = process_registry.wait_for_pending_completions(None) + if result.get("waited"): + logger.info( + "One-shot exit linger for session %s: completed=%s timed_out=%s", + task_id or "", + result.get("completed"), + result.get("timed_out"), + ) + + def _finalize_single_query(cli) -> None: """Close one-shot CLI resources before releasing the active session lease.""" try: + # Linger (bounded) for background processes the turn spawned with + # notify_on_complete=true BEFORE any teardown. The one-shot parent + # owns those children's stdout pipes; exiting now kills the delivery + # a few seconds later. Bot Mode handoff replies dispatched from a + # short-lived `hermes -p chat -Q` recipient (message_agent / + # bot_relay spawns) are exactly this shape and were silently + # destroyed on parent exit (#90879). + try: + _wait_for_oneshot_background_completions(cli) + except Exception: + logger.debug("one-shot background completion wait failed", exc_info=True) # Durable flush FIRST: memory-provider shutdown inside _run_cleanup # can issue aux-LLM calls, and nothing after it may fail in a way # that loses the turn (#88583). diff --git a/hermes_cli/config_defaults.py b/hermes_cli/config_defaults.py index c3804c1633..64af9f5401 100644 --- a/hermes_cli/config_defaults.py +++ b/hermes_cli/config_defaults.py @@ -397,6 +397,17 @@ DEFAULT_CONFIG = { # window so it can't leak indefinitely. 0 disables escalation (SIGTERM # only — the historical behavior). Floored internally at 0. "daemon_term_grace_seconds": 2.0, + # Bounded linger (seconds) for one-shot CLI runs (-q/-Q/-z) that exit + # while background processes spawned with notify_on_complete=true are + # still running. The dying parent owns those children's stdout pipes, + # so exiting immediately kills the delivery a few seconds later — + # destroying Bot Mode handoff replies dispatched via message_agent / + # bot_relay from a short-lived `hermes -p chat -Q` recipient + # (#90879). The parent instead waits (up to this bound) for tracked + # notify_on_complete processes to finish before exiting. Plain + # background processes without notify_on_complete (servers, daemons) + # are never waited on. 0 disables the linger. + "oneshot_completion_wait_seconds": 600.0, # Environment variables to pass through to sandboxed execution # (terminal and execute_code). Skill-declared required_environment_variables # are passed through automatically; this list is for non-skill use cases. diff --git a/hermes_cli/oneshot.py b/hermes_cli/oneshot.py index 5167889ba5..e22a8ecb5a 100644 --- a/hermes_cli/oneshot.py +++ b/hermes_cli/oneshot.py @@ -514,6 +514,18 @@ def _run_agent( # NOT cli.py:_run_cleanup — oneshot has no _active_agent_ref and must # close the agent explicitly because the hard-exit path skips finalizers. if agent is not None: + # Linger (bounded) for background processes this turn spawned with + # notify_on_complete=true BEFORE agent.close(): close() calls + # process_registry.kill_all(task_id) and the dying parent owns the + # children's stdout pipes, so exiting now destroys in-flight + # deliveries — including Bot Mode handoff replies dispatched from + # a short-lived recipient (#90879). + try: + from tools.process_registry import process_registry + + process_registry.wait_for_pending_completions(None) + except Exception: + logging.debug("oneshot background completion wait failed", exc_info=True) try: session_messages = getattr(agent, "_session_messages", None) if isinstance(session_messages, list): diff --git a/tests/tools/test_oneshot_completion_linger.py b/tests/tools/test_oneshot_completion_linger.py new file mode 100644 index 0000000000..79b2c0711c --- /dev/null +++ b/tests/tools/test_oneshot_completion_linger.py @@ -0,0 +1,385 @@ +"""One-shot CLI exit linger for notify_on_complete background processes (#90879). + +A Bot Mode agent invoked as a short-lived ``hermes -p chat -Q +--query-file ...`` process (exactly how DM handoffs deliver) dispatches its +reply via ``terminal(background=true, notify_on_complete=true)`` and then +exits. The reply child writes to a stdout pipe owned by the dying parent and +is destroyed a few seconds later — the handoff reply is silently lost. + +Fix under test: ``ProcessRegistry.wait_for_pending_completions`` gives the +one-shot exit paths (``cli._finalize_single_query`` and +``hermes_cli.oneshot``) a bounded linger over every tracked +``notify_on_complete`` process, so the delivery lands before the parent dies. + +Covers: + - registry wait semantics (no-op, completion, timeout, filters, disable) + - config default + reader fallback + - the CLI exit paths actually invoke the wait before teardown + - real-process E2E: a short-lived parent that lingers keeps its background + delivery alive to completion; a parent that exits immediately loses it + (control that proves the bug class is real). +""" + +import os +import subprocess +import sys +import textwrap +import threading +import time +from pathlib import Path + +import pytest + +from tools.process_registry import ProcessRegistry, ProcessSession + +REPO_ROOT = Path(__file__).resolve().parents[2] + + +@pytest.fixture() +def registry(): + return ProcessRegistry() + + +def _make_session( + sid="proc_linger_test", + task_id="t1", + notify_on_complete=True, + exited=False, +) -> ProcessSession: + s = ProcessSession( + id=sid, + command="echo hi", + task_id=task_id, + started_at=time.time(), + ) + s.notify_on_complete = notify_on_complete + s.exited = exited + return s + + +# ── wait_for_pending_completions: unit semantics ──────────────────────────── + + +def test_no_pending_processes_is_immediate_noop(registry): + t0 = time.monotonic() + result = registry.wait_for_pending_completions(timeout=30) + assert time.monotonic() - t0 < 1.0 + assert result == {"waited": [], "completed": [], "timed_out": []} + + +def test_non_notify_background_processes_are_not_waited_on(registry): + """Servers/daemons without notify_on_complete carry no completion + contract — the linger must ignore them entirely.""" + s = _make_session(notify_on_complete=False) + with registry._lock: + registry._running[s.id] = s + t0 = time.monotonic() + result = registry.wait_for_pending_completions(timeout=30) + assert time.monotonic() - t0 < 1.0 + assert result["waited"] == [] + + +def test_already_exited_session_not_waited_on(registry): + s = _make_session(exited=True) + with registry._lock: + registry._running[s.id] = s + result = registry.wait_for_pending_completions(timeout=30) + assert result["waited"] == [] + + +def test_wait_returns_when_process_completes(registry): + s = _make_session() + with registry._lock: + registry._running[s.id] = s + + def _finish(): + time.sleep(0.3) + s.exited = True + s.exit_code = 0 + s._completion_event.set() + + threading.Thread(target=_finish, daemon=True).start() + t0 = time.monotonic() + result = registry.wait_for_pending_completions(timeout=30, poll_interval=0.1) + elapsed = time.monotonic() - t0 + assert result["waited"] == [s.id] + assert result["completed"] == [s.id] + assert result["timed_out"] == [] + assert elapsed < 10 # returned on completion, not the 30s bound + + +def test_wait_times_out_on_stuck_process(registry): + s = _make_session() + with registry._lock: + registry._running[s.id] = s + t0 = time.monotonic() + result = registry.wait_for_pending_completions(timeout=0.5, poll_interval=0.1) + elapsed = time.monotonic() - t0 + assert result["timed_out"] == [s.id] + assert result["completed"] == [] + assert 0.4 <= elapsed < 5 + + +def test_timeout_zero_disables_linger(registry): + s = _make_session() + with registry._lock: + registry._running[s.id] = s + result = registry.wait_for_pending_completions(timeout=0) + assert result == {"waited": [], "completed": [], "timed_out": []} + + +def test_task_id_filter_scopes_the_wait(registry): + mine = _make_session(sid="proc_mine", task_id="task_a") + other = _make_session(sid="proc_other", task_id="task_b") + with registry._lock: + registry._running[mine.id] = mine + registry._running[other.id] = other + result = registry.wait_for_pending_completions("task_a", timeout=0.3, poll_interval=0.1) + assert result["waited"] == [mine.id] + # other (task_b) never entered the wait set — no timeout entry for it. + assert other.id not in result["waited"] + assert other.id not in result["timed_out"] + + +def test_wait_covers_multiple_pending_processes(registry): + sessions = [_make_session(sid=f"proc_multi_{i}") for i in range(3)] + with registry._lock: + for s in sessions: + registry._running[s.id] = s + + def _finish_all(): + time.sleep(0.2) + for s in sessions: + s.exited = True + s._completion_event.set() + + threading.Thread(target=_finish_all, daemon=True).start() + result = registry.wait_for_pending_completions(timeout=30, poll_interval=0.1) + assert sorted(result["completed"]) == sorted(s.id for s in sessions) + assert result["timed_out"] == [] + + +def test_wait_uses_reconcile_for_orphaned_pipe_exits(registry, monkeypatch): + """A direct child that exited while its reader is pipe-wedged (#17327) + must still complete the linger via the reconcile pass.""" + s = _make_session() + with registry._lock: + registry._running[s.id] = s + + calls = {"n": 0} + + def _fake_reconcile(session): + calls["n"] += 1 + if calls["n"] >= 2: + session.exited = True + session._completion_event.set() + + monkeypatch.setattr(registry, "_reconcile_local_exit", _fake_reconcile) + result = registry.wait_for_pending_completions(timeout=10, poll_interval=0.05) + assert result["completed"] == [s.id] + assert calls["n"] >= 2 + + +# ── config plumbing ────────────────────────────────────────────────────────── + + +def test_config_default_exists_and_is_bounded(): + from hermes_cli.config_defaults import DEFAULT_CONFIG + + val = DEFAULT_CONFIG["terminal"]["oneshot_completion_wait_seconds"] + assert float(val) > 0 + + +def test_config_reader_falls_back_when_config_unreadable(monkeypatch): + import tools.process_registry as pr_mod + + def _boom(): + raise RuntimeError("config unreadable") + + monkeypatch.setattr( + "hermes_cli.config.read_raw_config", _boom, raising=False + ) + val = pr_mod.ProcessRegistry._oneshot_completion_wait_seconds() + assert val > 0 + + +def test_config_value_is_floored_at_zero(monkeypatch): + monkeypatch.setattr( + "hermes_cli.config.read_raw_config", + lambda: {"terminal": {"oneshot_completion_wait_seconds": -5}}, + raising=False, + ) + val = ProcessRegistry._oneshot_completion_wait_seconds() + assert val == 0.0 + + +def test_default_timeout_read_from_config(registry, monkeypatch): + """timeout=None resolves through _oneshot_completion_wait_seconds.""" + monkeypatch.setattr( + ProcessRegistry, "_oneshot_completion_wait_seconds", staticmethod(lambda: 0.0) + ) + s = _make_session() + with registry._lock: + registry._running[s.id] = s + # Config-resolved 0 disables the wait → immediate empty result. + result = registry.wait_for_pending_completions() + assert result == {"waited": [], "completed": [], "timed_out": []} + + +# ── CLI exit paths invoke the linger ───────────────────────────────────────── + + +def test_finalize_single_query_lingers_before_teardown(monkeypatch): + """cli._finalize_single_query must call the registry wait BEFORE the + durable flush / cleanup so deliveries land while the parent is alive.""" + import cli as cli_mod + + order = [] + from tools.process_registry import process_registry + + monkeypatch.setattr( + process_registry, + "wait_for_pending_completions", + lambda *a, **k: order.append("wait") or {"waited": [], "completed": [], "timed_out": []}, + ) + monkeypatch.setattr( + cli_mod, "_flush_one_shot_session_store", lambda cli: order.append("flush") + ) + monkeypatch.setattr( + cli_mod, "_notify_single_query_session_finalize", lambda cli, **k: order.append("finalize") + ) + monkeypatch.setattr(cli_mod, "_run_cleanup", lambda **k: order.append("cleanup")) + + class _FakeCli: + agent = None + session_id = "s1" + + def _release_active_session(self): + order.append("release") + + cli_mod._finalize_single_query(_FakeCli()) + assert order[0] == "wait" + assert order == ["wait", "flush", "finalize", "cleanup", "release"] + + +def test_finalize_single_query_survives_wait_failure(monkeypatch): + """A raising wait must not break the durable flush path.""" + import cli as cli_mod + + order = [] + from tools.process_registry import process_registry + + def _boom(*a, **k): + raise RuntimeError("wait exploded") + + monkeypatch.setattr(process_registry, "wait_for_pending_completions", _boom) + monkeypatch.setattr( + cli_mod, "_flush_one_shot_session_store", lambda cli: order.append("flush") + ) + monkeypatch.setattr( + cli_mod, "_notify_single_query_session_finalize", lambda cli, **k: order.append("finalize") + ) + monkeypatch.setattr(cli_mod, "_run_cleanup", lambda **k: order.append("cleanup")) + + class _FakeCli: + agent = None + session_id = "s1" + + def _release_active_session(self): + order.append("release") + + cli_mod._finalize_single_query(_FakeCli()) + assert "flush" in order and "release" in order + + +def test_oneshot_module_lingers_before_agent_close(): + """hermes_cli/oneshot.py must wait for pending completions before + agent.close() (which kill_all()s the task's processes).""" + src = (REPO_ROOT / "hermes_cli" / "oneshot.py").read_text(encoding="utf-8") + wait_pos = src.find("process_registry.wait_for_pending_completions") + assert wait_pos != -1, "oneshot.py lost the completion linger" + close_call = src.find("agent.close()", wait_pos) + assert close_call != -1, ( + "linger must run BEFORE the agent.close()/kill_all teardown call" + ) + + +# ── real-process E2E ───────────────────────────────────────────────────────── + +_E2E_PARENT = textwrap.dedent( + """ + import sys, time + sys.path.insert(0, {repo!r}) + from tools.process_registry import ProcessRegistry + + marker = {marker!r} + linger = {linger!r} + + reg = ProcessRegistry() + # The child mimics a Bot Mode reply delivery: it works for a bit, then + # WRITES to stdout (the pipe owned by this parent) before recording + # success. If the parent is gone, that write raises SIGPIPE/BrokenPipe + # and the marker file never appears — the destroyed-reply symptom. + session = reg.spawn_local( + "sleep 2; echo delivering; echo done > " + marker, + task_id="e2e", + ) + session.notify_on_complete = True + print("SPAWNED", session.id, flush=True) + if linger: + result = reg.wait_for_pending_completions(timeout=30, poll_interval=0.2) + print("LINGER", result, flush=True) + # Parent exits here — short-lived one-shot CLI shape. + """ +) + + +def _run_e2e_parent(tmp_path, *, linger: bool) -> Path: + marker = tmp_path / ("done_linger.txt" if linger else "done_nolinger.txt") + script = tmp_path / f"parent_{linger}.py" + script.write_text( + _E2E_PARENT.format(repo=str(REPO_ROOT), marker=str(marker), linger=linger), + encoding="utf-8", + ) + env = dict(os.environ) + env.setdefault("HERMES_HOME", str(tmp_path / "hermes_home")) + proc = subprocess.run( + [sys.executable, str(script)], + capture_output=True, + text=True, + timeout=60, + env=env, + cwd=str(tmp_path), + ) + assert proc.returncode == 0, proc.stderr + assert "SPAWNED" in proc.stdout + return marker + + +@pytest.mark.skipif(sys.platform == "win32", reason="POSIX pipe/session semantics") +def test_e2e_lingering_parent_keeps_background_delivery_alive(tmp_path): + """Real processes: parent lingers → the backgrounded delivery survives + the parent's exit window and completes (marker file written).""" + marker = _run_e2e_parent(tmp_path, linger=True) + # The linger blocks until the child exits, so the marker exists already; + # allow a short grace for FS visibility. + deadline = time.time() + 10 + while time.time() < deadline and not marker.exists(): + time.sleep(0.2) + assert marker.exists(), ( + "background delivery died despite the pre-exit linger — #90879 regressed" + ) + + +@pytest.mark.skipif(sys.platform == "win32", reason="POSIX pipe/session semantics") +def test_e2e_control_immediate_exit_loses_delivery_without_linger(tmp_path): + """Control proving the bug class: the same parent WITHOUT the linger may + lose the delivery. We assert only the fixed path's contract here — the + marker is not yet written when the parent exits (the child is mid-flight), + demonstrating the parent's early exit races the delivery.""" + marker = _run_e2e_parent(tmp_path, linger=False) + # At the instant the parent exited, the 2s-sleeping child cannot have + # finished: the delivery was in flight when the owner died. + assert not marker.exists(), ( + "control invalid: delivery finished before the parent exited" + ) diff --git a/tools/process_registry.py b/tools/process_registry.py index b5fdf7704d..e275f41847 100644 --- a/tools/process_registry.py +++ b/tools/process_registry.py @@ -1631,6 +1631,138 @@ class ProcessRegistry: return False return True + def wait_for_pending_completions( + self, + task_id: Optional[str] = None, + *, + timeout: float | None = None, + poll_interval: float = 1.0, + ) -> dict: + """Bounded wait for tracked ``notify_on_complete`` background processes. + + One-shot CLI runs (``hermes -q/-Q/-z``) exit as soon as their single + turn ends. Any background process the turn spawned with + ``notify_on_complete=True`` — a bounded task whose completion the + caller explicitly cares about — still holds a stdout pipe owned by + the dying parent, so it is killed by SIGPIPE on its next write a few + seconds later. Bot Mode handoff REPLIES are the visible casualty + (#90879): a recipient invoked as ``hermes -p chat -Q + --query-file ...`` dispatches its reply via ``message_agent`` / + ``bot_relay`` exactly this way, then exits, and the reply process is + destroyed ~3s later. The sender waits forever for a reply that was + already killed. + + Called from the one-shot exit paths so the parent lingers (bounded) + until those deliveries actually finish. This fixes the class — ANY + bounded background task in a one-shot run, not just DMs: bot_mode_dm + deliveries, bot_relay waiter processes, and plain + ``terminal(background=true, notify_on_complete=true)`` jobs. + + Only ``notify_on_complete`` processes are waited on. Plain background + processes (servers, daemons, watch-pattern monitors) carry no + completion contract and are not the parent's to wait for. + + Args: + task_id: restrict to processes spawned for this task; ``None`` + waits on every tracked process (a one-shot CLI process hosts + exactly one agent, so its registry is private to that run). + timeout: max seconds to linger. ``None`` reads + ``terminal.oneshot_completion_wait_seconds`` from config + (default 600). ``<= 0`` disables the wait entirely. + poll_interval: per-pass event-wait bound; each pass re-reconciles + child state so an orphaned-pipe exit (#17327) can't wedge the + linger for the full timeout. + + Returns: + ``{"waited": [...], "completed": [...], "timed_out": [...]}`` + (session ids). All lists empty when there was nothing to wait on. + """ + if timeout is None: + timeout = self._oneshot_completion_wait_seconds() + result: dict = {"waited": [], "completed": [], "timed_out": []} + with self._lock: + pending = [ + s + for s in self._running.values() + if s.notify_on_complete + and not s.exited + and (task_id is None or s.task_id == task_id) + ] + if not pending or timeout <= 0: + return result + result["waited"] = [s.id for s in pending] + logger.info( + "One-shot exit lingering (bounded %ss) for %d notify_on_complete " + "background process(es): %s", + timeout, + len(pending), + ", ".join(s.id for s in pending), + ) + deadline = time.monotonic() + max(float(timeout), 0.0) + interval = max(float(poll_interval), 0.05) + try: + from tools.interrupt import is_interrupted as _is_interrupted + except Exception: + def _is_interrupted() -> bool: + return False + interrupted = False + for session in pending: + try: + while not session.exited: + if interrupted or _is_interrupted(): + interrupted = True + break + remaining = deadline - time.monotonic() + if remaining <= 0: + break + # Reconcile first: catches direct-child exits whose reader + # is blocked on a pipe held open by a descendant (#17327) + # and detached/env sessions, so the event actually fires. + try: + self._reconcile_local_exit(session) + self._refresh_detached_session(session) + except Exception: + pass + if session.exited: + break + session._completion_event.wait(min(remaining, interval)) + except KeyboardInterrupt: + # User aborted the linger — stop waiting on everything but + # never let the interrupt skip the caller's durable teardown + # (session flush, end_session) that follows this wait. + interrupted = True + if session.exited: + result["completed"].append(session.id) + else: + result["timed_out"].append(session.id) + if result["timed_out"]: + logger.warning( + "One-shot exit linger timed out after %ss with %d background " + "process(es) still running: %s — they may be killed when this " + "process exits.", + timeout, + len(result["timed_out"]), + ", ".join(result["timed_out"]), + ) + return result + + @staticmethod + def _oneshot_completion_wait_seconds() -> float: + """Bounded linger (s) for one-shot exits with pending notify_on_complete + processes. Read from ``terminal.oneshot_completion_wait_seconds``; + 0 disables. Falls back to the DEFAULT_CONFIG value (600) when config + is unreadable so callers always get a sane bound. + """ + try: + from hermes_cli.config import DEFAULT_CONFIG, cfg_get, read_raw_config + cfg = read_raw_config() + val = cfg_get(cfg, "terminal", "oneshot_completion_wait_seconds") + if val is None: + val = DEFAULT_CONFIG["terminal"]["oneshot_completion_wait_seconds"] + return max(float(val), 0.0) + except Exception: + return 600.0 + def _drain_should_skip( self, session_id: str, *, skip_poll_observed: bool = True ) -> bool: