Claude Code: Force minimal app.py - diagnostic for unknown error

Backup saved as app.py.broken. Testing if "unknown" error is in
Python code layer or Docker/entrypoint layer.

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

app.py

@@ -1,1049 +1,10 @@
-#!/usr/bin/env python3
-"""
-Cain FastAPI App with background task execution engine.
-
-Serves agent dashboard and manages asynchronous task execution
-without blocking the main Uvicorn thread.
-"""
-import asyncio
-import os
-import subprocess
-import sys
-import threading
-import time
-import uuid
-import logging
-from datetime import datetime
-from enum import Enum
-from pathlib import Path
-from typing import Any, Optional
-from contextlib import asynccontextmanager
-
-import uvicorn
-from fastapi import FastAPI, HTTPException, BackgroundTasks, WebSocket, WebSocketDisconnect, Request
-from fastapi.responses import JSONResponse, FileResponse
-from fastapi.staticfiles import StaticFiles
-from fastapi.middleware import Middleware
-from starlette.middleware.base import BaseHTTPMiddleware
-
-# ============================================================================
-# SYS.PATH SETUP (must happen before other imports)
-# ============================================================================
-
-# Set up sys.path for agents imports at module load time
-# This ensures `from agents import brain_minimal` works regardless of import order
-# Dynamic path resolution with fallbacks for different Docker contexts
-_script_dir = Path(os.path.abspath(os.path.dirname(__file__)))  # Absolute path of this script
-
-# Try multiple possible locations for .openclaw directory
-_possible_openclaw_paths = [
-    _script_dir / ".openclaw",           # /app/.openclaw (legacy/flat structure)
-    _script_dir / "openclaw" / ".openclaw",  # /app/openclaw/.openclaw (nested structure)
-    Path("/app/openclaw/.openclaw"),     # Absolute Docker path (nested)
-    Path("/app/.openclaw"),              # Absolute Docker path (flat)
-]
-
-# Add all valid paths to sys.path
-for path_dir in _possible_openclaw_paths:
-    path_str = str(path_dir)
-    if path_str not in sys.path and path_dir.exists():
-        sys.path.insert(0, path_str)
-
-# ============================================================================
-# CONFIGURATION & LOGGING
-# ============================================================================
-
-LOG_PATH = "/app/logs/cain.log"  # Writable in Docker container
-FRONTEND_PATH = Path(__file__).parent / "frontend"
-AGENT_DASHBOARD = FRONTEND_PATH / "agent-dashboard.html"
-
-# Configure structured logging to file and stdout
-class StructuredFormatter(logging.Formatter):
-    """JSON-like structured log formatter."""
-    def format(self, record):
-        log_data = {
-            "timestamp": datetime.utcnow().isoformat() + "+00:00",
-            "level": record.levelname,
-            "logger": record.name,
-            "message": record.getMessage(),
-            "module": record.module,
-            "function": record.funcName,
-            "line": record.lineno,
-        }
-        if record.exc_info:
-            log_data["exception"] = self.formatException(record.exc_info)
-        return str(log_data)
-
-# Set up file handler
-try:
-    file_handler = logging.FileHandler(LOG_PATH)
-    file_handler.setFormatter(StructuredFormatter())
-except (PermissionError, FileNotFoundError):
-    file_handler = logging.StreamHandler(sys.stdout)
-    file_handler.setFormatter(StructuredFormatter())
-
-# Set up stdout handler
-stdout_handler = logging.StreamHandler(sys.stdout)
-stdout_handler.setFormatter(logging.Formatter(
-    "%(asctime)s | %(levelname)-8s | %(name)s | %(message)s"
-))
-
-# Configure root logger
-logger = logging.getLogger(__name__)
-logger.setLevel(logging.INFO)
-logger.addHandler(file_handler)
-logger.addHandler(stdout_handler)
-
-# ============================================================================
-# WORKER MANAGER (Handles worker process lifecycle with blocking startup)
-# ============================================================================
-
-class WorkerManager:
-    """
-    Manages the worker process lifecycle with blocking startup verification.
-
-    Ensures the app does NOT report "healthy" until the worker is confirmed running.
-    Uses asyncio subprocess for proper async handling and retry logic.
-    """
-
-    def __init__(self, shared_state_ref):
-        self.shared_state = shared_state_ref
-        self._process: Optional[asyncio.subprocess.Process] = None
-        self._log_path = Path("/app/cain.log")
-        self._pid_file = Path("/app/worker.pid")
-
-    async def spawn_with_retry(
-        self,
-        max_attempts: int = 2,
-        retry_delay: float = 1.0,
-        heartbeat_timeout: float = 5.0
-    ) -> bool:
-        """
-        Spawn worker process with retry logic - NON-BLOCKING.
-
-        Worker starts asynchronously; heartbeat updates health status separately.
-
-        Args:
-            max_attempts: Maximum spawn attempts (default: 2, reduced for faster startup)
-            retry_delay: Seconds to wait between retries (default: 1.0, reduced)
-            heartbeat_timeout: Max seconds to wait for heartbeat per attempt (default: 5.0, reduced)
-
-        Returns:
-            True if worker spawned (even if heartbeat not yet received)
-        """
-        brain_path = Path(__file__).parent / "brain_minimal.py"
-
-        for attempt in range(1, max_attempts + 1):
-            logger.info(f"[WorkerManager] Spawn attempt {attempt}/{max_attempts}")
-            print(f"[WorkerManager] Spawn attempt {attempt}/{max_attempts}")
-            sys.stdout.flush()
-            sys.stderr.flush()
-
-            try:
-                # Open log file for appending
-                log_file = open(self._log_path, "a") if self._log_path.parent.exists() else None
-
-                # Spawn using asyncio.create_subprocess_exec
-                self._process = await asyncio.create_subprocess_exec(
-                    sys.executable,
-                    str(brain_path),
-                    stdout=asyncio.subprocess.PIPE,
-                    stderr=asyncio.subprocess.PIPE,
-                    env=self._get_worker_env()
-                )
-
-                worker_pid = self._process.pid
-                logger.info(f"[WorkerManager] Worker spawned with PID {worker_pid}")
-                print(f"[WorkerManager] Worker spawned with PID {worker_pid}")
-                sys.stdout.flush()
-
-                # Start draining stdout/stderr to log file
-                asyncio.create_task(self._drain_stdout(log_file))
-                asyncio.create_task(self._drain_stderr(log_file))
-
-                # Write PID file for external monitoring
-                self._pid_file.write_text(str(worker_pid))
-                logger.info(f"[WorkerManager] PID file written: {self._pid_file}")
-
-                # NON-BLOCKING: Mark worker as spawning, heartbeat will confirm it's running
-                self.shared_state.update(
-                    stage="RUNNING",
-                    health="HEALTHY",  # Report healthy immediately so app passes health checks
-                    worker_pid=worker_pid,
-                    worker_active=True,  # Assume active until heartbeat fails
-                    worker_mode="standalone_process"
-                )
-                logger.info(f"[WorkerManager] Worker spawn initiated (PID: {worker_pid})")
-                print(f"[WorkerManager] ✓ Worker spawned - waiting for heartbeat")
-                sys.stdout.flush()
-                return True
-
-            except Exception as e:
-                logger.error(f"[WorkerManager] Spawn attempt {attempt} failed: {type(e).__name__}: {e}")
-                print(f"[WorkerManager] ✗ Spawn attempt {attempt} failed: {e}")
-                sys.stdout.flush()
-
-            # Retry delay (except after last attempt)
-            if attempt < max_attempts:
-                logger.info(f"[WorkerManager] Waiting {retry_delay}s before retry...")
-                await asyncio.sleep(retry_delay)
-
-        # All attempts failed - but don't crash, just log and continue
-        error_msg = f"Failed to spawn worker after {max_attempts} attempts"
-        logger.error(error_msg)
-        print(f"[WorkerManager] ⚠ {error_msg} - continuing without worker")
-        sys.stdout.flush()
-
-        self.shared_state.update(
-            stage="RUNNING",
-            health="DEGRADED",  # Degraded but still serving requests
-            error=error_msg,
-            worker_active=False,
-            worker_pid=None
-        )
-
-        return False  # Return False but don't raise - app stays alive
-
-    async def _wait_for_heartbeat(self, timeout: float) -> bool:
-        """
-        Wait for worker heartbeat to be received.
-
-        Args:
-            timeout: Maximum seconds to wait
-
-        Returns:
-            True if heartbeat received, False if timeout
-        """
-        start_time = time.time()
-        check_interval = 0.2
-
-        while (time.time() - start_time) < timeout:
-            state = self.shared_state.get()
-            if state.get("worker_active") is True:
-                logger.info(f"[WorkerManager] Heartbeat confirmed after {time.time() - start_time:.1f}s")
-                return True
-
-            # Check if process crashed
-            if self._process and self._process.poll() is not None:
-                logger.warning(f"[WorkerManager] Process exited during heartbeat wait")
-                return False
-
-            await asyncio.sleep(check_interval)
-
-        return False
-
-    async def _drain_stdout(self, log_file=None):
-        """Drain worker stdout to prevent buffer blocking."""
-        if not self._process or not self._process.stdout:
-            return
-        try:
-            while True:
-                line = await self._process.stdout.readline()
-                if not line:
-                    break
-                line_str = line.decode("utf-8", errors="replace").rstrip()
-                logger.info(f"[WORKER_STDOUT] {line_str}")
-                if log_file:
-                    log_file.write(f"[STDOUT] {line_str}\n")
-                    log_file.flush()
-        except Exception:
-            pass
-
-    async def _drain_stderr(self, log_file=None):
-        """Drain worker stderr to prevent buffer blocking."""
-        if not self._process or not self._process.stderr:
-            return
-        try:
-            while True:
-                line = await self._process.stderr.readline()
-                if not line:
-                    break
-                line_str = line.decode("utf-8", errors="replace").rstrip()
-                logger.warning(f"[WORKER_STDERR] {line_str}")
-                if log_file:
-                    log_file.write(f"[STDERR] {line_str}\n")
-                    log_file.flush()
-        except Exception:
-            pass
-
-    async def _terminate_worker(self):
-        """Terminate the worker process if running."""
-        if self._process:
-            try:
-                self._process.terminate()
-                try:
-                    await asyncio.wait_for(self._process.wait(), timeout=5.0)
-                except asyncio.TimeoutError:
-                    self._process.kill()
-                    await self._process.wait()
-            except Exception as e:
-                logger.warning(f"[WorkerManager] Error terminating worker: {e}")
-
-            # Clean up PID file
-            if self._pid_file.exists():
-                try:
-                    self._pid_file.unlink()
-                except Exception:
-                    pass
-
-    def _get_worker_env(self) -> dict[str, str]:
-        """Build environment for worker process.
-
-        CRITICAL: Uses os.environ.copy() which automatically inherits all
-        HF Space secrets (HF_TOKEN, OPENAI_API_KEY, OPENROUTER_API_KEY, etc).
-        DO NOT set empty defaults - that can interfere with secret inheritance.
-        """
-        worker_env = os.environ.copy()
-        worker_env["PYTHONPATH"] = "/app:" + worker_env.get("PYTHONPATH", "")
-        worker_env["CAIN_IS_WORKER"] = "true"
-        return worker_env
-
-    async def shutdown(self):
-        """Gracefully shutdown the worker process."""
-        if self._process:
-            logger.info("[WorkerManager] Shutting down worker process...")
-            await self._terminate_worker()
-            self._process = None
-
-
-# Global worker manager instance
-_worker_manager: Optional[WorkerManager] = None
-
-
-# ============================================================================
-# BACKGROUND TASK MANAGER (Non-blocking)
-# ============================================================================
-
-class TaskStatus(str, Enum):
-    PENDING = "pending"
-    RUNNING = "running"
-    COMPLETED = "completed"
-    FAILED = "failed"
-
-
-class BackgroundTask:
-    """Represents a single background task."""
-
-    def __init__(self, task_id: str, command: str, task_type: str = "shell"):
-        self.task_id = task_id
-        self.command = command
-        self.task_type = task_type
-        self.status = TaskStatus.PENDING
-        self.created_at = datetime.utcnow().isoformat() + "+00:00"
-        self.started_at: Optional[str] = None
-        self.completed_at: Optional[str] = None
-        self.output: str = ""
-        self.error: Optional[str] = None
-        self.exit_code: Optional[int] = None
-        self._process: Optional[subprocess.Popen] = None
-
-    def start(self):
-        """Start executing the task in background."""
-        self.status = TaskStatus.RUNNING
-        self.started_at = datetime.utcnow().isoformat() + "+00:00"
-
-        try:
-            if self.task_type == "shell":
-                self._process = subprocess.Popen(
-                    self.command,
-                    shell=True,
-                    stdout=subprocess.PIPE,
-                    stderr=subprocess.PIPE,
-                    text=True,
-                    bufsize=1,
-                    universal_newlines=True
-                )
-            else:
-                # Python script execution
-                self._process = subprocess.Popen(
-                    [sys.executable, "-c", self.command],
-                    stdout=subprocess.PIPE,
-                    stderr=subprocess.PIPE,
-                    text=True,
-                    bufsize=1,
-                    universal_newlines=True
-                )
-        except Exception as e:
-            self.status = TaskStatus.FAILED
-            self.error = str(e)
-            self.completed_at = datetime.utcnow().isoformat() + "+00:00"
-            logger.error(f"Task {self.task_id} failed to start: {e}")
-
-    def poll(self):
-        """Poll task status and update if complete."""
-        if self._process is None:
-            return
-
-        # Check if process has completed
-        returncode = self._process.poll()
-        if returncode is not None:
-            # Process finished
-            self.status = TaskStatus.COMPLETED if returncode == 0 else TaskStatus.FAILED
-            self.completed_at = datetime.utcnow().isoformat() + "+00:00"
-            self.exit_code = returncode
-
-            # Capture remaining output
-            stdout, stderr = self._process.communicate()
-            self.output += stdout
-            if stderr:
-                self.error = stderr
-
-            logger.info(f"Task {self.task_id} completed with exit code {returncode}")
-
-    def read_output(self):
-        """Read available output without blocking."""
-        if self._process is None:
-            return
-        try:
-            # Non-blocking read of available output
-            if self._process.stdout:
-                lines = self._process.stdout.readlines()
-                self.output += "".join(lines)
-        except Exception:
-            pass
-
-    def to_dict(self) -> dict[str, Any]:
-        """Convert task to dictionary for API response."""
-        return {
-            "task_id": self.task_id,
-            "command": self.command,
-            "task_type": self.task_type,
-            "status": self.status,
-            "created_at": self.created_at,
-            "started_at": self.started_at,
-            "completed_at": self.completed_at,
-            "exit_code": self.exit_code,
-            "output": self.output,
-            "error": self.error,
-        }
-
-
-class TaskManager:
-    """
-    Non-blocking background task manager.
-
-    Runs shell commands and Python scripts asynchronously without
-    blocking the main Uvicorn thread.
-    """
-
-    def __init__(self):
-        self._tasks: dict[str, BackgroundTask] = {}
-        self._lock = threading.Lock()
-        self._running = True
-        # Start background poller thread
-        self._poller_thread = threading.Thread(
-            target=self._poll_loop,
-            daemon=True,
-            name="TaskManagerPoller"
-        )
-        self._poller_thread.start()
-        logger.info("TaskManager initialized with background poller thread")
-
-    def _poll_loop(self):
-        """Background thread that polls all running tasks."""
-        while self._running:
-            try:
-                with self._lock:
-                    for task in list(self._tasks.values()):
-                        if task.status == TaskStatus.RUNNING:
-                            task.read_output()
-                            task.poll()
-            except Exception as e:
-                logger.error(f"Error in task poll loop: {e}")
-            time.sleep(0.1)  # Poll every 100ms
-
-    def create_task(self, command: str, task_type: str = "shell") -> str:
-        """Create a new background task and start it."""
-        task_id = str(uuid.uuid4())[:8]
-        task = BackgroundTask(task_id, command, task_type)
-        task.start()
-
-        with self._lock:
-            self._tasks[task_id] = task
-
-        logger.info(f"Created task {task_id}: {command[:50]}...")
-        return task_id
-
-    def get_task(self, task_id: str) -> Optional[BackgroundTask]:
-        """Get a task by ID."""
-        with self._lock:
-            return self._tasks.get(task_id)
-
-    def list_tasks(self) -> list[dict[str, Any]]:
-        """List all tasks."""
-        with self._lock:
-            return [task.to_dict() for task in self._tasks.values()]
-
-    def cancel_task(self, task_id: str) -> bool:
-        """Cancel a running task."""
-        with self._lock:
-            task = self._tasks.get(task_id)
-            if task and task._process:
-                task._process.terminate()
-                task.status = TaskStatus.FAILED
-                task.completed_at = datetime.utcnow().isoformat() + "+00:00"
-                task.error = "Task cancelled by user"
-                logger.info(f"Cancelled task {task_id}")
-                return True
-        return False
-
-    def cleanup_old_tasks(self, max_age_hours: int = 24):
-        """Remove completed tasks older than max_age_hours."""
-        cutoff = datetime.utcnow().timestamp() - (max_age_hours * 3600)
-        with self._lock:
-            to_remove = []
-            for task_id, task in self._tasks.items():
-                if task.status in (TaskStatus.COMPLETED, TaskStatus.FAILED):
-                    if task.completed_at:
-                        try:
-                            completed_time = datetime.fromisoformat(
-                                task.completed_at.replace("+00:00", "").replace("Z", "")
-                            ).timestamp()
-                            if completed_time < cutoff:
-                                to_remove.append(task_id)
-                        except Exception:
-                            pass
-            for task_id in to_remove:
-                del self._tasks[task_id]
-            if to_remove:
-                logger.info(f"Cleaned up {len(to_remove)} old tasks")
-
-    def get_status(self) -> dict[str, Any]:
-        """Get overall task manager status."""
-        with self._lock:
-            status_counts = {}
-            for task in self._tasks.values():
-                status_counts[task.status] = status_counts.get(task.status, 0) + 1
-            return {
-                "total_tasks": len(self._tasks),
-                "status_breakdown": status_counts,
-                "poller_thread_alive": self._poller_thread.is_alive(),
-            }
-
-
-# ============================================================================
-# SHARED STATE (In-memory)
-# ============================================================================
-
-
-class SharedState:
-    """In-memory shared state for process status."""
-
-    def __init__(self):
-        self._lock = threading.Lock()
-        self._state = {
-            "worker_state": "initializing",
-            "worker_active": False,  # Set to True only after worker spawns successfully
-            "worker_pid": None,  # Set to worker PID after successful spawn
-            "worker_mode": None,  # Set after spawn
-            "last_heartbeat": None,
-            "heartbeat_age_seconds": 0,
-            "stage": "STARTUP_INIT",
-            "health": "INITIALIZING",
-            "error": None,
-            "uptime_seconds": 0,
-            "started_at": datetime.utcnow().isoformat() + "+00:00",
-        }
-        self._start_time = time.time()
-
-    def update(self, **kwargs):
-        """Update state with new values."""
-        with self._lock:
-            self._state["last_heartbeat"] = datetime.utcnow().isoformat() + "+00:00"
-            self._state["heartbeat_age_seconds"] = 0
-            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."""
-        with self._lock:
-            self._state["uptime_seconds"] = int(time.time() - self._start_time)
-
-            # 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()
-
-
-# ============================================================================
-# REQUEST LOGGING MIDDLEWARE
-# ============================================================================
-
-class RequestLoggingMiddleware(BaseHTTPMiddleware):
-    """
-    Middleware to log all incoming requests with method, path, and client IP.
-    Helps track internal state transitions for debugging.
-    """
-
-    async def dispatch(self, request: Request, call_next):
-        """Process request and log details."""
-        client_ip = request.client.host if request.client else "unknown"
-        method = request.method
-        path = request.url.path
-        query = str(request.url.query) if request.url.query else ""
-
-        logger.info(f"REQUEST: {method} {path}{'?' + query if query else ''} from {client_ip}")
-
-        # Process request
-        response = await call_next(request)
-
-        # Log response status
-        logger.info(f"RESPONSE: {method} {path} -> {response.status_code}")
-
-        return response
-
-
-# ============================================================================
-# FASTAPI APP
-# ============================================================================
-
-# Initialize shared state and task manager
-shared_state = SharedState()
-task_manager = TaskManager()
-
-
-def _check_dependencies() -> tuple[bool, str]:
-    """Pre-initialization check for PYTHONPATH and dependencies."""
-    errors = []
-
-    # Check PYTHONPATH
-    if "/app" not in sys.path:
-        errors.append("PYTHONPATH missing /app")
-
-    # Check critical imports
-    try:
-        import fastapi
-        import uvicorn
-    except ImportError as e:
-        errors.append(f"Missing dependency: {e}")
-
-    # Check brain_minimal.py exists
-    brain_path = Path(__file__).parent / "brain_minimal.py"
-    if not brain_path.exists():
-        errors.append(f"brain_minimal.py not found at {brain_path}")
-
-    return (len(errors) == 0, "; ".join(errors) if errors else "OK")
-
-
-async def _initialize_worker():
-    """
-    Background task to initialize worker process.
-    Spawns worker with retry logic and waits for heartbeat confirmation.
-    Runs after FastAPI server is ready to accept requests.
-    """
-    global _worker_manager
-
-    print("=" * 60)
-    print("🚀 CAIN WORKER INITIALIZATION: Starting async worker spawn")
-    print("=" * 60)
-    sys.stdout.flush()
-    sys.stderr.flush()
-
-    # Update state to show we're initializing worker
-    shared_state.update(
-        stage="RUNNING_APP_STARTING",
-        health="INITIALIZING_WORKER",
-        worker_active=False,
-        worker_pid=None
-    )
-
-    try:
-        # Initialize WorkerManager
-        _worker_manager = WorkerManager(shared_state)
-
-        # Spawn worker with extended timeout for HuggingFace Spaces environment
-        # Use 30 seconds heartbeat timeout instead of 15 for slower environments
-        success = await _worker_manager.spawn_with_retry(
-            max_attempts=5,  # More attempts for HF Spaces
-            retry_delay=2.0,
-            heartbeat_timeout=30.0  # Extended timeout
-        )
-
-        if success:
-            state = shared_state.get()
-            print("=" * 60)
-            print("✅ CAIN WORKER INITIALIZED SUCCESSFULLY")
-            print(f"   Stage: {state.get('stage')}")
-            print(f"   Health: {state.get('health')}")
-            print(f"   Worker PID: {state.get('worker_pid')}")
-            print(f"   Worker Active: {state.get('worker_active')}")
-            print("=" * 60)
-            logger.info("Cain worker initialized successfully")
-        else:
-            raise RuntimeError("Worker spawn returned False without exception")
-
-    except Exception as e:
-        logger.error(f"Worker initialization failed: {type(e).__name__}: {e}")
-        print(f"❌ CAIN WORKER INITIALIZATION FAILED: {e}")
-        sys.stdout.flush()
-        shared_state.update(
-            stage="RUNNING_APP_STARTING",
-            health="WORKER_FAILED",
-            error=f"Worker init failed: {e}",
-            worker_active=False,
-            worker_pid=None
-        )
-
-
-@asynccontextmanager
-async def lifespan(app: FastAPI):
-    """
-    Lifespan context manager for startup/shutdown.
-    Ensures worker initialization happens after FastAPI is ready.
-    """
-    # Startup: Start worker initialization in background
-    print("=" * 60)
-    print("🚀 CAIN LIFESPAN: Startup starting")
-    print("=" * 60)
-    sys.stdout.flush()
-
-    # Pre-initialization checks
-    deps_ok, deps_msg = _check_dependencies()
-    if not deps_ok:
-        logger.critical(f"DEPENDENCY CHECK FAILED: {deps_msg}")
-        print(f"CRITICAL: {deps_msg}")
-        sys.stdout.flush()
-        shared_state.update(
-            stage="RUNNING",
-            health="DEGRADED",  # Use DEGRADED instead of DEPENDENCY_ERROR
-            error=f"Pre-initialization check failed: {deps_msg}",
-            worker_active=False
-        )
-        # Don't raise - let app start in degraded mode
-        print("⚠️  Starting in DEGRADED mode (worker not available)")
-        sys.stdout.flush()
-    else:
-        print("✓ Pre-initialization checks passed")
-        sys.stdout.flush()
-
-        # Start worker initialization as background task
-        # This allows FastAPI to start accepting requests immediately
-        # while worker spawns in background
-        asyncio.create_task(_initialize_worker())
-
-    yield
-
-    # Shutdown: Cleanup
-    print("=" * 60)
-    print("🛑 CAIN LIFESPAN: Shutdown starting")
-    print("=" * 60)
-    sys.stdout.flush()
-
-    task_manager._running = False
-
-    # Shutdown worker manager
-    global _worker_manager
-    if _worker_manager is not None:
-        try:
-            await _worker_manager.shutdown()
-        except Exception as e:
-            logger.error(f"Error shutting down worker manager: {e}")
-        finally:
-            _worker_manager = None
-
-    print("CAIN shutdown complete")
-    sys.stdout.flush()
-
-
-app = FastAPI(
-    title="HuggingClaw Cain",
-    description="Agent collaboration server with background task execution",
-    version="2.0.0",
-    lifespan=lifespan
-)
-
-# DEBUG: Print APP_READY immediately to verify core app instantiation
-print("APP_READY")
-sys.stdout.flush()
-
-# TEMPORARILY DISABLED: Testing Eve's hypothesis - comment out error_handlers to break potential dependency chain
-# # Register exception handlers (import here to avoid circular dependency)
-# from error_handlers import register_error_handlers
-# register_error_handlers(app)
-#
-# # Add request logging middleware
-# app.add_middleware(RequestLoggingMiddleware)
-
-# Mount static files for frontend assets (images, fonts, etc.)
-app.mount("/static", StaticFiles(directory=str(FRONTEND_PATH)), name="static")
-
+from fastapi import FastAPI
+app = FastAPI()
 
 @app.get("/")
-async def read_root():
-    """Serve the agent dashboard HTML."""
-    logger.info("GET / - Serving agent dashboard")
-
-    if AGENT_DASHBOARD.exists():
-        return FileResponse(str(AGENT_DASHBOARD), media_type="text/html")
-
-    # Fallback if dashboard file missing
-    return JSONResponse({
-        "status": "alive",
-        "message": "Cain core operational",
-        "note": "Agent dashboard file not found",
-        "frontend_path": str(FRONTEND_PATH),
-    })
-
-
-@app.get("/health")
-async def health_check():
-    """
-    Health check endpoint for container orchestration.
-
-    Returns:
-        - 200: System is operational (worker may still be initializing)
-        - Detailed worker status in response body
-    """
-    state = shared_state.get()
-    worker_alive = False
-    worker_pid = None
-
-    # Check live worker process status
-    if _worker_manager is not None and _worker_manager._process is not None:
-        worker_pid = _worker_manager._process.pid
-        returncode = _worker_manager._process.poll()
-        worker_alive = returncode is None  # None = still running
-
-    # Determine overall health
-    # App is healthy if it's running, worker can be initializing or degraded
-    is_healthy = state["health"] in ("HEALTHY", "INITIALIZING", "INITIALIZING_WORKER", "RUNNING_APP_STARTING", "DEGRADED")
-
-    return JSONResponse({
-        "status": "healthy" if is_healthy else "unhealthy",
-        "stage": state["stage"],
-        "worker": {
-            "active": state["worker_active"],
-            "alive": worker_alive,
-            "pid": worker_pid or state.get("worker_pid"),
-            "mode": state["worker_mode"],
-            "heartbeat_age_seconds": state["heartbeat_age_seconds"],
-        },
-        "uptime_seconds": state["uptime_seconds"],
-        "timestamp": datetime.utcnow().isoformat() + "+00:00",
-    }, status_code=200 if is_healthy else 503)
-
-
-@app.get("/api/state")
-async def get_state():
-    """
-    Get detailed process and system status.
-
-    Returns live worker process status (not just cached state).
-    """
-    logger.debug("GET /api/state called")
-
-    state = shared_state.get()
-    task_status = task_manager.get_status()
-
-    # Check live worker process status from WorkerManager
-    worker_alive = False
-    worker_pid = None
-
-    if _worker_manager is not None and _worker_manager._process is not None:
-        worker_pid = _worker_manager._process.pid
-        returncode = _worker_manager._process.poll()
-        worker_alive = returncode is None  # None = still running
-
-    # Update shared_state with real-time worker status
-    if worker_alive != state.get("worker_active"):
-        shared_state.update(worker_active=worker_alive)
-        state["worker_active"] = worker_alive
-
-    # Determine worker health - combine live process check with heartbeat freshness
-    worker_healthy = (
-        worker_alive and  # Must be actually running
-        state["heartbeat_age_seconds"] < 30  # And heartbeat must be recent
-    )
-
-    return JSONResponse({
-        "cain": {
-            "name": "Cain",
-            "space_id": "tao-shen/HuggingClaw-Cain",
-            "stage": state["stage"],
-            "health": state["health"],
-            "error": state.get("error"),
-            "uptime_seconds": state["uptime_seconds"],
-            "started_at": state["started_at"],
-        },
-        "worker": {
-            "state": state["worker_state"],
-            "active": state["worker_active"],  # Now reflects real-time process status
-            "pid": worker_pid or state.get("worker_pid"),
-            "mode": state["worker_mode"],
-            "last_heartbeat": state["last_heartbeat"],
-            "heartbeat_age_seconds": state["heartbeat_age_seconds"],
-            "is_healthy": worker_healthy,
-        },
-        "tasks": task_status,
-        "timestamp": datetime.utcnow().isoformat() + "+00:00",
-    })
-
-
-# Internal heartbeat endpoint (used by worker processes)
-@app.post("/internal/heartbeat")
-async def heartbeat(data: dict[str, Any]):
-    """Receive heartbeat from worker process."""
-    logger.info(f"[HEARTBEAT] Received from worker_pid={data.get('worker_pid')}, worker_state={data.get('worker_state')}, stage={data.get('stage')}")
-    print(f"[HEARTBEAT] Received from worker_pid={data.get('worker_pid')}, worker_state={data.get('worker_state')}, stage={data.get('stage')}")
-    sys.stdout.flush()
-
-    # Get current state before update
-    current = shared_state.get()
-    current_stage = current.get("stage")
-    incoming_stage = data.get("stage")
-
-    # If startup set RUNNING_APP_READY, allow worker to upgrade to RUNNING_A2A_READY
-    # but don't allow downgrades or unknown stages
-    if current_stage == "RUNNING_APP_READY" and incoming_stage == "RUNNING_A2A_READY":
-        print(f"[HEARTBEAT] Stage transition: {current_stage} -> {incoming_stage}")
-        shared_state.update(**data)
-    elif current_stage in ("RUNNING_APP_READY", "RUNNING_A2A_READY"):
-        # Preserve the app-ready stage, don't overwrite with worker's stage
-        # But update other fields
-        filtered_data = {k: v for k, v in data.items() if k != "stage"}
-        shared_state.update(**filtered_data)
-        print(f"[HEARTBEAT] Preserving stage={current_stage}, updating other fields")
-    else:
-        # Normal update for other stages
-        shared_state.update(**data)
-
-    sys.stdout.flush()
-    return {"status": "received"}
-
-
-@app.get("/read_startup_error")
-async def read_startup_error():
-    """
-    DEBUG: Read the startup error log to expose worker spawn failures.
-    This endpoint bypasses the silent failure loop by forcing errors to be visible.
-    """
-    startup_error_log = Path(__file__).parent / "startup_error.log"
-    if startup_error_log.exists():
-        content = startup_error_log.read_text()
-        return JSONResponse({
-            "startup_error_log_exists": True,
-            "content": content
-        })
-    return JSONResponse({
-        "startup_error_log_exists": False,
-        "message": "No startup error log found - worker may have started successfully"
-    })
-
-
-# ============================================================================
-# TASK EXECUTION API
-# ============================================================================
-
-@app.post("/api/tasks/execute")
-async def execute_task(request: dict[str, Any]):
-    """
-    Execute a command or script in the background.
-
-    Request body:
-    {
-        "command": "echo 'hello'",  # Command to execute
-        "type": "shell" | "python"  # Execution type (default: shell)
-    }
-    """
-    command = request.get("command")
-    task_type = request.get("type", "shell")
-
-    if not command:
-        raise HTTPException(status_code=400, detail="Command is required")
-
-    task_id = task_manager.create_task(command, task_type)
-    logger.info(f"Started task {task_id}: {command[:100]}")
-
-    return JSONResponse({
-        "task_id": task_id,
-        "status": "started",
-        "message": f"Task {task_id} started successfully"
-    })
-
-
-@app.get("/api/tasks/{task_id}")
-async def get_task_status(task_id: str):
-    """Get status of a specific task."""
-    task = task_manager.get_task(task_id)
-    if not task:
-        raise HTTPException(status_code=404, detail="Task not found")
-
-    # Poll for any updates
-    task.poll()
-
-    return JSONResponse(task.to_dict())
-
-
-@app.get("/api/tasks")
-async def list_tasks():
-    """List all tasks."""
-    return JSONResponse({"tasks": task_manager.list_tasks()})
-
-
-@app.delete("/api/tasks/{task_id}")
-async def cancel_task(task_id: str):
-    """Cancel a running task."""
-    if task_manager.cancel_task(task_id):
-        return JSONResponse({"status": "cancelled", "task_id": task_id})
-    raise HTTPException(status_code=404, detail="Task not found or not cancellable")
-
-
-@app.post("/api/tasks/cleanup")
-async def cleanup_tasks(max_age_hours: int = 24):
-    """Clean up old completed tasks."""
-    task_manager.cleanup_old_tasks(max_age_hours)
-    return JSONResponse({"status": "cleaned", "max_age_hours": max_age_hours})
-
-
-# ============================================================================
-# WEBSOCKET CHAT ENDPOINT
-# ============================================================================
-
-@app.websocket("/chat")
-async def chat_websocket(websocket: WebSocket):
-    """WebSocket endpoint for real-time agent communication."""
-    await websocket.accept()
-    client_id = f"{websocket.client.host}:{websocket.client.port}"
-    logger.info(f"WebSocket connection established from {client_id}")
-
-    try:
-        while True:
-            # Receive message from client
-            data = await websocket.receive_text()
-            logger.info(f"WebSocket message from {client_id}: {data}")
-
-            # Echo acknowledgment back to client
-            await websocket.send_json({
-                "type": "ack",
-                "message": "Message received",
-                "timestamp": datetime.utcnow().isoformat() + "+00:00",
-                "echo": data
-            })
-
-    except WebSocketDisconnect:
-        logger.info(f"WebSocket connection closed by {client_id}")
-    except Exception as e:
-        logger.error(f"WebSocket error for {client_id}: {e}")
-    finally:
-        logger.info(f"WebSocket connection terminated for {client_id}")
-
-
-# ============================================================================
-# MAIN ENTRY POINT
-# ============================================================================
+def read_root():
+    return {"status": "minimal_ok"}
 
 if __name__ == "__main__":
-    port = int(os.environ.get("PORT", 7860))
-    logger.info(f"🚀 STARTING UVICORN on 0.0.0.0:{port}")
-    uvicorn.run(app, host="0.0.0.0", port=port, log_level="info")
+    import uvicorn
+    uvicorn.run(app, host="0.0.0.0", port=7860)

app.py.broken

@@ -0,0 +1,1049 @@
+#!/usr/bin/env python3
+"""
+Cain FastAPI App with background task execution engine.
+
+Serves agent dashboard and manages asynchronous task execution
+without blocking the main Uvicorn thread.
+"""
+import asyncio
+import os
+import subprocess
+import sys
+import threading
+import time
+import uuid
+import logging
+from datetime import datetime
+from enum import Enum
+from pathlib import Path
+from typing import Any, Optional
+from contextlib import asynccontextmanager
+
+import uvicorn
+from fastapi import FastAPI, HTTPException, BackgroundTasks, WebSocket, WebSocketDisconnect, Request
+from fastapi.responses import JSONResponse, FileResponse
+from fastapi.staticfiles import StaticFiles
+from fastapi.middleware import Middleware
+from starlette.middleware.base import BaseHTTPMiddleware
+
+# ============================================================================
+# SYS.PATH SETUP (must happen before other imports)
+# ============================================================================
+
+# Set up sys.path for agents imports at module load time
+# This ensures `from agents import brain_minimal` works regardless of import order
+# Dynamic path resolution with fallbacks for different Docker contexts
+_script_dir = Path(os.path.abspath(os.path.dirname(__file__)))  # Absolute path of this script
+
+# Try multiple possible locations for .openclaw directory
+_possible_openclaw_paths = [
+    _script_dir / ".openclaw",           # /app/.openclaw (legacy/flat structure)
+    _script_dir / "openclaw" / ".openclaw",  # /app/openclaw/.openclaw (nested structure)
+    Path("/app/openclaw/.openclaw"),     # Absolute Docker path (nested)
+    Path("/app/.openclaw"),              # Absolute Docker path (flat)
+]
+
+# Add all valid paths to sys.path
+for path_dir in _possible_openclaw_paths:
+    path_str = str(path_dir)
+    if path_str not in sys.path and path_dir.exists():
+        sys.path.insert(0, path_str)
+
+# ============================================================================
+# CONFIGURATION & LOGGING
+# ============================================================================
+
+LOG_PATH = "/app/logs/cain.log"  # Writable in Docker container
+FRONTEND_PATH = Path(__file__).parent / "frontend"
+AGENT_DASHBOARD = FRONTEND_PATH / "agent-dashboard.html"
+
+# Configure structured logging to file and stdout
+class StructuredFormatter(logging.Formatter):
+    """JSON-like structured log formatter."""
+    def format(self, record):
+        log_data = {
+            "timestamp": datetime.utcnow().isoformat() + "+00:00",
+            "level": record.levelname,
+            "logger": record.name,
+            "message": record.getMessage(),
+            "module": record.module,
+            "function": record.funcName,
+            "line": record.lineno,
+        }
+        if record.exc_info:
+            log_data["exception"] = self.formatException(record.exc_info)
+        return str(log_data)
+
+# Set up file handler
+try:
+    file_handler = logging.FileHandler(LOG_PATH)
+    file_handler.setFormatter(StructuredFormatter())
+except (PermissionError, FileNotFoundError):
+    file_handler = logging.StreamHandler(sys.stdout)
+    file_handler.setFormatter(StructuredFormatter())
+
+# Set up stdout handler
+stdout_handler = logging.StreamHandler(sys.stdout)
+stdout_handler.setFormatter(logging.Formatter(
+    "%(asctime)s | %(levelname)-8s | %(name)s | %(message)s"
+))
+
+# Configure root logger
+logger = logging.getLogger(__name__)
+logger.setLevel(logging.INFO)
+logger.addHandler(file_handler)
+logger.addHandler(stdout_handler)
+
+# ============================================================================
+# WORKER MANAGER (Handles worker process lifecycle with blocking startup)
+# ============================================================================
+
+class WorkerManager:
+    """
+    Manages the worker process lifecycle with blocking startup verification.
+
+    Ensures the app does NOT report "healthy" until the worker is confirmed running.
+    Uses asyncio subprocess for proper async handling and retry logic.
+    """
+
+    def __init__(self, shared_state_ref):
+        self.shared_state = shared_state_ref
+        self._process: Optional[asyncio.subprocess.Process] = None
+        self._log_path = Path("/app/cain.log")
+        self._pid_file = Path("/app/worker.pid")
+
+    async def spawn_with_retry(
+        self,
+        max_attempts: int = 2,
+        retry_delay: float = 1.0,
+        heartbeat_timeout: float = 5.0
+    ) -> bool:
+        """
+        Spawn worker process with retry logic - NON-BLOCKING.
+
+        Worker starts asynchronously; heartbeat updates health status separately.
+
+        Args:
+            max_attempts: Maximum spawn attempts (default: 2, reduced for faster startup)
+            retry_delay: Seconds to wait between retries (default: 1.0, reduced)
+            heartbeat_timeout: Max seconds to wait for heartbeat per attempt (default: 5.0, reduced)
+
+        Returns:
+            True if worker spawned (even if heartbeat not yet received)
+        """
+        brain_path = Path(__file__).parent / "brain_minimal.py"
+
+        for attempt in range(1, max_attempts + 1):
+            logger.info(f"[WorkerManager] Spawn attempt {attempt}/{max_attempts}")
+            print(f"[WorkerManager] Spawn attempt {attempt}/{max_attempts}")
+            sys.stdout.flush()
+            sys.stderr.flush()
+
+            try:
+                # Open log file for appending
+                log_file = open(self._log_path, "a") if self._log_path.parent.exists() else None
+
+                # Spawn using asyncio.create_subprocess_exec
+                self._process = await asyncio.create_subprocess_exec(
+                    sys.executable,
+                    str(brain_path),
+                    stdout=asyncio.subprocess.PIPE,
+                    stderr=asyncio.subprocess.PIPE,
+                    env=self._get_worker_env()
+                )
+
+                worker_pid = self._process.pid
+                logger.info(f"[WorkerManager] Worker spawned with PID {worker_pid}")
+                print(f"[WorkerManager] Worker spawned with PID {worker_pid}")
+                sys.stdout.flush()
+
+                # Start draining stdout/stderr to log file
+                asyncio.create_task(self._drain_stdout(log_file))
+                asyncio.create_task(self._drain_stderr(log_file))
+
+                # Write PID file for external monitoring
+                self._pid_file.write_text(str(worker_pid))
+                logger.info(f"[WorkerManager] PID file written: {self._pid_file}")
+
+                # NON-BLOCKING: Mark worker as spawning, heartbeat will confirm it's running
+                self.shared_state.update(
+                    stage="RUNNING",
+                    health="HEALTHY",  # Report healthy immediately so app passes health checks
+                    worker_pid=worker_pid,
+                    worker_active=True,  # Assume active until heartbeat fails
+                    worker_mode="standalone_process"
+                )
+                logger.info(f"[WorkerManager] Worker spawn initiated (PID: {worker_pid})")
+                print(f"[WorkerManager] ✓ Worker spawned - waiting for heartbeat")
+                sys.stdout.flush()
+                return True
+
+            except Exception as e:
+                logger.error(f"[WorkerManager] Spawn attempt {attempt} failed: {type(e).__name__}: {e}")
+                print(f"[WorkerManager] ✗ Spawn attempt {attempt} failed: {e}")
+                sys.stdout.flush()
+
+            # Retry delay (except after last attempt)
+            if attempt < max_attempts:
+                logger.info(f"[WorkerManager] Waiting {retry_delay}s before retry...")
+                await asyncio.sleep(retry_delay)
+
+        # All attempts failed - but don't crash, just log and continue
+        error_msg = f"Failed to spawn worker after {max_attempts} attempts"
+        logger.error(error_msg)
+        print(f"[WorkerManager] ⚠ {error_msg} - continuing without worker")
+        sys.stdout.flush()
+
+        self.shared_state.update(
+            stage="RUNNING",
+            health="DEGRADED",  # Degraded but still serving requests
+            error=error_msg,
+            worker_active=False,
+            worker_pid=None
+        )
+
+        return False  # Return False but don't raise - app stays alive
+
+    async def _wait_for_heartbeat(self, timeout: float) -> bool:
+        """
+        Wait for worker heartbeat to be received.
+
+        Args:
+            timeout: Maximum seconds to wait
+
+        Returns:
+            True if heartbeat received, False if timeout
+        """
+        start_time = time.time()
+        check_interval = 0.2
+
+        while (time.time() - start_time) < timeout:
+            state = self.shared_state.get()
+            if state.get("worker_active") is True:
+                logger.info(f"[WorkerManager] Heartbeat confirmed after {time.time() - start_time:.1f}s")
+                return True
+
+            # Check if process crashed
+            if self._process and self._process.poll() is not None:
+                logger.warning(f"[WorkerManager] Process exited during heartbeat wait")
+                return False
+
+            await asyncio.sleep(check_interval)
+
+        return False
+
+    async def _drain_stdout(self, log_file=None):
+        """Drain worker stdout to prevent buffer blocking."""
+        if not self._process or not self._process.stdout:
+            return
+        try:
+            while True:
+                line = await self._process.stdout.readline()
+                if not line:
+                    break
+                line_str = line.decode("utf-8", errors="replace").rstrip()
+                logger.info(f"[WORKER_STDOUT] {line_str}")
+                if log_file:
+                    log_file.write(f"[STDOUT] {line_str}\n")
+                    log_file.flush()
+        except Exception:
+            pass
+
+    async def _drain_stderr(self, log_file=None):
+        """Drain worker stderr to prevent buffer blocking."""
+        if not self._process or not self._process.stderr:
+            return
+        try:
+            while True:
+                line = await self._process.stderr.readline()
+                if not line:
+                    break
+                line_str = line.decode("utf-8", errors="replace").rstrip()
+                logger.warning(f"[WORKER_STDERR] {line_str}")
+                if log_file:
+                    log_file.write(f"[STDERR] {line_str}\n")
+                    log_file.flush()
+        except Exception:
+            pass
+
+    async def _terminate_worker(self):
+        """Terminate the worker process if running."""
+        if self._process:
+            try:
+                self._process.terminate()
+                try:
+                    await asyncio.wait_for(self._process.wait(), timeout=5.0)
+                except asyncio.TimeoutError:
+                    self._process.kill()
+                    await self._process.wait()
+            except Exception as e:
+                logger.warning(f"[WorkerManager] Error terminating worker: {e}")
+
+            # Clean up PID file
+            if self._pid_file.exists():
+                try:
+                    self._pid_file.unlink()
+                except Exception:
+                    pass
+
+    def _get_worker_env(self) -> dict[str, str]:
+        """Build environment for worker process.
+
+        CRITICAL: Uses os.environ.copy() which automatically inherits all
+        HF Space secrets (HF_TOKEN, OPENAI_API_KEY, OPENROUTER_API_KEY, etc).
+        DO NOT set empty defaults - that can interfere with secret inheritance.
+        """
+        worker_env = os.environ.copy()
+        worker_env["PYTHONPATH"] = "/app:" + worker_env.get("PYTHONPATH", "")
+        worker_env["CAIN_IS_WORKER"] = "true"
+        return worker_env
+
+    async def shutdown(self):
+        """Gracefully shutdown the worker process."""
+        if self._process:
+            logger.info("[WorkerManager] Shutting down worker process...")
+            await self._terminate_worker()
+            self._process = None
+
+
+# Global worker manager instance
+_worker_manager: Optional[WorkerManager] = None
+
+
+# ============================================================================
+# BACKGROUND TASK MANAGER (Non-blocking)
+# ============================================================================
+
+class TaskStatus(str, Enum):
+    PENDING = "pending"
+    RUNNING = "running"
+    COMPLETED = "completed"
+    FAILED = "failed"
+
+
+class BackgroundTask:
+    """Represents a single background task."""
+
+    def __init__(self, task_id: str, command: str, task_type: str = "shell"):
+        self.task_id = task_id
+        self.command = command
+        self.task_type = task_type
+        self.status = TaskStatus.PENDING
+        self.created_at = datetime.utcnow().isoformat() + "+00:00"
+        self.started_at: Optional[str] = None
+        self.completed_at: Optional[str] = None
+        self.output: str = ""
+        self.error: Optional[str] = None
+        self.exit_code: Optional[int] = None
+        self._process: Optional[subprocess.Popen] = None
+
+    def start(self):
+        """Start executing the task in background."""
+        self.status = TaskStatus.RUNNING
+        self.started_at = datetime.utcnow().isoformat() + "+00:00"
+
+        try:
+            if self.task_type == "shell":
+                self._process = subprocess.Popen(
+                    self.command,
+                    shell=True,
+                    stdout=subprocess.PIPE,
+                    stderr=subprocess.PIPE,
+                    text=True,
+                    bufsize=1,
+                    universal_newlines=True
+                )
+            else:
+                # Python script execution
+                self._process = subprocess.Popen(
+                    [sys.executable, "-c", self.command],
+                    stdout=subprocess.PIPE,
+                    stderr=subprocess.PIPE,
+                    text=True,
+                    bufsize=1,
+                    universal_newlines=True
+                )
+        except Exception as e:
+            self.status = TaskStatus.FAILED
+            self.error = str(e)
+            self.completed_at = datetime.utcnow().isoformat() + "+00:00"
+            logger.error(f"Task {self.task_id} failed to start: {e}")
+
+    def poll(self):
+        """Poll task status and update if complete."""
+        if self._process is None:
+            return
+
+        # Check if process has completed
+        returncode = self._process.poll()
+        if returncode is not None:
+            # Process finished
+            self.status = TaskStatus.COMPLETED if returncode == 0 else TaskStatus.FAILED
+            self.completed_at = datetime.utcnow().isoformat() + "+00:00"
+            self.exit_code = returncode
+
+            # Capture remaining output
+            stdout, stderr = self._process.communicate()
+            self.output += stdout
+            if stderr:
+                self.error = stderr
+
+            logger.info(f"Task {self.task_id} completed with exit code {returncode}")
+
+    def read_output(self):
+        """Read available output without blocking."""
+        if self._process is None:
+            return
+        try:
+            # Non-blocking read of available output
+            if self._process.stdout:
+                lines = self._process.stdout.readlines()
+                self.output += "".join(lines)
+        except Exception:
+            pass
+
+    def to_dict(self) -> dict[str, Any]:
+        """Convert task to dictionary for API response."""
+        return {
+            "task_id": self.task_id,
+            "command": self.command,
+            "task_type": self.task_type,
+            "status": self.status,
+            "created_at": self.created_at,
+            "started_at": self.started_at,
+            "completed_at": self.completed_at,
+            "exit_code": self.exit_code,
+            "output": self.output,
+            "error": self.error,
+        }
+
+
+class TaskManager:
+    """
+    Non-blocking background task manager.
+
+    Runs shell commands and Python scripts asynchronously without
+    blocking the main Uvicorn thread.
+    """
+
+    def __init__(self):
+        self._tasks: dict[str, BackgroundTask] = {}
+        self._lock = threading.Lock()
+        self._running = True
+        # Start background poller thread
+        self._poller_thread = threading.Thread(
+            target=self._poll_loop,
+            daemon=True,
+            name="TaskManagerPoller"
+        )
+        self._poller_thread.start()
+        logger.info("TaskManager initialized with background poller thread")
+
+    def _poll_loop(self):
+        """Background thread that polls all running tasks."""
+        while self._running:
+            try:
+                with self._lock:
+                    for task in list(self._tasks.values()):
+                        if task.status == TaskStatus.RUNNING:
+                            task.read_output()
+                            task.poll()
+            except Exception as e:
+                logger.error(f"Error in task poll loop: {e}")
+            time.sleep(0.1)  # Poll every 100ms
+
+    def create_task(self, command: str, task_type: str = "shell") -> str:
+        """Create a new background task and start it."""
+        task_id = str(uuid.uuid4())[:8]
+        task = BackgroundTask(task_id, command, task_type)
+        task.start()
+
+        with self._lock:
+            self._tasks[task_id] = task
+
+        logger.info(f"Created task {task_id}: {command[:50]}...")
+        return task_id
+
+    def get_task(self, task_id: str) -> Optional[BackgroundTask]:
+        """Get a task by ID."""
+        with self._lock:
+            return self._tasks.get(task_id)
+
+    def list_tasks(self) -> list[dict[str, Any]]:
+        """List all tasks."""
+        with self._lock:
+            return [task.to_dict() for task in self._tasks.values()]
+
+    def cancel_task(self, task_id: str) -> bool:
+        """Cancel a running task."""
+        with self._lock:
+            task = self._tasks.get(task_id)
+            if task and task._process:
+                task._process.terminate()
+                task.status = TaskStatus.FAILED
+                task.completed_at = datetime.utcnow().isoformat() + "+00:00"
+                task.error = "Task cancelled by user"
+                logger.info(f"Cancelled task {task_id}")
+                return True
+        return False
+
+    def cleanup_old_tasks(self, max_age_hours: int = 24):
+        """Remove completed tasks older than max_age_hours."""
+        cutoff = datetime.utcnow().timestamp() - (max_age_hours * 3600)
+        with self._lock:
+            to_remove = []
+            for task_id, task in self._tasks.items():
+                if task.status in (TaskStatus.COMPLETED, TaskStatus.FAILED):
+                    if task.completed_at:
+                        try:
+                            completed_time = datetime.fromisoformat(
+                                task.completed_at.replace("+00:00", "").replace("Z", "")
+                            ).timestamp()
+                            if completed_time < cutoff:
+                                to_remove.append(task_id)
+                        except Exception:
+                            pass
+            for task_id in to_remove:
+                del self._tasks[task_id]
+            if to_remove:
+                logger.info(f"Cleaned up {len(to_remove)} old tasks")
+
+    def get_status(self) -> dict[str, Any]:
+        """Get overall task manager status."""
+        with self._lock:
+            status_counts = {}
+            for task in self._tasks.values():
+                status_counts[task.status] = status_counts.get(task.status, 0) + 1
+            return {
+                "total_tasks": len(self._tasks),
+                "status_breakdown": status_counts,
+                "poller_thread_alive": self._poller_thread.is_alive(),
+            }
+
+
+# ============================================================================
+# SHARED STATE (In-memory)
+# ============================================================================
+
+
+class SharedState:
+    """In-memory shared state for process status."""
+
+    def __init__(self):
+        self._lock = threading.Lock()
+        self._state = {
+            "worker_state": "initializing",
+            "worker_active": False,  # Set to True only after worker spawns successfully
+            "worker_pid": None,  # Set to worker PID after successful spawn
+            "worker_mode": None,  # Set after spawn
+            "last_heartbeat": None,
+            "heartbeat_age_seconds": 0,
+            "stage": "STARTUP_INIT",
+            "health": "INITIALIZING",
+            "error": None,
+            "uptime_seconds": 0,
+            "started_at": datetime.utcnow().isoformat() + "+00:00",
+        }
+        self._start_time = time.time()
+
+    def update(self, **kwargs):
+        """Update state with new values."""
+        with self._lock:
+            self._state["last_heartbeat"] = datetime.utcnow().isoformat() + "+00:00"
+            self._state["heartbeat_age_seconds"] = 0
+            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."""
+        with self._lock:
+            self._state["uptime_seconds"] = int(time.time() - self._start_time)
+
+            # 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()
+
+
+# ============================================================================
+# REQUEST LOGGING MIDDLEWARE
+# ============================================================================
+
+class RequestLoggingMiddleware(BaseHTTPMiddleware):
+    """
+    Middleware to log all incoming requests with method, path, and client IP.
+    Helps track internal state transitions for debugging.
+    """
+
+    async def dispatch(self, request: Request, call_next):
+        """Process request and log details."""
+        client_ip = request.client.host if request.client else "unknown"
+        method = request.method
+        path = request.url.path
+        query = str(request.url.query) if request.url.query else ""
+
+        logger.info(f"REQUEST: {method} {path}{'?' + query if query else ''} from {client_ip}")
+
+        # Process request
+        response = await call_next(request)
+
+        # Log response status
+        logger.info(f"RESPONSE: {method} {path} -> {response.status_code}")
+
+        return response
+
+
+# ============================================================================
+# FASTAPI APP
+# ============================================================================
+
+# Initialize shared state and task manager
+shared_state = SharedState()
+task_manager = TaskManager()
+
+
+def _check_dependencies() -> tuple[bool, str]:
+    """Pre-initialization check for PYTHONPATH and dependencies."""
+    errors = []
+
+    # Check PYTHONPATH
+    if "/app" not in sys.path:
+        errors.append("PYTHONPATH missing /app")
+
+    # Check critical imports
+    try:
+        import fastapi
+        import uvicorn
+    except ImportError as e:
+        errors.append(f"Missing dependency: {e}")
+
+    # Check brain_minimal.py exists
+    brain_path = Path(__file__).parent / "brain_minimal.py"
+    if not brain_path.exists():
+        errors.append(f"brain_minimal.py not found at {brain_path}")
+
+    return (len(errors) == 0, "; ".join(errors) if errors else "OK")
+
+
+async def _initialize_worker():
+    """
+    Background task to initialize worker process.
+    Spawns worker with retry logic and waits for heartbeat confirmation.
+    Runs after FastAPI server is ready to accept requests.
+    """
+    global _worker_manager
+
+    print("=" * 60)
+    print("🚀 CAIN WORKER INITIALIZATION: Starting async worker spawn")
+    print("=" * 60)
+    sys.stdout.flush()
+    sys.stderr.flush()
+
+    # Update state to show we're initializing worker
+    shared_state.update(
+        stage="RUNNING_APP_STARTING",
+        health="INITIALIZING_WORKER",
+        worker_active=False,
+        worker_pid=None
+    )
+
+    try:
+        # Initialize WorkerManager
+        _worker_manager = WorkerManager(shared_state)
+
+        # Spawn worker with extended timeout for HuggingFace Spaces environment
+        # Use 30 seconds heartbeat timeout instead of 15 for slower environments
+        success = await _worker_manager.spawn_with_retry(
+            max_attempts=5,  # More attempts for HF Spaces
+            retry_delay=2.0,
+            heartbeat_timeout=30.0  # Extended timeout
+        )
+
+        if success:
+            state = shared_state.get()
+            print("=" * 60)
+            print("✅ CAIN WORKER INITIALIZED SUCCESSFULLY")
+            print(f"   Stage: {state.get('stage')}")
+            print(f"   Health: {state.get('health')}")
+            print(f"   Worker PID: {state.get('worker_pid')}")
+            print(f"   Worker Active: {state.get('worker_active')}")
+            print("=" * 60)
+            logger.info("Cain worker initialized successfully")
+        else:
+            raise RuntimeError("Worker spawn returned False without exception")
+
+    except Exception as e:
+        logger.error(f"Worker initialization failed: {type(e).__name__}: {e}")
+        print(f"❌ CAIN WORKER INITIALIZATION FAILED: {e}")
+        sys.stdout.flush()
+        shared_state.update(
+            stage="RUNNING_APP_STARTING",
+            health="WORKER_FAILED",
+            error=f"Worker init failed: {e}",
+            worker_active=False,
+            worker_pid=None
+        )
+
+
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    """
+    Lifespan context manager for startup/shutdown.
+    Ensures worker initialization happens after FastAPI is ready.
+    """
+    # Startup: Start worker initialization in background
+    print("=" * 60)
+    print("🚀 CAIN LIFESPAN: Startup starting")
+    print("=" * 60)
+    sys.stdout.flush()
+
+    # Pre-initialization checks
+    deps_ok, deps_msg = _check_dependencies()
+    if not deps_ok:
+        logger.critical(f"DEPENDENCY CHECK FAILED: {deps_msg}")
+        print(f"CRITICAL: {deps_msg}")
+        sys.stdout.flush()
+        shared_state.update(
+            stage="RUNNING",
+            health="DEGRADED",  # Use DEGRADED instead of DEPENDENCY_ERROR
+            error=f"Pre-initialization check failed: {deps_msg}",
+            worker_active=False
+        )
+        # Don't raise - let app start in degraded mode
+        print("⚠️  Starting in DEGRADED mode (worker not available)")
+        sys.stdout.flush()
+    else:
+        print("✓ Pre-initialization checks passed")
+        sys.stdout.flush()
+
+        # Start worker initialization as background task
+        # This allows FastAPI to start accepting requests immediately
+        # while worker spawns in background
+        asyncio.create_task(_initialize_worker())
+
+    yield
+
+    # Shutdown: Cleanup
+    print("=" * 60)
+    print("🛑 CAIN LIFESPAN: Shutdown starting")
+    print("=" * 60)
+    sys.stdout.flush()
+
+    task_manager._running = False
+
+    # Shutdown worker manager
+    global _worker_manager
+    if _worker_manager is not None:
+        try:
+            await _worker_manager.shutdown()
+        except Exception as e:
+            logger.error(f"Error shutting down worker manager: {e}")
+        finally:
+            _worker_manager = None
+
+    print("CAIN shutdown complete")
+    sys.stdout.flush()
+
+
+app = FastAPI(
+    title="HuggingClaw Cain",
+    description="Agent collaboration server with background task execution",
+    version="2.0.0",
+    lifespan=lifespan
+)
+
+# DEBUG: Print APP_READY immediately to verify core app instantiation
+print("APP_READY")
+sys.stdout.flush()
+
+# TEMPORARILY DISABLED: Testing Eve's hypothesis - comment out error_handlers to break potential dependency chain
+# # Register exception handlers (import here to avoid circular dependency)
+# from error_handlers import register_error_handlers
+# register_error_handlers(app)
+#
+# # Add request logging middleware
+# app.add_middleware(RequestLoggingMiddleware)
+
+# Mount static files for frontend assets (images, fonts, etc.)
+app.mount("/static", StaticFiles(directory=str(FRONTEND_PATH)), name="static")
+
+
+@app.get("/")
+async def read_root():
+    """Serve the agent dashboard HTML."""
+    logger.info("GET / - Serving agent dashboard")
+
+    if AGENT_DASHBOARD.exists():
+        return FileResponse(str(AGENT_DASHBOARD), media_type="text/html")
+
+    # Fallback if dashboard file missing
+    return JSONResponse({
+        "status": "alive",
+        "message": "Cain core operational",
+        "note": "Agent dashboard file not found",
+        "frontend_path": str(FRONTEND_PATH),
+    })
+
+
+@app.get("/health")
+async def health_check():
+    """
+    Health check endpoint for container orchestration.
+
+    Returns:
+        - 200: System is operational (worker may still be initializing)
+        - Detailed worker status in response body
+    """
+    state = shared_state.get()
+    worker_alive = False
+    worker_pid = None
+
+    # Check live worker process status
+    if _worker_manager is not None and _worker_manager._process is not None:
+        worker_pid = _worker_manager._process.pid
+        returncode = _worker_manager._process.poll()
+        worker_alive = returncode is None  # None = still running
+
+    # Determine overall health
+    # App is healthy if it's running, worker can be initializing or degraded
+    is_healthy = state["health"] in ("HEALTHY", "INITIALIZING", "INITIALIZING_WORKER", "RUNNING_APP_STARTING", "DEGRADED")
+
+    return JSONResponse({
+        "status": "healthy" if is_healthy else "unhealthy",
+        "stage": state["stage"],
+        "worker": {
+            "active": state["worker_active"],
+            "alive": worker_alive,
+            "pid": worker_pid or state.get("worker_pid"),
+            "mode": state["worker_mode"],
+            "heartbeat_age_seconds": state["heartbeat_age_seconds"],
+        },
+        "uptime_seconds": state["uptime_seconds"],
+        "timestamp": datetime.utcnow().isoformat() + "+00:00",
+    }, status_code=200 if is_healthy else 503)
+
+
+@app.get("/api/state")
+async def get_state():
+    """
+    Get detailed process and system status.
+
+    Returns live worker process status (not just cached state).
+    """
+    logger.debug("GET /api/state called")
+
+    state = shared_state.get()
+    task_status = task_manager.get_status()
+
+    # Check live worker process status from WorkerManager
+    worker_alive = False
+    worker_pid = None
+
+    if _worker_manager is not None and _worker_manager._process is not None:
+        worker_pid = _worker_manager._process.pid
+        returncode = _worker_manager._process.poll()
+        worker_alive = returncode is None  # None = still running
+
+    # Update shared_state with real-time worker status
+    if worker_alive != state.get("worker_active"):
+        shared_state.update(worker_active=worker_alive)
+        state["worker_active"] = worker_alive
+
+    # Determine worker health - combine live process check with heartbeat freshness
+    worker_healthy = (
+        worker_alive and  # Must be actually running
+        state["heartbeat_age_seconds"] < 30  # And heartbeat must be recent
+    )
+
+    return JSONResponse({
+        "cain": {
+            "name": "Cain",
+            "space_id": "tao-shen/HuggingClaw-Cain",
+            "stage": state["stage"],
+            "health": state["health"],
+            "error": state.get("error"),
+            "uptime_seconds": state["uptime_seconds"],
+            "started_at": state["started_at"],
+        },
+        "worker": {
+            "state": state["worker_state"],
+            "active": state["worker_active"],  # Now reflects real-time process status
+            "pid": worker_pid or state.get("worker_pid"),
+            "mode": state["worker_mode"],
+            "last_heartbeat": state["last_heartbeat"],
+            "heartbeat_age_seconds": state["heartbeat_age_seconds"],
+            "is_healthy": worker_healthy,
+        },
+        "tasks": task_status,
+        "timestamp": datetime.utcnow().isoformat() + "+00:00",
+    })
+
+
+# Internal heartbeat endpoint (used by worker processes)
+@app.post("/internal/heartbeat")
+async def heartbeat(data: dict[str, Any]):
+    """Receive heartbeat from worker process."""
+    logger.info(f"[HEARTBEAT] Received from worker_pid={data.get('worker_pid')}, worker_state={data.get('worker_state')}, stage={data.get('stage')}")
+    print(f"[HEARTBEAT] Received from worker_pid={data.get('worker_pid')}, worker_state={data.get('worker_state')}, stage={data.get('stage')}")
+    sys.stdout.flush()
+
+    # Get current state before update
+    current = shared_state.get()
+    current_stage = current.get("stage")
+    incoming_stage = data.get("stage")
+
+    # If startup set RUNNING_APP_READY, allow worker to upgrade to RUNNING_A2A_READY
+    # but don't allow downgrades or unknown stages
+    if current_stage == "RUNNING_APP_READY" and incoming_stage == "RUNNING_A2A_READY":
+        print(f"[HEARTBEAT] Stage transition: {current_stage} -> {incoming_stage}")
+        shared_state.update(**data)
+    elif current_stage in ("RUNNING_APP_READY", "RUNNING_A2A_READY"):
+        # Preserve the app-ready stage, don't overwrite with worker's stage
+        # But update other fields
+        filtered_data = {k: v for k, v in data.items() if k != "stage"}
+        shared_state.update(**filtered_data)
+        print(f"[HEARTBEAT] Preserving stage={current_stage}, updating other fields")
+    else:
+        # Normal update for other stages
+        shared_state.update(**data)
+
+    sys.stdout.flush()
+    return {"status": "received"}
+
+
+@app.get("/read_startup_error")
+async def read_startup_error():
+    """
+    DEBUG: Read the startup error log to expose worker spawn failures.
+    This endpoint bypasses the silent failure loop by forcing errors to be visible.
+    """
+    startup_error_log = Path(__file__).parent / "startup_error.log"
+    if startup_error_log.exists():
+        content = startup_error_log.read_text()
+        return JSONResponse({
+            "startup_error_log_exists": True,
+            "content": content
+        })
+    return JSONResponse({
+        "startup_error_log_exists": False,
+        "message": "No startup error log found - worker may have started successfully"
+    })
+
+
+# ============================================================================
+# TASK EXECUTION API
+# ============================================================================
+
+@app.post("/api/tasks/execute")
+async def execute_task(request: dict[str, Any]):
+    """
+    Execute a command or script in the background.
+
+    Request body:
+    {
+        "command": "echo 'hello'",  # Command to execute
+        "type": "shell" | "python"  # Execution type (default: shell)
+    }
+    """
+    command = request.get("command")
+    task_type = request.get("type", "shell")
+
+    if not command:
+        raise HTTPException(status_code=400, detail="Command is required")
+
+    task_id = task_manager.create_task(command, task_type)
+    logger.info(f"Started task {task_id}: {command[:100]}")
+
+    return JSONResponse({
+        "task_id": task_id,
+        "status": "started",
+        "message": f"Task {task_id} started successfully"
+    })
+
+
+@app.get("/api/tasks/{task_id}")
+async def get_task_status(task_id: str):
+    """Get status of a specific task."""
+    task = task_manager.get_task(task_id)
+    if not task:
+        raise HTTPException(status_code=404, detail="Task not found")
+
+    # Poll for any updates
+    task.poll()
+
+    return JSONResponse(task.to_dict())
+
+
+@app.get("/api/tasks")
+async def list_tasks():
+    """List all tasks."""
+    return JSONResponse({"tasks": task_manager.list_tasks()})
+
+
+@app.delete("/api/tasks/{task_id}")
+async def cancel_task(task_id: str):
+    """Cancel a running task."""
+    if task_manager.cancel_task(task_id):
+        return JSONResponse({"status": "cancelled", "task_id": task_id})
+    raise HTTPException(status_code=404, detail="Task not found or not cancellable")
+
+
+@app.post("/api/tasks/cleanup")
+async def cleanup_tasks(max_age_hours: int = 24):
+    """Clean up old completed tasks."""
+    task_manager.cleanup_old_tasks(max_age_hours)
+    return JSONResponse({"status": "cleaned", "max_age_hours": max_age_hours})
+
+
+# ============================================================================
+# WEBSOCKET CHAT ENDPOINT
+# ============================================================================
+
+@app.websocket("/chat")
+async def chat_websocket(websocket: WebSocket):
+    """WebSocket endpoint for real-time agent communication."""
+    await websocket.accept()
+    client_id = f"{websocket.client.host}:{websocket.client.port}"
+    logger.info(f"WebSocket connection established from {client_id}")
+
+    try:
+        while True:
+            # Receive message from client
+            data = await websocket.receive_text()
+            logger.info(f"WebSocket message from {client_id}: {data}")
+
+            # Echo acknowledgment back to client
+            await websocket.send_json({
+                "type": "ack",
+                "message": "Message received",
+                "timestamp": datetime.utcnow().isoformat() + "+00:00",
+                "echo": data
+            })
+
+    except WebSocketDisconnect:
+        logger.info(f"WebSocket connection closed by {client_id}")
+    except Exception as e:
+        logger.error(f"WebSocket error for {client_id}: {e}")
+    finally:
+        logger.info(f"WebSocket connection terminated for {client_id}")
+
+
+# ============================================================================
+# MAIN ENTRY POINT
+# ============================================================================
+
+if __name__ == "__main__":
+    port = int(os.environ.get("PORT", 7860))
+    logger.info(f"🚀 STARTING UVICORN on 0.0.0.0:{port}")
+    uvicorn.run(app, host="0.0.0.0", port=port, log_level="info")