Refactor heartbeat from File I/O to In-Memory IPC

- Add shared_state.py: Thread-safe in-memory worker state cache
- brain_minimal.py: Replace file writes with HTTP POST to /internal/heartbeat
- app.py: Add /internal/heartbeat endpoint, read from shared state cache
- Decouple worker_status from PID check - now reflects activity level
- Eliminate Read-After-Write race conditions

This moves from File I/O to In-Memory/IPC for heartbeat monitoring,
removing the file-based race conditions diagnosed earlier.

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

app.py

@@ -14,6 +14,9 @@ from fastapi.staticfiles import StaticFiles
 import uvicorn
 import asyncio
 
+# Import shared state module for in-memory IPC with worker
+from shared_state import get_worker_state
+
 # CRITICAL: Initialize .env before any imports that might load environment variables
 # This ensures .env exists even if app.py is run directly (without entrypoint.sh)
 try:
@@ -189,7 +192,7 @@ class ProcessWatchdog:
         Check if the managed process is alive and actively running.
 
         Uses multiple checks in order:
-        1. File-based heartbeat (primary truth source - written by worker)
+        1. In-memory state cache via shared_state (primary - updated by worker HTTP POST)
         2. Subprocess polling (detects crashes)
         3. PID existence verification (detects ghost PIDs)
         4. Process name verification (ensures it's our python process)
@@ -197,24 +200,15 @@ class ProcessWatchdog:
         if self.process is None or self.pid is None:
             return False
 
-        # PRIMARY CHECK: File-based heartbeat (source of truth for worker activity)
-        heartbeat_path = Path("/app/logs/worker_heartbeat.txt")
-        if heartbeat_path.exists():
-            try:
-                with open(heartbeat_path, "r") as f:
-                    heartbeat_content = f.read().strip()
-                if heartbeat_content:
-                    heartbeat_time = datetime.fromisoformat(heartbeat_content.replace("+00:00", "").replace("Z", ""))
-                    # Make timezone-naive for comparison
-                    if heartbeat_time.tzinfo is not None:
-                        heartbeat_time = heartbeat_time.replace(tzinfo=None)
-                    heartbeat_age = (datetime.utcnow() - heartbeat_time).total_seconds()
-                    # Heartbeat is fresh if updated within 15 seconds
-                    if heartbeat_age < 15:
-                        # Fresh heartbeat means worker is actively running
-                        return True
-            except Exception as e:
-                self._log("warning", "Heartbeat file check failed", error=str(e))
+        # PRIMARY CHECK: In-memory state cache (IPC from worker via HTTP POST)
+        # This is the source of truth for worker activity
+        try:
+            worker_mem_state = get_worker_state()
+            if worker_mem_state.is_healthy(max_age_seconds=15):
+                # Fresh in-memory heartbeat means worker is actively running
+                return True
+        except Exception as e:
+            self._log("warning", "In-memory state check failed", error=str(e))
 
         # SECONDARY CHECK: Try polling the subprocess
         try:
@@ -249,8 +243,8 @@ class ProcessWatchdog:
 
         # All checks passed (but heartbeat was stale)
         # Log this case - process exists but heartbeat is stale
-        self._log("info", "Process exists but heartbeat stale", pid=self.pid,
-                  note="Heartbeat file exists but is >15s old")
+        self._log("info", "Process exists but in-memory heartbeat stale", pid=self.pid,
+                  note="Heartbeat in cache is >15s old")
         return False
 
     def _update_status_file(self):
@@ -467,6 +461,31 @@ async def health(request: Request):
     }
 
 
+@app.post("/internal/heartbeat")
+async def internal_heartbeat(request_data: dict):
+    """
+    Internal endpoint for worker process (brain_minimal.py) to report heartbeat.
+    Updates in-memory state cache - NO FILE WRITES.
+    """
+    try:
+        worker_state = get_worker_state()
+
+        # Update in-memory state from worker heartbeat
+        worker_state.update(
+            worker_pid=request_data.get("worker_pid"),
+            worker_mode=request_data.get("worker_mode"),
+            worker_state=request_data.get("worker_state", "active"),  # activity level
+            current_state=request_data.get("current_state", "idle"),
+            worker_active=request_data.get("worker_active", True),
+            stage=request_data.get("stage", "RUNNING_A2A_READY"),
+        )
+
+        return {"status": "ok", "received": True}
+    except Exception as e:
+        logger.warning(f"[HEARTBEAT] Failed to update worker state: {e}")
+        return {"status": "error", "error": str(e)}
+
+
 @app.get("/api/status")
 async def status():
     """
@@ -500,7 +519,12 @@ async def status():
 
 @app.get("/api/state")
 async def api_state(request: Request):
-    """Return Cain's current state for Office/A2A polling."""
+    """
+    Return Cain's current state for Office/A2A polling.
+
+    NOW USES IN-MEMORY STATE CACHE (no file I/O for heartbeat).
+    Worker state reflects activity level, not just PID existence.
+    """
     user_agent = request.headers.get("User-Agent")
     if not user_agent:
         return JSONResponse(status_code=401, content={})
@@ -513,75 +537,32 @@ async def api_state(request: Request):
 
     watchdog_status = _watchdog.get_status() if _watchdog else {"is_alive": False, "pid": None}
 
-    # PRIMARY HEARTBEAT CHECK: Read file-based heartbeat (written directly by worker process)
-    # This is the source of truth for detecting a live worker process
-    heartbeat_path = Path("/app/logs/worker_heartbeat.txt")
-    file_heartbeat_time = None
-    file_heartbeat_age = None
-    heartbeat_fresh = False
+    # PRIMARY: Use in-memory state cache (IPC from worker via HTTP POST)
+    worker_mem_state = get_worker_state()
+    mem_state_data = worker_mem_state.get()
 
-    if heartbeat_path.exists():
-        try:
-            with open(heartbeat_path, "r") as f:
-                content = f.read().strip()
-            if content:
-                # Parse ISO timestamp
-                file_heartbeat_time = datetime.fromisoformat(content.replace("+00:00", "").replace("Z", "+00:00"))
-                if file_heartbeat_time.tzinfo is not None:
-                    file_heartbeat_time = file_heartbeat_time.replace(tzinfo=None)
-                file_heartbeat_age = (datetime.utcnow() - file_heartbeat_time).total_seconds()
-                # Heartbeat is fresh if updated within 15 seconds
-                heartbeat_fresh = file_heartbeat_age < 15
-        except Exception as e:
-            logger.warning(f"[STATE] Failed to read heartbeat file: {e}")
-
-    # FALLBACK: Check heartbeat in JSON file (secondary source)
-    json_heartbeat_str = status_data.get("worker_heartbeat")
-    json_heartbeat_age = None
-    if json_heartbeat_str and not heartbeat_fresh:
-        try:
-            json_heartbeat_time = datetime.fromisoformat(json_heartbeat_str.replace("+00:00", "").replace("Z", "+00:00"))
-            if json_heartbeat_time.endswith("+00:00"):
-                json_heartbeat_time = json_heartbeat_time.replace(tzinfo=None)
-            json_heartbeat_age = (datetime.utcnow() - json_heartbeat_time.replace(tzinfo=None)).total_seconds()
-            # Use JSON heartbeat if it's fresh (even if file check failed)
-            if json_heartbeat_age < 15:
-                heartbeat_fresh = True
-                file_heartbeat_age = json_heartbeat_age
-        except Exception as e:
-            logger.warning(f"[STATE] Failed to parse JSON heartbeat: {e}")
-
-    # DETAILED DIAGNOSTICS: Log why state is determined as unhealthy
-    failure_reason = None
-    if not heartbeat_fresh:
-        if file_heartbeat_age is not None:
-            failure_reason = f"File heartbeat stale: {file_heartbeat_age:.1f}s old (threshold: 15s)"
-        elif json_heartbeat_age is not None:
-            failure_reason = f"JSON heartbeat stale: {json_heartbeat_age:.1f}s old"
-        else:
-            failure_reason = "No heartbeat found (file or JSON)"
+    # Heartbeat freshness from in-memory state
+    heartbeat_fresh = worker_mem_state.is_healthy(max_age_seconds=15)
+    heartbeat_age = mem_state_data.get("heartbeat_age_seconds", 0)
 
-        # Log with full diagnostic info
-        logger.warning(f"[STATE] Heartbeat check failed: {failure_reason} | PID={watchdog_status.get('pid')} | File exists={heartbeat_path.exists()}")
+    # Worker activity state (decoupled from PID check)
+    # This reflects actual worker activity, not just process existence
+    worker_activity_state = worker_mem_state.get_worker_state()
 
-    # Determine actual state based on watchdog health and heartbeat freshness
-    worker_state = status_data.get("_worker_state", "")
-    current_state = status_data.get("current_state", "idle")
+    # Current processing state (idle/processing)
+    current_state = mem_state_data.get("current_state", "idle")
 
-    # If watchdog says process is dead OR heartbeat is stale, force state to idle
+    # Determine reported state
     if not watchdog_status["is_alive"] or not heartbeat_fresh:
         reported_state = "idle"
-    elif worker_state:
-        # Only trust _worker_state if heartbeat is fresh and process is alive
-        reported_state = worker_state
     else:
-        reported_state = current_state
+        reported_state = worker_activity_state
 
-    # Build detail message with diagnostic info
+    # Build detail message
     if watchdog_status["is_alive"] and heartbeat_fresh:
         detail = "Cain is operational"
-    elif failure_reason:
-        detail = f"Brain process issue: {failure_reason}"
+    elif not heartbeat_fresh:
+        detail = f"Worker heartbeat stale: {heartbeat_age:.1f}s old (threshold: 15s)"
     else:
         detail = "Brain process not running or stale"
 
@@ -589,17 +570,18 @@ async def api_state(request: Request):
         "state": reported_state,
         "detail": detail,
         "updated_at": datetime.utcnow().isoformat() + "+00:00",
-        "stage": status_data.get("stage", "RUNNING"),
+        "stage": mem_state_data.get("stage", "RUNNING_A2A_READY"),
         "health": "HEALTHY" if watchdog_status["is_alive"] and heartbeat_fresh else "UNHEALTHY",
         "is_alive": watchdog_status["is_alive"],
         "brain_pid": watchdog_status.get("pid"),
         "heartbeat_fresh": heartbeat_fresh,
-        "heartbeat_age_seconds": round(file_heartbeat_age, 2) if file_heartbeat_age is not None else None,
+        "heartbeat_age_seconds": round(heartbeat_age, 2),
+        "worker_activity_state": worker_activity_state,  # NEW: activity level
         "_diagnostics": {
-            "heartbeat_file_exists": heartbeat_path.exists(),
-            "failure_reason": failure_reason,
-            "file_heartbeat_age": round(file_heartbeat_age, 2) if file_heartbeat_age is not None else None,
-            "json_heartbeat_age": round(json_heartbeat_age, 2) if json_heartbeat_age is not None else None
+            "heartbeat_source": "in_memory_ipc",
+            "worker_pid": mem_state_data.get("worker_pid"),
+            "worker_mode": mem_state_data.get("worker_mode"),
+            "last_heartbeat": mem_state_data.get("last_heartbeat")
         }
     }
 

brain_minimal.py

@@ -18,6 +18,8 @@ import signal
 import time
 import logging
 import asyncio
+import urllib.request
+import urllib.error
 from pathlib import Path
 from datetime import datetime
 from threading import Thread, Event
@@ -210,47 +212,80 @@ async def _poll_a2a_ready_state():
     logger.info("[WORKER] A2A polling loop stopped")
 
 
+def _send_heartbeat_to_app(worker_state: str = "active", current_state: str = "idle"):
+    """
+    Send heartbeat to app.py via HTTP POST (in-memory IPC).
+    This replaces file-based heartbeat writes entirely.
+    """
+    try:
+        port = os.environ.get("PORT", "7860")
+        url = f"http://127.0.0.1:{port}/internal/heartbeat"
+
+        payload = {
+            "worker_pid": os.getpid(),
+            "worker_mode": "standalone_process",
+            "worker_state": worker_state,  # activity state, not PID-based
+            "current_state": current_state,
+            "worker_active": True,
+            "stage": "RUNNING_A2A_READY",
+            "timestamp": datetime.utcnow().isoformat() + "+00:00"
+        }
+
+        data = json.dumps(payload).encode("utf-8")
+        req = urllib.request.Request(
+            url,
+            data=data,
+            headers={"Content-Type": "application/json"},
+            method="POST"
+        )
+
+        # Set short timeout to avoid blocking worker loop
+        with urllib.request.urlopen(req, timeout=2) as resp:
+            if resp.status != 200:
+                logger.warning(f"[WORKER] Heartbeat POST returned {resp.status}")
+
+    except urllib.error.HTTPError as e:
+        if e.code != 404:
+            logger.debug(f"[WORKER] Heartbeat POST HTTP error: {e.code}")
+    except urllib.error.URLError:
+        # App may not be ready yet - silent retry
+        pass
+    except Exception as e:
+        logger.debug(f"[WORKER] Heartbeat POST failed: {e}")
+
+
 def worker_heartbeat_loop():
     """
-    Write heartbeat to file for health monitoring.
-    This is the main heartbeat that WorkerManager checks.
+    Send heartbeat to app.py via HTTP POST (in-memory IPC).
+    NO FILE WRITES - eliminates race conditions.
     """
-    logger.info("[WORKER] Heartbeat loop started")
+    logger.info("[WORKER] Heartbeat loop started (in-memory IPC mode)")
 
-    data_dir = Path(os.environ.get("OPENCLAW_DATA_DIR", "/data"))
-    status_file = data_dir / "cain_status.json"
+    # Optional: Keep minimal file writes for backward compatibility/debugging
+    # But PRIMARY heartbeat is now HTTP-based
     heartbeat_path = Path("/app/logs/worker_heartbeat.txt")
     heartbeat_path.parent.mkdir(parents=True, exist_ok=True)
 
+    # Write initial timestamp for debugging
+    try:
+        with open(heartbeat_path, "w") as f:
+            f.write(datetime.utcnow().isoformat() + "+00:00\n")
+    except Exception:
+        pass
+
     while not _shutdown_event.is_set():
         try:
-            # Update worker heartbeat in status file
-            if status_file.exists():
-                try:
-                    with open(status_file, "r") as f:
-                        data = json.load(f)
-                    data["worker_heartbeat"] = datetime.utcnow().isoformat() + "+00:00"
-                    data["_worker_pid"] = os.getpid()
-                    data["_worker_mode"] = "standalone_process"
-                    data["worker_active"] = True
-                    data["stage"] = "RUNNING_A2A_READY"
-                    # CRITICAL: Use separate key for heartbeat activity to avoid race condition
-                    # The polling loop needs current_state to stay "idle" for trigger detection
-                    data["_worker_state"] = "active"  # Worker is actively running and polling
-                    # Only update current_state if it doesn't exist (init) or is processing-complete
-                    if data.get("current_state") not in ("idle", "processing"):
-                        data["current_state"] = "idle"
-                    with open(status_file, "w") as f:
-                        json.dump(data, f, indent=2)
-                except (json.JSONDecodeError, IOError):
-                    pass
-
-            # Write heartbeat timestamp to file (file-based truth check)
+            # PRIMARY: Send heartbeat via HTTP to app.py (in-memory state)
+            # Worker state reflects activity level, not just PID existence
+            _send_heartbeat_to_app(worker_state="active", current_state="idle")
+
+            # SECONDARY: Optional fallback file write (for debugging only)
+            # Not used for health checks - just for manual inspection
             try:
                 with open(heartbeat_path, "w") as f:
                     f.write(datetime.utcnow().isoformat() + "+00:00\n")
-            except Exception as e:
-                logger.warning(f"[WORKER] Failed to write heartbeat: {e}")
+            except Exception:
+                pass
 
             # Sleep until next heartbeat (5 seconds)
             if _shutdown_event.wait(5):
@@ -275,10 +310,8 @@ def main():
     # Ensure logs directory exists
     Path("/app/logs").mkdir(parents=True, exist_ok=True)
 
-    # Write initial heartbeat
-    heartbeat_path = Path("/app/logs/worker_heartbeat.txt")
-    with open(heartbeat_path, "w") as f:
-        f.write(datetime.utcnow().isoformat() + "+00:00\n")
+    # Initial heartbeat via HTTP (in-memory IPC)
+    _send_heartbeat_to_app(worker_state="active", current_state="idle")
 
     # Start heartbeat loop in thread
     heartbeat_thread = Thread(target=worker_heartbeat_loop, daemon=True)

shared_state.py

@@ -0,0 +1,120 @@
+#!/usr/bin/env python3
+"""
+Shared in-memory state for Cain worker process.
+
+This module provides a thread-safe in-memory cache for worker state
+that replaces file-based heartbeat writes. The worker process (brain_minimal.py)
+updates this state via HTTP POST to app.py's internal endpoint.
+
+Key design:
+- No file I/O for heartbeat (eliminates race conditions)
+- In-memory state cache in app.py
+- Worker state reflects activity level, not just PID existence
+"""
+import threading
+import time
+from datetime import datetime
+from typing import Any, Optional
+
+
+class WorkerState:
+    """
+    Thread-safe in-memory cache for worker state.
+
+    This is the source of truth for worker activity status.
+    Updated by brain_minimal.py via HTTP POST to /internal/heartbeat.
+    """
+
+    def __init__(self):
+        self._lock = threading.Lock()
+        self._state = {
+            # Worker activity state (decoupled from PID)
+            "worker_state": "idle",  # idle, active, processing
+            "worker_active": False,
+            "current_state": "idle",  # backward compatibility
+
+            # Process info (for logging/debugging)
+            "worker_pid": None,
+            "worker_mode": None,
+
+            # Heartbeat timestamps
+            "last_heartbeat": None,
+            "heartbeat_age_seconds": 0,
+
+            # Stage info
+            "stage": "RUNNING_A2A_READY",
+
+            # Health status
+            "health": "HEALTHY",
+            "error": None,
+        }
+
+    def update(self, **kwargs):
+        """Update worker state with new values (thread-safe)."""
+        with self._lock:
+            # Update heartbeat timestamp
+            self._state["last_heartbeat"] = datetime.utcnow().isoformat() + "+00:00"
+            self._state["heartbeat_age_seconds"] = 0
+
+            # Update provided fields
+            for key, value in kwargs.items():
+                if value is not None:
+                    self._state[key] = value
+
+    def get(self) -> dict[str, Any]:
+        """Get current state snapshot (thread-safe)."""
+        with self._lock:
+            # Calculate heartbeat age
+            if self._state["last_heartbeat"]:
+                try:
+                    heartbeat_time = datetime.fromisoformat(
+                        self._state["last_heartbeat"].replace("+00:00", "").replace("Z", "")
+                    )
+                    if heartbeat_time.tzinfo is not None:
+                        heartbeat_time = heartbeat_time.replace(tzinfo=None)
+                    age = (datetime.utcnow() - heartbeat_time).total_seconds()
+                    self._state["heartbeat_age_seconds"] = age
+                except Exception:
+                    self._state["heartbeat_age_seconds"] = 999
+
+            return self._state.copy()
+
+    def is_healthy(self, max_age_seconds: int = 15) -> bool:
+        """
+        Check if worker is healthy based on heartbeat age.
+
+        Args:
+            max_age_seconds: Maximum acceptable heartbeat age (default 15s)
+
+        Returns:
+            True if heartbeat is fresh, False otherwise
+        """
+        with self._lock:
+            return self._state["heartbeat_age_seconds"] < max_age_seconds
+
+    def get_worker_state(self) -> str:
+        """
+        Get worker activity state (decoupled from PID).
+
+        Returns:
+            "idle", "active", or "processing" based on actual activity
+        """
+        with self._lock:
+            # If heartbeat is stale, worker is not active
+            if not self.is_healthy():
+                return "idle"
+
+            # Return the worker activity state (not PID-based)
+            return self._state.get("worker_state", "idle")
+
+
+# Global singleton
+_worker_state: Optional[WorkerState] = None
+
+
+def get_worker_state() -> WorkerState:
+    """Get or create the global worker state singleton."""
+    global _worker_state
+    if _worker_state is None:
+        _worker_state = WorkerState()
+    return _worker_state