brain_minimal.py

#!/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