#!/usr/bin/env python3 """ Standalone Worker Process for HuggingClaw Cain. This is the main worker process that runs independently of the FastAPI server. It can be spawned as a subprocess and managed by WorkerManager in app.py. Features: - Graceful shutdown on SIGTERM - Heartbeat file for health monitoring - Persistent polling loop - Defensive .env initialization to prevent startup crashes """ import os import sys import json 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 # CRITICAL: Defensive .env initialization before any other imports # This prevents FileNotFoundError if brain_minimal.py runs before entrypoint.sh _fallback_mode = False _init_success = False # Ensure /app is in path for utils import if "/app" not in sys.path: sys.path.insert(0, "/app") # Try to import and run env_init try: from utils.env_init import init_env _init_success = init_env() if not _init_success: logging.basicConfig(level=logging.WARNING) logging.warning("[BRAIN_MINIMAL] env_init returned False - using emergency defaults") _fallback_mode = True except ImportError as e: # If utils.env_init is not available, use emergency fallback logging.basicConfig(level=logging.WARNING) logging.error(f"[BRAIN_MINIMAL] Failed to import env_init: {e}") logging.warning("[BRAIN_MINIMAL] Using emergency .env fallback (manual initialization)") _fallback_mode = True # Manual fallback: create .env if missing env_file = Path("/app/.env") if not env_file.exists(): env_example = Path("/app/.env.example") if env_example.exists(): import shutil shutil.copy(env_example, env_file) logging.warning("[BRAIN_MINIMAL] .env not found - copied from .env.example (emergency fallback)") else: # Create .env with critical defaults with open(env_file, "w") as f: f.write("# Emergency fallback .env\n") f.write("OPENCLAW_DATA_DIR=/data\n") f.write("PORT=7860\n") f.write("WORKER_MODE=auto\n") logging.warning("[BRAIN_MINIMAL] .env.example not found - created .env with safe defaults (emergency fallback)") except Exception as e: # Don't crash on .env init errors - log warning and continue logging.basicConfig(level=logging.WARNING) logging.error(f"[BRAIN_MINIMAL] .env initialization failed: {type(e).__name__}: {e}") logging.warning("[BRAIN_MINIMAL] Using emergency safe defaults") _fallback_mode = True # Set emergency defaults directly os.environ.setdefault("OPENCLAW_DATA_DIR", "/data") os.environ.setdefault("PORT", "7860") os.environ.setdefault("WORKER_MODE", "auto") # Configure logging logging.basicConfig( level=logging.INFO, format='[%(asctime)s] %(levelname)s %(message)s', datefmt='%Y-%m-%dT%H:%M:%SZ' ) logger = logging.getLogger(__name__) # Log initialization status if _init_success: logger.info("[BRAIN_MINIMAL] Environment initialized successfully via env_init") elif _fallback_mode: logger.warning("[BRAIN_MINIMAL] FALLBACK MODE ACTIVE - .env handling used emergency defaults") else: logger.warning("[BRAIN_MINIMAL] Environment status unknown - proceeding with caution") # CRITICAL: Log runtime environment for debugging logger.info("[BRAIN_MINIMAL] ========== RUNTIME ENVIRONMENT DUMP ==========") logger.info(f"[BRAIN_MINIMAL] Working Directory: {os.getcwd()}") logger.info(f"[BRAIN_MINIMAL] Python Path: {sys.path[:3]}...") # First 3 entries logger.info(f"[BRAIN_MINIMAL] OPENCLAW_DATA_DIR: {os.environ.get('OPENCLAW_DATA_DIR', 'NOT SET')}") logger.info(f"[BRAIN_MINIMAL] WORKER_MODE: {os.environ.get('WORKER_MODE', 'NOT SET')}") logger.info(f"[BRAIN_MINIMAL] AGENT_MODE: {os.environ.get('AGENT_MODE', 'NOT SET')}") logger.info(f"[BRAIN_MINIMAL] SLEEP_INTERVAL: {os.environ.get('SLEEP_INTERVAL', 'NOT SET')}") logger.info(f"[BRAIN_MINIMAL] WORKER_START_TIMEOUT: {os.environ.get('WORKER_START_TIMEOUT', 'NOT SET')}") logger.info(f"[BRAIN_MINIMAL] OPENAI_API_KEY: {'SET' if os.environ.get('OPENAI_API_KEY') else 'NOT SET'}") logger.info(f"[BRAIN_MINIMAL] OPENROUTER_API_KEY: {'SET' if os.environ.get('OPENROUTER_API_KEY') else 'NOT SET'}") logger.info(f"[BRAIN_MINIMAL] HF_TOKEN: {'SET' if os.environ.get('HF_TOKEN') else 'NOT SET'}") logger.info(f"[BRAIN_MINIMAL] OPENCLAW_DATASET_REPO: {os.environ.get('OPENCLAW_DATASET_REPO', 'NOT SET')}") logger.info("[BRAIN_MINIMAL] ========== END ENVIRONMENT DUMP ==========") # Global shutdown flag _shutdown_event = Event() def signal_handler(signum, frame): """Handle SIGTERM for graceful shutdown.""" logger.info(f"[WORKER] Received signal {signum}, initiating graceful shutdown...") _shutdown_event.set() # Register signal handlers signal.signal(signal.SIGTERM, signal_handler) signal.signal(signal.SIGINT, signal_handler) def run_asyncio_polling(): """ Run the asyncio polling task in its own event loop within a daemon thread. This ensures the polling loop starts immediately at worker start. """ logger.info("[WORKER] Starting asyncio polling loop...") # Create new event loop for this thread loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) try: # Run the polling task - this blocks forever in while True loop.run_until_complete(_poll_a2a_ready_state()) except Exception as e: logger.error(f"[WORKER] Polling loop failed: {e}") finally: loop.close() async def _poll_a2a_ready_state(): """ Background polling task to detect RUNNING_A2A_READY state and trigger agent execution. This bridges the synchronization deadlock where the worker ignores push events. """ logger.info("[WORKER] A2A polling loop started") data_dir = Path(os.environ.get("OPENCLAW_DATA_DIR", "/data")) status_file = data_dir / "cain_status.json" # Get SLEEP_INTERVAL from environment, default to 5 seconds sleep_interval = int(os.environ.get("SLEEP_INTERVAL", "5")) logger.info(f"[WORKER] Polling interval: {sleep_interval} seconds (from SLEEP_INTERVAL)") iteration = 0 while not _shutdown_event.is_set(): try: # Read current status if status_file.exists(): with open(status_file, "r") as f: status_data = json.load(f) current_stage = status_data.get("stage", "") current_state = status_data.get("current_state", "idle") # Check if we're in the A2A ready state but idle (deadlock condition) if current_stage == "RUNNING_A2A_READY" and current_state == "idle": logger.info("[WORKER] Detected RUNNING_A2A_READY + idle - triggering agent execution") # Update state to processing before triggering try: status_data["current_state"] = "processing" status_data["last_triggered"] = datetime.utcnow().isoformat() + "+00:00" with open(status_file, "w") as f: json.dump(status_data, f, indent=2) except Exception as e: logger.warning(f"[WORKER] Failed to update state to processing: {e}") # Trigger agent execution by processing a dummy message try: from error_handlers import handle_brain_response response = handle_brain_response("") logger.info(f"[WORKER] Agent execution completed: {response[:100] if response else 'No response'}") except Exception as e: logger.warning(f"[WORKER] Failed to trigger agent: {e}") # Reset state back to idle after triggering (ready for next trigger) try: with open(status_file, "r") as f: status_data = json.load(f) status_data["current_state"] = "idle" with open(status_file, "w") as f: json.dump(status_data, f, indent=2) except Exception as e: logger.warning(f"[WORKER] Failed to reset state to idle: {e}") iteration += 1 log_interval = max(1, 60 // sleep_interval) # Log approximately every minute if iteration % log_interval == 0: logger.info(f"[WORKER] Polling iteration {iteration}") await asyncio.sleep(sleep_interval) except Exception as e: logger.warning(f"[WORKER] Error in polling loop: {e}") await asyncio.sleep(1) 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}") else: logger.info(f"[WORKER] Heartbeat sent successfully to {url}") except urllib.error.HTTPError as e: if e.code != 404: logger.debug(f"[WORKER] Heartbeat POST HTTP error: {e.code}") except urllib.error.URLError as e: # App may not be ready yet - log for visibility but don't crash logger.debug(f"[WORKER] Heartbeat POST URLError: {e}") except Exception as e: logger.warning(f"[WORKER] Heartbeat POST failed: {type(e).__name__}: {e}") def worker_heartbeat_loop(): """ Send heartbeat to app.py via HTTP POST (in-memory IPC). NO FILE WRITES - eliminates race conditions. """ logger.info("[WORKER] Heartbeat loop started (in-memory IPC mode)") # 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: # 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: pass # Sleep until next heartbeat (5 seconds) if _shutdown_event.wait(5): break except Exception as e: logger.warning(f"[WORKER] Error in heartbeat loop: {e}") _shutdown_event.wait(5) logger.info("[WORKER] Heartbeat loop stopped") def main(): """Main worker entry point.""" # CRITICAL: Flush immediately to prevent buffer deadlock sys.stdout.flush() sys.stderr.flush() logger.info("=" * 60) logger.info("[WORKER] ========== CAIN WORKER PROCESS STARTED ==========") logger.info(f"[WORKER] Timestamp: {datetime.utcnow().isoformat()}") logger.info(f"[WORKER] PID: {os.getpid()}") logger.info(f"[WORKER] Working Directory: {os.getcwd()}") logger.info("=" * 60) # Flush after critical startup messages sys.stdout.flush() sys.stderr.flush() # Ensure logs directory exists Path("/app/logs").mkdir(parents=True, exist_ok=True) # CRITICAL: Send initial heartbeat with retry until app is ready # This fixes the race condition where worker starts before FastAPI is accepting requests logger.info("[WORKER] Attempting initial heartbeat (will retry until app responds)...") heartbeat_sent = False max_retries = 30 # Try for up to 30 seconds retry_delay = 1.0 for attempt in range(max_retries): try: _send_heartbeat_to_app(worker_state="active", current_state="idle") heartbeat_sent = True logger.info(f"[WORKER] Initial heartbeat succeeded on attempt {attempt + 1}") break except Exception as e: if attempt < max_retries - 1: logger.debug(f"[WORKER] Heartbeat attempt {attempt + 1} failed: {e}, retrying in {retry_delay}s...") time.sleep(retry_delay) else: logger.error(f"[WORKER] Initial heartbeat failed after {max_retries} attempts: {e}") sys.stdout.flush() sys.stderr.flush() # Start heartbeat loop in thread heartbeat_thread = Thread(target=worker_heartbeat_loop, daemon=True) heartbeat_thread.start() # Start asyncio polling in thread polling_thread = Thread(target=run_asyncio_polling, daemon=True) polling_thread.start() logger.info("[WORKER] Worker threads started, waiting for shutdown signal...") # Wait for shutdown signal try: while not _shutdown_event.is_set(): time.sleep(1) except KeyboardInterrupt: logger.info("[WORKER] Keyboard interrupt received") _shutdown_event.set() # Graceful shutdown logger.info("[WORKER] Shutting down worker process...") _shutdown_event.set() heartbeat_thread.join(timeout=5) polling_thread.join(timeout=5) logger.info("[WORKER] Worker process stopped cleanly") if __name__ == "__main__": # Ensure /app is in path for imports if "/app" not in sys.path: sys.path.insert(0, "/app") # FAIL-LOUD: Catch any crash and write traceback to startup_error.log try: main() except Exception as e: import traceback from pathlib import Path startup_error_log = Path("/app/startup_error.log") error_details = f"""WORKER CRASH - {datetime.utcnow().isoformat()} Exception Type: {type(e).__name__} Exception Message: {str(e)} Traceback: {traceback.format_exc()} Environment: - Working Directory: {os.getcwd()} - PYTHONPATH: {sys.path[:3]}... - CLAUDE_API_KEY: {'SET' if os.environ.get('CLAUDE_API_KEY') else 'NOT SET'} - ANTHROPIC_API_KEY: {'SET' if os.environ.get('ANTHROPIC_API_KEY') else 'NOT SET'} """ # SCREAM to file try: startup_error_log.write_text(error_details) except Exception: pass # SCREAM to stdout print(f"\n{'='*60}\nWORKER CRASH:\n{error_details}{'='*60}\n") sys.stdout.flush() sys.stderr.flush() # Re-raise to exit with error code raise