Harden engine status during active watches

.gitignore

@@ -7,7 +7,9 @@ data/*.sqlite
 data/*.sqlite-shm
 data/*.sqlite-wal
 data/app_visual_state.*
+data/supervisor.pid
 data/raw/
+reports/ACTIVE_WATCH.json
 *.log
 macos/BraiinsRatchet/.build/
 macos/build/

src/braiins_ratchet/engine.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 from dataclasses import dataclass
+import json
 import os
 from pathlib import Path
 import signal
@@ -8,6 +9,7 @@ import subprocess
 import sys
 
 from .config import REPO_ROOT
+from .experiments import ACTIVE_WATCH
 from .storage import DATA_DIR
 
 
@@ -45,6 +47,17 @@ def get_engine_status() -> EngineStatus:
             log_path=str(SUPERVISOR_LOG.relative_to(REPO_ROOT)),
         )
 
+    active_watch_pid = _active_watch_pid()
+    if active_watch_pid is not None and _pid_exists(active_watch_pid):
+        SUPERVISOR_PID.parent.mkdir(parents=True, exist_ok=True)
+        SUPERVISOR_PID.write_text(str(active_watch_pid), encoding="utf-8")
+        return EngineStatus(
+            running=True,
+            pid=active_watch_pid,
+            detail=f"forever monitor watch is running as pid {active_watch_pid}",
+            log_path=str(SUPERVISOR_LOG.relative_to(REPO_ROOT)),
+        )
+
     if pid is not None:
         _clear_pid_file()
     return EngineStatus(
@@ -150,7 +163,19 @@ def _clear_pid_file() -> None:
 
 def _pid_matches_supervisor(pid: int) -> bool:
     command = _command_for_pid(pid)
-    return command is not None and _is_supervisor_command(command)
+    if command is None:
+        return _pid_exists(pid)
+    return _is_supervisor_command(command)
+
+
+def _pid_exists(pid: int) -> bool:
+    try:
+        os.kill(pid, 0)
+    except ProcessLookupError:
+        return False
+    except PermissionError:
+        return True
+    return True
 
 
 def _command_for_pid(pid: int) -> str | None:
@@ -193,6 +218,15 @@ def _find_supervisor_pid() -> int | None:
     return None
 
 
+def _active_watch_pid() -> int | None:
+    try:
+        payload = json.loads(ACTIVE_WATCH.read_text(encoding="utf-8"))
+    except (FileNotFoundError, json.JSONDecodeError):
+        return None
+    pid = payload.get("pid")
+    return pid if isinstance(pid, int) else None
+
+
 def _is_supervisor_command(command: str) -> bool:
     return (
         "braiins_ratchet.cli supervise" in command

tests/test_engine.py

@@ -39,6 +39,21 @@ class EngineStatusTests(unittest.TestCase):
             self.assertEqual(status.pid, 12345)
             self.assertEqual(paths["pid"].read_text(encoding="utf-8"), "12345")
 
+    def test_engine_status_uses_active_watch_when_process_table_is_unavailable(self) -> None:
+        with _isolated_engine_paths() as paths:
+            paths["active_watch"].parent.mkdir(parents=True, exist_ok=True)
+            paths["active_watch"].write_text('{"pid": 456, "run_id": "run-example"}', encoding="utf-8")
+            with (
+                patch.object(engine, "_find_supervisor_pid", return_value=None),
+                patch.object(engine, "_pid_exists", return_value=True),
+            ):
+                status = engine.get_engine_status()
+
+            self.assertTrue(status.running)
+            self.assertEqual(status.pid, 456)
+            self.assertIn("watch is running", status.detail)
+            self.assertEqual(paths["pid"].read_text(encoding="utf-8"), "456")
+
     def test_render_engine_status_is_noob_readable(self) -> None:
         text = engine.render_engine_status(
             engine.EngineStatus(
@@ -85,6 +100,7 @@ class _isolated_engine_paths:
             "logs": root / "logs",
             "pid": root / "data" / "supervisor.pid",
             "log": root / "logs" / "supervisor.log",
+            "active_watch": root / "reports" / "ACTIVE_WATCH.json",
         }
         self.patcher = patch.multiple(
             engine,
@@ -93,6 +109,7 @@ class _isolated_engine_paths:
             LOG_DIR=self.paths["logs"],
             SUPERVISOR_PID=self.paths["pid"],
             SUPERVISOR_LOG=self.paths["log"],
+            ACTIVE_WATCH=self.paths["active_watch"],
         )
         self.patcher.start()
         return self.paths