Claude Code: Implement /health monitoring endpoint with process polling

- Add psutil import for process checking
- Add global health cache with 5s TTL to reduce overhead
- Implement _check_brain_process_health() function for PID polling
- Update /health endpoint to return status, timestamp, uptime_seconds, last_heartbeat
- Graceful degradation: endpoint works even if brain process is dead

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

app.py

@@ -4,6 +4,8 @@ import json
 import sys
 import time
 import signal
+import threading
+import psutil
 from pathlib import Path
 from datetime import datetime
 from fastapi import FastAPI, HTTPException, Request
@@ -102,36 +104,34 @@ async def root():
 
 @app.get("/health")
 async def health(request: Request):
+    """
+    Health monitoring endpoint for brain_minimal.py process.
+
+    Returns JSON with:
+    - status: "healthy" if brain_minimal.py is running, "unhealthy" otherwise
+    - timestamp: current UTC time
+    - uptime_seconds: process uptime duration
+    - last_heartbeat: timestamp of last successful heartbeat
+
+    Graceful degradation: works even if brain process is dead.
+    Results cached for 5s max to reduce overhead.
+    """
     # Honey Trap: Require User-Agent header to prevent browser fetches
     user_agent = request.headers.get("User-Agent")
     if not user_agent:
         return JSONResponse(status_code=401, content={})
 
-    # Force refresh status file timestamp to break stale "unknown" display
-    data_dir = Path(os.environ.get("OPENCLAW_DATA_DIR", "/data"))
-    data_dir.mkdir(parents=True, exist_ok=True)
+    # Get health status (uses cached result if within TTL)
+    health_data = _check_brain_process_health()
 
-    # Update all status files with fresh timestamp
-    for status_file in [
-        data_dir / "cain_status.json",
-        Path("/app/memory/cain_status.json"),
-    ]:
-        try:
-            if status_file.exists():
-                with open(status_file, "r") as f:
-                    data = json.load(f)
-                # Force fresh timestamp and health
-                data["last_updated"] = datetime.utcnow().isoformat() + "+00:00"
-                data["health"] = "HEALTHY"
-                data["error"] = None
-                data["error_display"] = "None"
-                data["_refreshed_at"] = f"health_endpoint_{datetime.utcnow().timestamp()}"
-                with open(status_file, "w") as f:
-                    json.dump(data, f, indent=2)
-        except Exception:
-            pass
-
-    return {"status": "healthy", "agent": "Cain"}
+    return {
+        "status": health_data["status"],
+        "timestamp": datetime.utcnow().isoformat() + "+00:00",
+        "uptime_seconds": health_data["uptime_seconds"],
+        "last_heartbeat": health_data["last_heartbeat"],
+        "agent": "Cain",
+        "brain_pid": health_data["brain_pid"]
+    }
 
 @app.get("/api/status")
 async def status():
@@ -213,6 +213,95 @@ _registered_agents = {}
 _agent_name_routing = {}
 
 
+# ========== Health Monitoring System ==========
+# Global health cache state (refresh every 5s max)
+_health_cache = {
+    "status": "unknown",
+    "timestamp": None,
+    "uptime_seconds": 0,
+    "last_heartbeat": None,
+    "brain_pid": None,
+    "brain_running": False,
+    "process_start_time": time.time()
+}
+_health_cache_lock = threading.Lock()
+_HEALTH_CACHE_TTL = 5.0  # Cache refresh interval in seconds
+
+
+def _check_brain_process_health() -> dict:
+    """
+    Check if brain_minimal.py process is running via PID polling.
+    Returns cached results if within TTL, otherwise performs fresh check.
+    """
+    global _health_cache
+
+    current_time = time.time()
+    cache_age = current_time - _health_cache["process_start_time"]
+
+    with _health_cache_lock:
+        # Return cached result if within TTL
+        if _health_cache["timestamp"] and (current_time - _health_cache["timestamp"]) < _HEALTH_CACHE_TTL:
+            return _health_cache.copy()
+
+    # Perform fresh health check
+    brain_running = False
+    brain_pid = None
+    last_heartbeat = None
+
+    try:
+        data_dir = Path(os.environ.get("OPENCLAW_DATA_DIR", "/data"))
+        status_file = data_dir / "cain_status.json"
+
+        # Try to get brain PID from status file
+        if status_file.exists():
+            try:
+                with open(status_file, "r") as f:
+                    status_data = json.load(f)
+                    brain_pid = status_data.get("_worker_pid")
+                    last_heartbeat = status_data.get("worker_heartbeat")
+
+                    # Check if process with PID is running
+                    if brain_pid:
+                        if psutil.pid_exists(brain_pid):
+                            try:
+                                proc = psutil.Process(brain_pid)
+                                # Verify it's actually a Python process
+                                if "python" in proc.name().lower():
+                                    brain_running = True
+                            except (psutil.NoSuchProcess, psutil.AccessDenied):
+                                brain_running = False
+            except (json.JSONDecodeError, IOError):
+                pass
+
+        # Fallback: Check for any python process with brain_minimal in cmdline
+        if not brain_running:
+            for proc in psutil.process_iter(['pid', 'name', 'cmdline']):
+                try:
+                    cmdline = proc.info['cmdline']
+                    if cmdline and any('brain_minimal.py' in arg for arg in cmdline):
+                        brain_running = True
+                        brain_pid = proc.info['pid']
+                        break
+                except (psutil.NoSuchProcess, psutil.AccessDenied):
+                    continue
+
+    except Exception as e:
+        logger.warning(f"Health check error: {e}")
+
+    # Update cache
+    with _health_cache_lock:
+        _health_cache.update({
+            "status": "healthy" if brain_running else "unhealthy",
+            "timestamp": current_time,
+            "uptime_seconds": cache_age,
+            "last_heartbeat": last_heartbeat,
+            "brain_pid": brain_pid,
+            "brain_running": brain_running
+        })
+
+    return _health_cache.copy()
+
+
 @app.post("/join-agent")
 async def join_agent(request_data: dict):
     """Register a new agent with the office hub."""