fix(cli): recover detached background processes from checkpoint on CLI startup and shutdown

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

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

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

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

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.
@@ -1938,9 +1947,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 +1955,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 +2030,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()