From 87603ada1508bac61ce4aec10c5e5e62a783fd86 Mon Sep 17 00:00:00 2001 From: Stefan Verhey Date: Thu, 23 Jul 2026 08:06:33 +0200 Subject: [PATCH 1/2] fix(cli): recover detached background processes from checkpoint on CLI startup and shutdown Port the three approved process-registry recovery commits onto current main (origin/main @ de5ece994): 1. recover_from_checkpoint idempotency guard + recover_and_log wrapper 2. CLI startup recovery hook (_recover_background_processes_on_startup) 3. CLI shutdown checkpoint flush in _run_cleanup Preserves: - PID start-time / recycled-PID safety - sandbox entries never recovered - best-effort recovery that cannot block startup - no kill-on-CLI-exit behavior - Python 3.9 compatibility - prompt-cache and tool-schema stability Tests: 117/117 process_registry + 3/3 CLI recovery tests pass. Round-trip: spawn -> checkpoint -> clear -> recover -> list -> kill verified. --- cli.py | 31 +++++++ tests/cli/test_cli_shutdown_checkpoint.py | 19 ++++ tests/cli/test_cli_startup_recovery.py | 38 ++++++++ tests/tools/test_process_registry.py | 100 ++++++++++++++++++++++ tools/process_registry.py | 33 ++++++- 5 files changed, 218 insertions(+), 3 deletions(-) create mode 100644 tests/cli/test_cli_shutdown_checkpoint.py create mode 100644 tests/cli/test_cli_startup_recovery.py diff --git a/cli.py b/cli.py index afbdd6cb76ab..4eef20a6bc7a 100644 --- a/cli.py +++ b/cli.py @@ -1003,6 +1003,24 @@ def _mark_tui_input_modes_active() -> None: _tui_input_modes_active = True +def _recover_background_processes_on_startup() -> None: + """Recover detached background processes from the checkpoint file. + + The ProcessRegistry is an in-memory singleton; background processes spawned + with start_new_session=True survive a CLI exit as detached OS processes but + become invisible to process(action='list') on the next CLI start. The + gateway already calls recover_from_checkpoint() at startup; the CLI must do + the same so pollers/servers/watchers from a previous session are + re-adopted (read-only status + kill; no output pipe). Best-effort: any + failure is swallowed so a corrupt checkpoint can never block CLI startup. + """ + try: + from tools.process_registry import process_registry + process_registry.recover_and_log() + except Exception: + logger.debug("Background-process recovery on CLI startup failed", exc_info=True) + + def _prepare_deferred_agent_startup() -> None: """Run Termux-deferred agent discovery before the first real agent turn.""" global _deferred_agent_startup_done @@ -1202,6 +1220,14 @@ def _run_cleanup(*, notify_session_finalize: bool = True): shutdown_cached_clients() except Exception: pass + # Flush a final process checkpoint so the next CLI startup can recover + # detached background processes even if this exit bypassed the per-spawn + # writes (e.g. a fast /quit right after a spawn). Best-effort. + try: + from tools.process_registry import process_registry + process_registry._write_checkpoint() + except Exception: + pass # Shut down memory provider (on_session_end + shutdown_all) at actual # session boundary — NOT per-turn inside run_conversation(). if notify_session_finalize: @@ -13388,6 +13414,11 @@ def _prewarm_agent_runtime() -> None: self._startup_skills_line_shown = True self._console_print() + # Recover detached background processes from the checkpoint file so + # pollers/servers/watchers from a previous CLI session reappear in + # process(action='list'). Best-effort; never blocks startup. + _recover_background_processes_on_startup() + # State for async operation self._agent_running = False self._pending_input = queue.Queue() # For normal input (commands + new queries) diff --git a/tests/cli/test_cli_shutdown_checkpoint.py b/tests/cli/test_cli_shutdown_checkpoint.py new file mode 100644 index 000000000000..9c089c46accf --- /dev/null +++ b/tests/cli/test_cli_shutdown_checkpoint.py @@ -0,0 +1,19 @@ +"""CLI shutdown must flush a final process checkpoint so a crash that bypasses +atexit still leaves the most recent background-process state for the next +startup recovery.""" + + +def test_run_cleanup_writes_final_checkpoint(monkeypatch): + writes = [] + monkeypatch.setattr( + "tools.process_registry.process_registry._write_checkpoint", + lambda: writes.append(1), + ) + # _run_cleanup touches many subsystems; stub the heavy ones so the test + # only asserts the checkpoint write happened. + monkeypatch.setattr("cli._cleanup_all_terminals", lambda *a, **k: None) + monkeypatch.setattr("cli._cleanup_all_browsers", lambda *a, **k: None) + monkeypatch.setattr("tools.async_delegation.interrupt_all", lambda reason="": 0, raising=False) + from cli import _run_cleanup + _run_cleanup(notify_session_finalize=False) + assert len(writes) == 1 diff --git a/tests/cli/test_cli_startup_recovery.py b/tests/cli/test_cli_startup_recovery.py new file mode 100644 index 000000000000..777b48380a7b --- /dev/null +++ b/tests/cli/test_cli_startup_recovery.py @@ -0,0 +1,38 @@ +"""CLI startup must recover detached background processes from checkpoint. + +Regression test for the second root cause of disappearing pollers: the CLI +never called recover_from_checkpoint(), so background processes spawned in a +previous CLI session (pollers, servers) became invisible to +process(action='list') while their OS processes kept running. +""" + + +def test_cli_run_conversation_recovers_background_processes(monkeypatch): + """The CLI conversation loop must call process_registry.recover_and_log() + once at startup, after the agent is built and before the first user turn.""" + calls = [] + + def fake_recover_and_log(): + calls.append(1) + return 0 + + monkeypatch.setattr( + "tools.process_registry.process_registry.recover_and_log", + fake_recover_and_log, + ) + from cli import _recover_background_processes_on_startup + _recover_background_processes_on_startup() + assert len(calls) == 1 + + +def test_cli_startup_recovery_swallows_errors(monkeypatch): + """A failure in recover_and_log must never raise into CLI startup.""" + def fake_recover_and_log(): + raise RuntimeError("checkpoint disk exploded") + monkeypatch.setattr( + "tools.process_registry.process_registry.recover_and_log", + fake_recover_and_log, + ) + from cli import _recover_background_processes_on_startup + # Must not raise. + _recover_background_processes_on_startup() diff --git a/tests/tools/test_process_registry.py b/tests/tools/test_process_registry.py index beeb41fe1ea1..d799cd152352 100644 --- a/tests/tools/test_process_registry.py +++ b/tests/tools/test_process_registry.py @@ -2229,3 +2229,103 @@ def test_disabled_passes_through(self, monkeypatch): monkeypatch.setattr(pr, "process_registry", reg) out = json.loads(pr._handle_process({"action": "log", "session_id": sess.id})) assert "zzzopaque1234567890abcdef" in out["output"] + + +class TestCheckpointRecovery: + """Recovery from checkpoint: idempotency, recycled-PID safety, round-trip.""" + + def test_recover_idempotent_skips_already_tracked(self, registry, tmp_path): + """A session already in _running must not be overwritten by recovery.""" + checkpoint = tmp_path / "procs.json" + checkpoint.write_text(json.dumps([{ + "session_id": "proc_live", + "command": "sleep 999", + "pid": os.getpid(), + "task_id": "t1", + "session_key": "sk1", + }])) + # Pre-populate the live registry with a non-detached session of the same id + existing = _make_session(sid="proc_live") + existing.detached = False + registry._running["proc_live"] = existing + + with patch("tools.process_registry.CHECKPOINT_PATH", checkpoint): + recovered = registry.recover_from_checkpoint() + # The existing live (non-detached) entry must win — recovery skips it. + assert recovered == 0 + assert registry.get("proc_live") is existing + assert registry.get("proc_live").detached is False + + def test_recover_and_log_returns_count(self, registry, tmp_path, caplog): + checkpoint = tmp_path / "procs.json" + checkpoint.write_text(json.dumps([{ + "session_id": "proc_live", + "command": "sleep 999", + "pid": os.getpid(), + "task_id": "t1", + }])) + with patch("tools.process_registry.CHECKPOINT_PATH", checkpoint): + with caplog.at_level("INFO"): + n = registry.recover_and_log() + assert n == 1 + assert any("recover" in r.message.lower() for r in caplog.records) + + def test_spawn_checkpoint_clear_recover_list(self, registry, tmp_path): + """A process that was running, got checkpointed, then the in-memory + registry was cleared (simulating a CLI restart), must reappear in + list_sessions() after recover_from_checkpoint() with detached=True, + the correct PID, and status=running — as long as the OS process is + still alive.""" + import subprocess, time as _time + # Spawn a real short-lived sleep so the PID is genuinely alive. + proc = subprocess.Popen(["sleep", "30"], stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL) + try: + _time.sleep(0.2) # let it start + assert proc.poll() is None # alive + sid = "proc_roundtrip" + s = ProcessSession( + id=sid, command="sleep 30", pid=proc.pid, + host_start_time=registry._safe_host_start_time(proc.pid), + pid_scope="host", started_at=_time.time(), + task_id="t1", session_key="sk1", + ) + registry._running[sid] = s + with patch("tools.process_registry.CHECKPOINT_PATH", tmp_path / "procs.json"): + registry._write_checkpoint() + # Simulate CLI restart: wipe in-memory registry. + registry._running.clear() + registry._finished.clear() + assert registry.list_sessions() == [] + # Recover. + n = registry.recover_from_checkpoint() + assert n == 1 + sessions = registry.list_sessions() + assert len(sessions) == 1 + got = sessions[0] + assert got["session_id"] == sid + assert got["pid"] == proc.pid + assert got["status"] == "running" + assert got.get("detached") is True + finally: + proc.kill() + proc.wait(timeout=5) + + def test_recycled_pid_not_adopted_on_recovery(self, registry, tmp_path): + """A checkpoint entry whose PID is alive but whose kernel start time + no longer matches (PID was recycled onto an unrelated process) must + NOT be adopted by recovery.""" + import os + checkpoint = tmp_path / "procs.json" + # Use our own PID but a bogus start time that will never match. + checkpoint.write_text(json.dumps([{ + "session_id": "proc_recycled", + "command": "sleep 999", + "pid": os.getpid(), + "host_start_time": 1, # impossibly old — guaranteed mismatch + "task_id": "t1", + }])) + with patch("tools.process_registry.CHECKPOINT_PATH", checkpoint): + n = registry.recover_from_checkpoint() + assert n == 0 + assert registry.get("proc_recycled") is None diff --git a/tools/process_registry.py b/tools/process_registry.py index 97cf89b3bda8..62ea225d4c41 100644 --- a/tools/process_registry.py +++ b/tools/process_registry.py @@ -1938,9 +1938,6 @@ def recover_from_checkpoint(self) -> int: pid_scope = entry.get("pid_scope", "host") if pid_scope != "host": - # Sandbox-backed processes keep only in-sandbox PIDs in the - # checkpoint, which are not meaningful to the restarted host - # process once the original environment handle is gone. logger.info( "Skipping recovery for non-host process: %s (pid=%s, scope=%s)", entry.get("command", "unknown")[:60], @@ -1949,6 +1946,19 @@ def recover_from_checkpoint(self) -> int: ) continue + # Idempotency guard: a session already tracked in the live + # registry must not be overwritten by recovery. This prevents + # double-adoption when recover_from_checkpoint is called twice + # (e.g. gateway-embedded CLI) and preserves the live entry's + # non-detached status (output pipe, reader thread, etc.). + sid = entry.get("session_id") + if sid and sid in self._running: + logger.debug( + "Skipping recovery for session %s: already tracked in live registry.", + sid, + ) + continue + # The PID must be alive AND still the same process we spawned. A # bare liveness check is unsafe: across a restart (especially a # reboot or long uptime) the kernel may have recycled this number @@ -2011,6 +2021,23 @@ def recover_from_checkpoint(self) -> int: return recovered + def recover_and_log(self) -> int: + """Recover detached processes from checkpoint and log the count. + + Convenience wrapper for startup paths (gateway and CLI) so the + "Recovered N detached background process(es)" message is emitted + consistently and the caller does not have to format the log line. + Best-effort: any exception is swallowed and logged at warning. + """ + try: + n = self.recover_from_checkpoint() + except Exception as exc: + logger.warning("Background-process recovery failed: %s", exc, exc_info=True) + return 0 + if n: + logger.info("Recovered %d detached background process(es) from checkpoint.", n) + return n + # Module-level singleton process_registry = ProcessRegistry() From 01ac88945fd846442c38f10eb281664b4d978a0d Mon Sep 17 00:00:00 2001 From: Stefan Verhey Date: Thu, 23 Jul 2026 08:19:55 +0200 Subject: [PATCH 2/2] refactor: expose public flush_checkpoint() on ProcessRegistry, use from cli.py --- cli.py | 2 +- tests/cli/test_cli_shutdown_checkpoint.py | 2 +- tools/process_registry.py | 9 +++++++++ 3 files changed, 11 insertions(+), 2 deletions(-) diff --git a/cli.py b/cli.py index 4eef20a6bc7a..9352c55c75cb 100644 --- a/cli.py +++ b/cli.py @@ -1225,7 +1225,7 @@ def _run_cleanup(*, notify_session_finalize: bool = True): # writes (e.g. a fast /quit right after a spawn). Best-effort. try: from tools.process_registry import process_registry - process_registry._write_checkpoint() + process_registry.flush_checkpoint() except Exception: pass # Shut down memory provider (on_session_end + shutdown_all) at actual diff --git a/tests/cli/test_cli_shutdown_checkpoint.py b/tests/cli/test_cli_shutdown_checkpoint.py index 9c089c46accf..7c18796439a2 100644 --- a/tests/cli/test_cli_shutdown_checkpoint.py +++ b/tests/cli/test_cli_shutdown_checkpoint.py @@ -6,7 +6,7 @@ def test_run_cleanup_writes_final_checkpoint(monkeypatch): writes = [] monkeypatch.setattr( - "tools.process_registry.process_registry._write_checkpoint", + "tools.process_registry.process_registry.flush_checkpoint", lambda: writes.append(1), ) # _run_cleanup touches many subsystems; stub the heavy ones so the test diff --git a/tools/process_registry.py b/tools/process_registry.py index 62ea225d4c41..e5f8fd991f9d 100644 --- a/tools/process_registry.py +++ b/tools/process_registry.py @@ -1916,6 +1916,15 @@ def _write_checkpoint(self): except Exception as e: logger.debug("Failed to write checkpoint file: %s", e, exc_info=True) + def flush_checkpoint(self): + """Flush running-process metadata to the checkpoint file. + + Public entry point for external callers (CLI shutdown, gateway + lifecycle) that need to persist the current registry state without + coupling to the private persistence implementation. + """ + self._write_checkpoint() + def recover_from_checkpoint(self) -> int: """ On gateway startup, probe PIDs from checkpoint file.