diff --git a/agent/client_lifecycle.py b/agent/client_lifecycle.py index 9d1b0b16f3..bf65b6edaf 100644 --- a/agent/client_lifecycle.py +++ b/agent/client_lifecycle.py @@ -112,6 +112,11 @@ class ClientLifecycleMixin: owners = getattr(self, "_process_owner_task_ids", ()) for process in process_registry.list_sessions(): if process["owner_task_id"] in owners and process["status"] == "running": + # An explicitly persisted job (terminal persist_on_release=true) survives + # agent close — session end, compression, error recovery (#41225). The + # user can still stop it on purpose via process_manage kill. + if process.get("persist_on_release"): + continue process_registry.kill_process( process["session_id"], source="agent_close", consume_output=True, ) diff --git a/gateway/run_shutdown.py b/gateway/run_shutdown.py index 72ffa097e7..02d64e65bd 100644 --- a/gateway/run_shutdown.py +++ b/gateway/run_shutdown.py @@ -1737,7 +1737,11 @@ class GatewayShutdownMixin: def _kill_processes() -> None: from tools.process_registry import process_registry - _count_step("Shutdown (%s): killed %d tool subprocess(es)", process_registry.kill_all) + # Host shutdown: kill even persist_on_release jobs or they become + # PPID=1 orphans (#41225/#46778); an explicit source reaches them. + _count_step( + "Shutdown (%s): killed %d tool subprocess(es)", + lambda: process_registry.kill_all(source="gateway_shutdown")) def _mark_cron_interrupted() -> list: # kill_all() is global: a cron job mid-dispatch lost its tool subprocess and its agent thread may diff --git a/hermes_cli/cli_commands_mixin.py b/hermes_cli/cli_commands_mixin.py index 768b119214..eaba807d22 100644 --- a/hermes_cli/cli_commands_mixin.py +++ b/hermes_cli/cli_commands_mixin.py @@ -925,7 +925,7 @@ class CLICommandsMixin: return print(" No running background processes.") if running: print(f" Stopping {len(running)} background process(es)...") - print(f" ✅ Stopped {process_registry.kill_all()} process(es).") + print(f" ✅ Stopped {process_registry.kill_all(source='cli.stop')} process(es).") if n_async: from tools.async_delegation import interrupt_all print(f" ✅ Interrupted {interrupt_all(reason='/stop')} background delegation(s).") diff --git a/tests/gateway/test_shutdown_kill_off_loop.py b/tests/gateway/test_shutdown_kill_off_loop.py index 2fba183647..12950cdb65 100644 --- a/tests/gateway/test_shutdown_kill_off_loop.py +++ b/tests/gateway/test_shutdown_kill_off_loop.py @@ -23,7 +23,11 @@ def _make_phase_runner(monkeypatch, events): loop_thread = threading.current_thread() - def _fake_kill_all(task_id=None): + def _fake_kill_all(task_id=None, **kwargs): + # kwargs carry kill_all's keyword-only args (source, consume_output); + # the shutdown sweep passes source="gateway_shutdown" so a + # persist_on_release job (#41225) is still reached on host exit. + assert kwargs.get("source") == "gateway_shutdown", kwargs events.append(("kill_all", threading.current_thread())) return 2 diff --git a/tests/tools/test_persist_on_release.py b/tests/tools/test_persist_on_release.py new file mode 100644 index 0000000000..4375097917 --- /dev/null +++ b/tests/tools/test_persist_on_release.py @@ -0,0 +1,154 @@ +"""Lifecycle kill sweeps must skip terminal(background=true, persist_on_release=true) jobs (#41225). + +Background processes are killed from three agent-lifecycle paths — the release() +``kill_all`` sweep, a gateway turn timeout's ``kill_started_since``, and +``agent_close``'s owned-process loop. An explicitly persisted job must survive +all three, while an operator-driven stop still reaches it. +""" + +import pytest +from unittest.mock import MagicMock, patch + +from tools.process_registry import ProcessRegistry, ProcessSession, _CHECKPOINT_FIELDS + + +def _make_session(sid="proc_test123", task_id="t1", persist=False) -> ProcessSession: + s = ProcessSession(id=sid, command="sleep 60", task_id=task_id, started_at=0.0) + s.persist_on_release = persist + return s + + +@pytest.fixture() +def registry(): + return ProcessRegistry() + + +def _fake_kill_collector(registry): + """Replace kill_process so tests observe targeting without signalling.""" + calls = [] + + def fake_kill(session_id, **kwargs): + calls.append((session_id, kwargs)) + registry._running[session_id].exited = True + return {"status": "killed"} + + registry.kill_process = fake_kill + return calls + + +def test_kill_all_release_sweep_skips_persisted_sessions(registry): + """The default-source kill_all (agent release()) must not kill a + persist_on_release job owned by the task (#41225).""" + volatile = _make_session(sid="proc_volatile", task_id="session-a") + persisted = _make_session(sid="proc_persisted", task_id="session-a", persist=True) + registry._running[volatile.id] = volatile + registry._running[persisted.id] = persisted + calls = _fake_kill_collector(registry) + + assert registry.kill_all("session-a") == 1 + assert [c[0] for c in calls] == ["proc_volatile"] + assert persisted.exited is False + + +def test_gateway_turn_timeout_reap_skips_persisted_sessions(registry): + """A timed-out turn's kill_started_since (source=gateway_turn_timeout) is a + lifecycle sweep: a persisted job started mid-turn must survive it (#41225).""" + persisted = _make_session(sid="proc_persisted", task_id="session-a", persist=True) + volatile = _make_session(sid="proc_new", task_id="session-a") + registry._running[persisted.id] = persisted + registry._running[volatile.id] = volatile + calls = _fake_kill_collector(registry) + + assert registry.kill_started_since("session-a", frozenset(), source="gateway_turn_timeout") == 1 + assert [c[0] for c in calls] == ["proc_new"] + assert persisted.exited is False + + +def test_agent_close_owned_loop_skips_persisted_sessions(registry, monkeypatch): + """_close_task_resources' owned-process loop (source=agent_close) walks + list_sessions() directly, bypassing kill_all's filter: a persisted session + must be skipped there too (#41225).""" + from types import SimpleNamespace + + from agent.client_lifecycle import ClientLifecycleMixin + import tools.process_registry as registry_module + + persisted = _make_session(sid="proc_persisted", task_id="turn-1", persist=True) + volatile = _make_session(sid="proc_volatile", task_id="turn-1") + registry._running[persisted.id] = persisted + registry._running[volatile.id] = volatile + calls = _fake_kill_collector(registry) + + agent = SimpleNamespace(_process_owner_task_ids=("turn-1",)) + monkeypatch.setattr(registry_module, "process_registry", registry) + # _close_task_resources also runs cleanup_vm/cleanup_browser/release_computer_use; + # stub them so the loop under test is the only thing that runs. + import run_agent as ra + monkeypatch.setattr(ra, "cleanup_vm", lambda *a, **k: None) + monkeypatch.setattr(ra, "cleanup_browser", lambda *a, **k: None) + import tools.computer_use.tool as cu + monkeypatch.setattr(cu, "release_computer_use_session", lambda *a, **k: None) + + ClientLifecycleMixin._close_task_resources(agent, "turn-1") + + assert [c[0] for c in calls] == ["proc_volatile"] + assert all(c[1]["source"] == "agent_close" for c in calls) + assert persisted.exited is False + + +def test_explicit_operator_stop_still_reaches_persisted_sessions(registry): + """persist_on_release is an agent-lifecycle opt-out, never a protection + against being stopped on purpose: a caller-driven source still kills.""" + persisted = _make_session(sid="proc_persisted", task_id="session-a", persist=True) + registry._running[persisted.id] = persisted + calls = _fake_kill_collector(registry) + + assert registry.kill_all("session-a", source="process.kill") == 1 + assert [c[0] for c in calls] == ["proc_persisted"] + + +def test_list_sessions_flags_persist_on_release(registry): + """The agent_close loop reads list_sessions(); the flag must be visible + there for it (and for the agent) to act on.""" + persisted = _make_session(sid="proc_persisted", task_id="session-a", persist=True) + volatile = _make_session(sid="proc_volatile", task_id="session-a") + registry._running[persisted.id] = persisted + registry._running[volatile.id] = volatile + + entries = {e["session_id"]: e for e in registry.list_sessions()} + assert entries["proc_persisted"].get("persist_on_release") is True + assert "persist_on_release" not in entries["proc_volatile"] + + +def test_spawn_local_stamps_persist_on_release(registry, monkeypatch, tmp_path): + """spawn_local(persist_on_release=True) stamps the flag onto the minted + ProcessSession so every kill filter can see it (#41225).""" + import os + + from tools import terminal_tool_sudo + + # Stay off the real hermes home: the spawn-path env sanitizer resolves the + # real console-script install (_resolve_hermes_bin_dir), which the test + # suite's HomeIOGuard forbids. None == "no managed install found". + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + from tools.environments import local as local_env + monkeypatch.setattr(local_env, "_resolve_hermes_bin_dir", lambda: None) + monkeypatch.setattr(registry, "_track_started", lambda *a, **k: None) + monkeypatch.setattr(terminal_tool_sudo, "_rewrite_compound_background", lambda c: c) + monkeypatch.setattr(ProcessRegistry, "_scope_argv", lambda *a, **k: None) + fake_popen = MagicMock() + fake_popen.pid = 4242 + monkeypatch.setattr("subprocess.Popen", MagicMock(return_value=fake_popen)) + + session = registry.spawn_local( + "python -c 'import time; time.sleep(60)'", task_id="t1", persist_on_release=True) + assert session.persist_on_release is True + + session = registry.spawn_local("echo hi", task_id="t1") + assert session.persist_on_release is False + + +def test_checkpoint_carries_persist_on_release(): + """A crash-recovery checkpoint must carry the flag, or a persisted job + resurrects as killable by the next session's release sweep (#41225).""" + assert "persist_on_release" in _CHECKPOINT_FIELDS diff --git a/tools/process_registry.py b/tools/process_registry.py index f02463baf6..89a2f360ef 100644 --- a/tools/process_registry.py +++ b/tools/process_registry.py @@ -543,6 +543,8 @@ class ProcessSession: pid_scope: str = "host" # "host" for local/PTY PIDs, "sandbox" for env-local PIDs systemd_unit: str = "" # transient scope unit name when spawned under systemd-run handoff_note: str = "" # why a subagent handed this process to its parent (rides the notice) + persist_on_release: bool = False # opt out of agent-lifecycle cleanup (release()/turn-abandon kill + # sweeps), per terminal(background=true, persist_on_release=true) (#41225) # Watcher/notification routing (persisted for crash recovery) # systemd_unit: str = "" # transient scope unit name when spawned under systemd-run # (#70716) @@ -612,7 +614,7 @@ _CHECKPOINT_FIELDS = ( "started_at", "task_id", "owner_task_id", "session_key", *(f"watcher_{k}" for k in _WATCHER_ROUTE_KEYS), "watcher_interval", "parent_session_id", "notify_on_complete", "completion_output_chars", "watch_patterns", - "heartbeat_seconds") + "heartbeat_seconds", "persist_on_release") _CHECKPOINT_DEFAULTS = { f.name: ([] if f.name == "watch_patterns" else f.default) for f in ProcessSession.__dataclass_fields__.values() @@ -1212,10 +1214,12 @@ class ProcessRegistry(ProcessCheckpointMixin): def spawn_local( self, command: str, cwd: str = None, task_id: str = "", session_key: str = "", - env_vars: dict = None, use_pty: bool = False, owner_task_id: str = "") -> ProcessSession: + env_vars: dict = None, use_pty: bool = False, owner_task_id: str = "", + persist_on_release: bool = False) -> ProcessSession: """Spawn a background process locally (TERMINAL_ENV=local; other backends use spawn_via_env()). ``use_pty`` requests a pseudo-terminal via ptyprocess/pywinpty - for interactive CLIs, falling back to a plain pipe when unavailable or failing.""" + for interactive CLIs, falling back to a plain pipe when unavailable or failing. + ``persist_on_release`` keeps the process out of agent-lifecycle kill sweeps (#41225).""" # Bash parses ``A && B &`` as ``(A && B) &`` — a subshell that holds our stdout # pipe open forever when B is a long-running server. The rewriter turns it into # ``A && { B & }``. Lazy import: terminal_tool imports this module. @@ -1223,7 +1227,8 @@ class ProcessRegistry(ProcessCheckpointMixin): from tools.terminal_tool_sudo import _rewrite_compound_background as _rewrite_bg safe_command = _rewrite_bg(command) - session = self._new_session(command, task_id, owner_task_id, session_key, _resolve_safe_cwd(cwd or os.getcwd())) + session = self._new_session(command, task_id, owner_task_id, session_key, _resolve_safe_cwd(cwd or os.getcwd()), + persist_on_release=persist_on_release) pty_scope_attempted = False if use_pty: try: @@ -1310,12 +1315,14 @@ class ProcessRegistry(ProcessCheckpointMixin): def spawn_via_env( self, env: Any, command: str, cwd: str = None, task_id: str = "", session_key: str = "", - timeout: int = 10, owner_task_id: str = "") -> ProcessSession: + timeout: int = 10, owner_task_id: str = "", persist_on_release: bool = False) -> ProcessSession: """Spawn a background process inside a non-local backend's sandbox. The command is wrapped to capture its in-sandbox PID and redirect output to a log file that later execute() calls poll. No live pipe or stdin, but it runs in - the correct sandbox context.""" - session = self._new_session(command, task_id, owner_task_id, session_key, cwd, env_ref=env, pid_scope="sandbox") + the correct sandbox context. ``persist_on_release`` keeps the process out of + agent-lifecycle kill sweeps (#41225).""" + session = self._new_session(command, task_id, owner_task_id, session_key, cwd, env_ref=env, pid_scope="sandbox", + persist_on_release=persist_on_release) temp_dir = self._env_temp_dir(env) log_path, pid_path, exit_path = (f"{temp_dir}/hermes_bg_{session.id}.{ext}" for ext in ("log", "pid", "exit")) q = shlex.quote @@ -2367,6 +2374,8 @@ class ProcessRegistry(ProcessCheckpointMixin): entry.update(watch_patterns=list(s.watch_patterns), watch_hit=s._watch_hits > 0) if s.notify_on_complete: entry["notify_on_complete"] = True + if s.persist_on_release: + entry["persist_on_release"] = True if s.exited: entry["exit_code"] = s.exit_code entry["exited_at"] = s.exited_at @@ -2444,6 +2453,12 @@ class ProcessRegistry(ProcessCheckpointMixin): ``default``), shared across turns and sessions, not the turn's id.""" return frozenset(s.id for s in self.running_owned_by(task_id)) + # Kill sources that are agent-lifecycle cleanup (session end / compression / error + # recovery / turn abandon), not a deliberate stop: those must skip persist_on_release + # sessions. An explicit stop (process_manage kill, CLI /stop) passes its own source + # string and still reaches them. + _LIFECYCLE_KILL_SOURCES = frozenset({"kill_all", "gateway_turn_timeout", "agent_close"}) + def kill_started_since(self, task_id: str, baseline_ids, *, source: str) -> int: """Kill ``task_id`` processes created after ``baseline_ids``. Output is consumed so an abandoned turn can't enqueue a follow-up reviving work the @@ -2454,12 +2469,21 @@ class ProcessRegistry(ProcessCheckpointMixin): self, task_id: Optional[str] = None, *, exclude_ids: frozenset = frozenset(), source: str = "kill_all", consume_output: bool = False) -> int: """Kill all running processes, optionally only those ``task_id`` spawned (its ``owner_task_id``). - Returns count killed.""" + Returns count killed. + + Sessions with ``persist_on_release=True`` are skipped when ``source`` is an + agent-lifecycle sweep (the default ``kill_all`` release path, a gateway turn + timeout, or ``agent_close``): an explicitly persisted background job must survive + session end / compression / error recovery (#41225). An explicit operator stop + (``process.kill`` via process_manage, CLI /stop) passes its own source and still + reaches them, so the user can always stop a persisted process on purpose.""" + lifecycle = source in self._LIFECYCLE_KILL_SOURCES with self._lock: targets = [ s for s in self._running.values() if (task_id is None or s.owner_task_id == task_id) and s.id not in exclude_ids and not s.exited + and not (lifecycle and s.persist_on_release) ] return sum( self.kill_process(s.id, source=source, consume_output=consume_output).get("status") diff --git a/tools/terminal_tool.py b/tools/terminal_tool.py index efb397fb9e..5a88943b65 100644 --- a/tools/terminal_tool.py +++ b/tools/terminal_tool.py @@ -165,6 +165,7 @@ Foreground (default): returns INSTANTLY when the command finishes, even with a h Background: set background=true (returns a session_id) only for commands that must keep running independently after this tool call returns; add notify=true for bounded tasks, leave silent only for servers/daemons that never exit. Do not start sleep, timers, cooldowns, delays, or polling loops with background=true — to wait a fixed time, run the wait as a normal foreground command with a high enough timeout. After starting a server, verify readiness with a health check in a separate call (no blind sleep loops); manage with process(action="poll"/"wait"). Working directory: use 'workdir' for per-command cwd; when a command changes the session cwd (cd, pushd), trust the result's "cwd" field instead of prefixing every command with 'cd'. PTY: pty=true + background=true for interactive CLIs (they hang without a terminal); drive them with process(action="write"/"submit"). Local backend only. +Persist: background=true, persist_on_release=true keeps the job alive across agent lifecycle cleanup (session end, /new, compression, error recovery, stop-on-max-iterations). Use ONLY for long-running jobs the user explicitly wants to outlive the conversation; the user can still stop it on purpose. """ # Environment lifecycle state. @@ -1260,6 +1261,7 @@ def terminal_tool( _host_local: bool = False, _completion_output_chars: int = 0, heartbeat: int = 0, + persist_on_release: bool = False, ) -> str: """Execute *command* in the configured terminal environment; returns a JSON string. @@ -1273,6 +1275,10 @@ def terminal_tool( use it only for rare one-shot signals on long-lived processes. ``heartbeat`` (seconds, background-only, implies notify_on_complete) emits a "still running + output since last time" event every N seconds so the agent stays current on a long job without polling. + ``persist_on_release`` (background-only) keeps the process alive across agent-lifecycle + cleanup — session end, context compression, error recovery, max-iteration stop — all of + which kill the task's background processes; the user can still stop it on purpose via + process_manage kill (#41225). ``_completion_output_chars`` (internal) sizes the completion notification's output for a spawner whose output is the payload (a bot DM's reply); 0 keeps the usual tail. ``_host_local`` forces the local backend for Hermes-owned control-plane @@ -1341,6 +1347,7 @@ def terminal_tool( pty_disabled_reason=_PTY_DISABLED_REASON if pty_disabled else None, completion_output_chars=_completion_output_chars, heartbeat_seconds=heartbeat, + persist_on_release=persist_on_release, ) if plan.promoted_from_foreground_timeout is not None: result = _with_promoted_note(result, plan.promoted_from_foreground_timeout) @@ -1413,6 +1420,11 @@ TERMINAL_SCHEMA = { "type": "integer", "minimum": 60, "description": "With background=true: also notify every N seconds (min 60) with the output since the last notice. For long jobs you must react to mid-run (merge trains, full suites); implies notify=true." + }, + "persist_on_release": { + "type": "boolean", + "default": False, + "description": "With background=true: keep the process alive across agent lifecycle cleanup (session end, /new, context compression, error recovery, max-iteration stop). Use ONLY for long-running jobs the user explicitly wants to outlive the conversation (overnight batches, watchful daemons); it still dies with the host process, and the user (or a later turn via process kill) can stop it on purpose. Default false." } # Legacy aliases (unadvertised, still accepted): notify_on_complete # (bool) and watch_patterns (list). notify=true|[...] maps onto @@ -1442,6 +1454,7 @@ def _handle_terminal(args, **kw): notify_on_complete = args.get("notify_on_complete", False) watch_patterns = args.get("watch_patterns") heartbeat = args.get("heartbeat") or 0 + persist_on_release = bool(args.get("persist_on_release", False)) if not isinstance(heartbeat, int) or isinstance(heartbeat, bool) or heartbeat < 0: return tool_error("heartbeat must be a whole number of seconds (min 60).") if not args.get("background", False): @@ -1458,6 +1471,12 @@ def _handle_terminal(args, **kw): "tracked background process). Retry as terminal(command=..., " "background=true, pty=true)." ) + if persist_on_release: + return tool_error( + "persist_on_release only applies to background commands (a foreground " + "process is awaited inline and has nothing to persist). Retry as " + "terminal(command=..., background=true, persist_on_release=true)." + ) if notify is not None: if isinstance(notify, bool): notify_on_complete = notify @@ -1483,6 +1502,7 @@ def _handle_terminal(args, **kw): notify_on_complete=notify_on_complete, watch_patterns=watch_patterns, heartbeat=heartbeat, + persist_on_release=persist_on_release, ) diff --git a/tools/terminal_tool_background.py b/tools/terminal_tool_background.py index 4fe822a774..0ee535c5f2 100644 --- a/tools/terminal_tool_background.py +++ b/tools/terminal_tool_background.py @@ -87,9 +87,10 @@ def _stamp_gateway_routing(proc_session, get_session_env) -> None: def _spawn(process_registry, *, env, env_type, command, cwd, effective_task_id, task_id, - session_key, effective_pty): + session_key, effective_pty, persist_on_release: bool = False): common = dict(command=command, cwd=cwd, task_id=effective_task_id, - owner_task_id=task_id or effective_task_id, session_key=session_key) + owner_task_id=task_id or effective_task_id, session_key=session_key, + persist_on_release=persist_on_release) if env_type == "local": return process_registry.spawn_local( env_vars=env.env if hasattr(env, 'env') else None, use_pty=effective_pty, **common) @@ -145,6 +146,7 @@ def spawn_background_process( completion_output_chars: int = 0, pty_disabled_reason: Optional[str], heartbeat_seconds: int = 0, + persist_on_release: bool = False, ) -> str: """Spawn *command* as a tracked background process and return the JSON result. @@ -163,10 +165,12 @@ def spawn_background_process( proc_session = _spawn( process_registry, env=env, env_type=env_type, command=command, cwd=effective_cwd, effective_task_id=effective_task_id, task_id=task_id, session_key=session_key, - effective_pty=effective_pty, + effective_pty=effective_pty, persist_on_release=persist_on_release, ) result_data = {"output": "Background process started", "session_id": proc_session.id, "pid": proc_session.pid, "exit_code": 0, "error": None} + if persist_on_release: + result_data["persist_on_release"] = True if approval_note: result_data["approval"] = approval_note if pty_disabled_reason: diff --git a/tui_gateway/methods_slash.py b/tui_gateway/methods_slash.py index 0e52b3bfee..38d6e01df4 100644 --- a/tui_gateway/methods_slash.py +++ b/tui_gateway/methods_slash.py @@ -340,7 +340,9 @@ def _mirror_reload_mcp(sid, session, agent, arg) -> None: def _mirror_stop(sid, session, agent, arg) -> None: from tools.process_registry import process_registry - process_registry.kill_all() + # Deliberate user stop: an explicit source keeps it reaching + # persist_on_release jobs (#41225). + process_registry.kill_all(source="slash.stop") # name → mirror(sid, session, agent, arg); a falsy return means "no warning". diff --git a/tui_gateway/methods_tools.py b/tui_gateway/methods_tools.py index a3002958f8..966160f62b 100644 --- a/tui_gateway/methods_tools.py +++ b/tui_gateway/methods_tools.py @@ -246,7 +246,7 @@ def _(rid, params: dict) -> dict: # One-expression handlers: name → (fail_code, payload builder(params)). _SIMPLE_RPCS = { # Session-scoped view of the background process registry (desktop status stack). - "process.stop": (5010, lambda params: {"killed": _tools_mod("tools.process_registry").process_registry.kill_all()}), + "process.stop": (5010, lambda params: {"killed": _tools_mod("tools.process_registry").process_registry.kill_all(source="process.stop")}), # Re-read ``~/.hermes/.env`` (CLI ``/reload`` parity); built agents keep their pool, ``/new`` resolves fresh. "reload.env": (5015, lambda params: {"updated": int(_tools_mod("hermes_cli.config").reload_env())}), "plugins.list": (5032, lambda params: {"plugins": [