fix(processes): persist_on_release keeps background jobs alive across lifecycle kill sweeps (#41225)

Background processes spawned with terminal(background=true) are killed from
three agent-lifecycle sweeps: agent release()'s kill_all, a gateway turn
timeout's kill_started_since, and agent close's owned-process loop. Jobs the
user explicitly wants to outlive the session (overnight batches, watchful
daemons) had no way to opt out.

Add terminal(background=true, persist_on_release=true):
- ProcessSession.persist_on_release, stamped by spawn_local/spawn_via_env,
  carried in crash-recovery checkpoints and exposed via list_sessions()
- kill_all skips persisted sessions only for lifecycle sources
  (_LIFECYCLE_KILL_SOURCES: kill_all, gateway_turn_timeout, agent_close);
  explicit operator stops (process_manage kill, /stop slash + RPC mirror,
  CLI /stop) now pass distinct sources so they still reach persisted jobs
- the agent_close owned-process loop in _close_task_resources skips
  persisted sessions the same way
- gateway shutdown keeps killing persisted jobs (source=gateway_shutdown):
  the host is going away and survivors would become PPID=1 orphans

Co-authored-by: salvaged from #109846 (persist_on_release plumbing) and
extended to the turn-timeout and agent_close paths.

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,
                     )

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

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).")

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
 

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

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")

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,
     )
 
 

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:

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".

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": [