Claude Code: Refactor worker to persistent FastAPI background thread

- Move worker from separate process to FastAPI startup event thread
- Worker now runs as daemon thread within uvicorn process
- Removes unreliable --worker flag and entrypoint.sh worker startup
- Worker writes heartbeat every 5s via threading.Event
- More tightly integrated with app lifecycle

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

app.py

@@ -3,7 +3,6 @@ import logging
 import json
 import sys
 import time
-import signal
 import threading
 import psutil
 from pathlib import Path
@@ -29,12 +28,60 @@ logger = logging.getLogger(__name__)
 
 app = FastAPI(title="HuggingClaw Cain Space")
 
-# ========== A2A Ready State Polling ==========
-# Global state for polling
+# ========== Persistent Background Worker ==========
+# Global state for background tasks
+_worker_thread = None
+_worker_stop_event = None
 _polling_task = None
 _polling_lock = threading.Lock()
 
 
+def _run_worker_thread():
+    """
+    Worker thread that runs persistently within FastAPI process.
+    Writes heartbeat to keep agent alive and prevent IDLE state.
+    """
+    global _worker_stop_event
+    data_dir = Path(os.environ.get("OPENCLAW_DATA_DIR", "/data"))
+    status_file = data_dir / "cain_status.json"
+    heartbeat_path = Path("memory/worker_heartbeat.txt")
+    heartbeat_path.parent.mkdir(parents=True, exist_ok=True)
+
+    logger.info("[Cain Worker] Background thread started")
+
+    while not _worker_stop_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"] = "fastapi_background"
+                    data["worker_active"] = True
+                    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)
+            try:
+                with open(heartbeat_path, "w") as f:
+                    f.write(datetime.utcnow().isoformat() + "+00:00\n")
+            except Exception as e:
+                logger.warning(f"[Cain Worker] Failed to write heartbeat: {e}")
+
+            # Sleep until next heartbeat (5 seconds)
+            _worker_stop_event.wait(5)
+
+        except Exception as e:
+            logger.warning(f"[Cain Worker] Error in worker loop: {e}")
+            _worker_stop_event.wait(5)
+
+    logger.info("[Cain Worker] Background thread stopped")
+
+
 async def _poll_a2a_ready_state():
     """
     Background polling task to detect RUNNING_A2A_READY state and trigger agent execution.
@@ -77,16 +124,35 @@ async def _poll_a2a_ready_state():
 
 @app.on_event("startup")
 async def startup_event():
-    """Start background polling task on app startup."""
-    global _polling_task
+    """Start background worker thread and A2A polling task on app startup."""
+    global _worker_thread, _worker_stop_event, _polling_task
+
+    # Check WORKER_MODE before starting worker
+    worker_mode = os.environ.get("WORKER_MODE", "auto").lower()
+    if worker_mode == "disabled":
+        logger.info("[Cain] Worker disabled by WORKER_MODE=disabled")
+    else:
+        logger.info("[Cain] Starting persistent background worker thread...")
+        _worker_stop_event = threading.Event()
+        _worker_thread = threading.Thread(target=_run_worker_thread, daemon=True)
+        _worker_thread.start()
+        logger.info(f"[Cain] Worker thread started: {_worker_thread.name}")
+
     logger.info("[Cain] Starting A2A ready state polling task...")
     _polling_task = asyncio.create_task(_poll_a2a_ready_state())
 
 
 @app.on_event("shutdown")
 async def shutdown_event():
-    """Stop background polling task on app shutdown."""
-    global _polling_task
+    """Stop background worker thread and polling task on app shutdown."""
+    global _worker_thread, _worker_stop_event, _polling_task
+
+    if _worker_stop_event:
+        logger.info("[Cain] Stopping background worker thread...")
+        _worker_stop_event.set()
+        if _worker_thread:
+            _worker_thread.join(timeout=5)
+
     if _polling_task:
         logger.info("[Cain] Stopping A2A ready state polling task...")
         _polling_task.cancel()
@@ -96,7 +162,7 @@ async def shutdown_event():
             pass
 
 
-# ========== End A2A Polling ==========
+# ========== End Background Worker ==========
 
 
 # CRITICAL: Initialize clean status file on startup to prevent stale "unknown" errors
@@ -755,73 +821,6 @@ async def handle_401():
 """
 
 
-def run_worker():
-    """
-    Cain's worker process - runs in background to keep agent alive and process tasks.
-    This keeps Cain's state as 'active' rather than 'idle' when processing work.
-
-    Worker behavior controlled by WORKER_MODE environment variable:
-    - enabled/auto: Worker runs normally
-    - disabled: Worker exits immediately
-    """
-    # Check WORKER_MODE before starting worker
-    worker_mode = os.environ.get("WORKER_MODE", "auto").lower()
-    if worker_mode == "disabled":
-        logger.info("[Cain Worker] Worker disabled by WORKER_MODE=disabled - exiting")
-        return
-
-    logger.info(f"[Cain Worker] Starting worker process (mode: {worker_mode})...")
-    data_dir = Path(os.environ.get("OPENCLAW_DATA_DIR", "/data"))
-    status_file = data_dir / "cain_status.json"
-
-    # Heartbeat file for thread-alive verification (bypasses logging buffering)
-    heartbeat_path = Path("memory/worker_heartbeat.txt")
-    heartbeat_path.parent.mkdir(parents=True, exist_ok=True)
-
-    running = True
-
-    def handle_shutdown(signum, frame):
-        nonlocal running
-        logger.info(f"[Cain Worker] Received signal {signum}, shutting down...")
-        running = False
-
-    signal.signal(signal.SIGTERM, handle_shutdown)
-    signal.signal(signal.SIGINT, handle_shutdown)
-
-    while running:
-        try:
-            # Update worker heartbeat in status file
-            if status_file.exists():
-                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"] = worker_mode
-                with open(status_file, "w") as f:
-                    json.dump(data, f, indent=2)
-
-            # Write heartbeat timestamp to file (file-based truth check)
-            try:
-                with open(heartbeat_path, "w") as f:
-                    f.write(datetime.utcnow().isoformat() + "+00:00\n")
-            except Exception as e:
-                logger.warning(f"[Cain Worker] Failed to write heartbeat: {e}")
-
-            # Worker is alive - sleep and repeat
-            time.sleep(5)
-
-        except Exception as e:
-            logger.warning(f"[Cain Worker] Error in worker loop: {e}")
-            time.sleep(5)
-
-    logger.info("[Cain Worker] Worker process stopped")
-
-
 if __name__ == "__main__":
-    # Check for --worker flag
-    if "--worker" in sys.argv:
-        run_worker()
-        sys.exit(0)
-
     port = int(os.environ.get("PORT", 7860))
     uvicorn.run(app, host="0.0.0.0", port=port)

entrypoint.sh

@@ -132,27 +132,6 @@ else
     echo "  WARNING: a2a-proxy.cjs not found - A2A routing disabled"
 fi
 
-echo ""
-echo "=========================================="
-echo "Starting Cain worker..."
-echo "=========================================="
-
-# Check WORKER_MODE before starting worker
-# Default: auto (start worker)
-WORKER_MODE=${WORKER_MODE:-auto}
-if [ "$WORKER_MODE" = "disabled" ]; then
-    echo "  Worker disabled by WORKER_MODE=disabled - skipping worker startup"
-    WORKER_PID=""
-else
-    # Start Cain's worker process in background
-    # This keeps the agent alive and processes background tasks
-    echo "✓ Starting Cain worker process (mode: $WORKER_MODE)..."
-    python3 app.py --worker > /app/logs/worker.log 2>&1 &
-    WORKER_PID=$!
-    echo "  Worker PID: $WORKER_PID"
-    echo $WORKER_PID > /tmp/worker.pid
-fi
-
 echo ""
 echo "=========================================="
 echo "Starting uvicorn..."
@@ -171,7 +150,7 @@ echo "=========================================="
 echo "Verifying processes..."
 echo "=========================================="
 sleep 2
-ps aux | grep -E "(uvicorn|python3.*app.py.*worker|node.*a2a-proxy)" | grep -v grep || echo "  WARNING: Some processes may not have started"
+ps aux | grep -E "(uvicorn|node.*a2a-proxy)" | grep -v grep || echo "  WARNING: Some processes may not have started"
 
 echo ""
 echo "=========================================="
@@ -179,14 +158,10 @@ echo "Cain is running!"
 echo "=========================================="
 echo "API: http://0.0.0.0:${PORT:-7860}"
 echo "A2A Proxy: http://0.0.0.0:7861"
+echo "Worker: Integrated (FastAPI background thread)"
 echo ""
 echo "PIDs:"
 echo "  - Uvicorn: $UVICORN_PID"
-if [ -n "$WORKER_PID" ]; then
-    echo "  - Worker: $WORKER_PID"
-else
-    echo "  - Worker: disabled"
-fi
 echo "  - A2A Proxy: $A2A_PID"
 echo ""