fix: Handle orphaned container state on orchestrator restart

orchestrator/api.py

@@ -173,6 +173,22 @@ def main() -> None:
         host_repo_map=host_repo_map,
     )
 
+    # Reconcile any pipelines left in RUNNING state from a previous crash.
+    if repo_path != "not set":
+        try:
+            from docker_client import get_docker_client
+            from startup_reconciliation import reconcile_stale_containers
+            from state_store import get_state_store
+
+            recovered = reconcile_stale_containers(get_state_store(repo_path), get_docker_client())
+            if recovered:
+                logger.warning("Recovered stale pipelines on startup", count=recovered)
+        except Exception as reconcile_err:
+            logger.warning(
+                "Startup reconciliation failed",
+                error=str(reconcile_err),
+            )
+
     if debug:
         # Use Flask's built-in server for development
         app.run(host=host, port=port, debug=True)

orchestrator/routes/pipelines.py

@@ -34,7 +34,7 @@ def get_logger(name: str, **kwargs) -> logging.Logger:  # type: ignore[misc]
 try:
     from ..container_spawner import ContainerSpawnError, get_container_spawner
     from ..decision_queue import get_decision_queue
-    from ..docker_client import DockerClientError
+    from ..docker_client import ContainerNotFoundError, ContainerOperationError, DockerClientError
     from ..models import (
         AgentRole,
         CycleTiming,
@@ -55,7 +55,11 @@ def get_logger(name: str, **kwargs) -> logging.Logger:  # type: ignore[misc]
 except ImportError:
     from container_spawner import ContainerSpawnError, get_container_spawner  # type: ignore
     from decision_queue import get_decision_queue  # type: ignore
-    from docker_client import DockerClientError  # type: ignore
+    from docker_client import (  # type: ignore
+        ContainerNotFoundError,
+        ContainerOperationError,
+        DockerClientError,
+    )
     from models import (  # type: ignore
         AgentRole,
         ComplexityTier,
@@ -3368,10 +3372,24 @@ def _spawn_and_wait(
             )
 
     docker_client = spawner.docker
-    final_info = docker_client.wait_for_container(
-        spawned.container_info.container_id,
-        timeout=timeout,
-    )
+    try:
+        final_info = docker_client.wait_for_container(
+            spawned.container_info.container_id,
+            timeout=timeout,
+        )
+    except (ContainerNotFoundError, ContainerOperationError) as e:
+        logger.warning(
+            "Container lost during wait, marking failed",
+            container_id=spawned.container_info.container_id,
+            error=str(e),
+        )
+        final_info = ContainerInfo(
+            container_id=spawned.container_info.container_id,
+            container_name=spawned.container_info.container_name,
+            status=ContainerStatus.FAILED,
+            exit_code=-1,
+            exited_at=datetime.utcnow(),
+        )
 
     container_logs = ""
     if final_info.exit_code != 0:
@@ -3395,8 +3413,8 @@ def _spawn_and_wait(
                 # Update container status
                 for ci in phase_execution.containers:
                     if ci.container_id == spawned.container_info.container_id:
-                        ci.status = ContainerStatus.EXITED
-                        ci.exited_at = datetime.utcnow()
+                        ci.status = final_info.status
+                        ci.exited_at = final_info.exited_at
                         ci.exit_code = final_info.exit_code
                         break
 

orchestrator/startup_reconciliation.py

@@ -0,0 +1,145 @@
+"""
+Startup reconciliation for orphaned container state.
+
+On restart, persisted RUNNING agents whose containers are no longer alive
+are detected and marked FAILED so operators (or CI) can retry via the
+existing POST /pipelines/{id}/start endpoint.
+"""
+
+import sys
+from datetime import datetime
+from pathlib import Path
+
+# Add shared directory to path for egg_logging
+_shared_path = Path(__file__).parent.parent / "shared"
+if _shared_path.exists() and str(_shared_path) not in sys.path:
+    sys.path.insert(0, str(_shared_path))
+
+try:
+    from egg_logging import get_logger
+except ImportError:
+    import logging
+
+    def get_logger(name: str, **kwargs) -> logging.Logger:  # type: ignore[misc]
+        return logging.getLogger(name)
+
+
+logger = get_logger("orchestrator.startup_reconciliation")
+
+
+def reconcile_stale_containers(store: object, docker_client: object) -> int:
+    """Detect and recover pipelines whose running containers are gone.
+
+    Called once at orchestrator startup before serving requests.  For each
+    pipeline that shows status=RUNNING, any agent/container whose container_id
+    is absent from the live Docker container set is marked FAILED.  If at
+    least one such stale entry is found the pipeline itself is marked FAILED
+    so that operators can restart it via POST /pipelines/{id}/start.
+
+    Args:
+        store: StateStore instance (already bound to the correct repo path).
+        docker_client: DockerClient instance.
+
+    Returns:
+        Number of pipelines that were recovered (marked FAILED).
+    """
+    try:
+        from models import AgentExecutionStatus, ContainerStatus, PipelineStatus
+    except ImportError:
+        from models import AgentExecutionStatus, ContainerStatus, PipelineStatus  # type: ignore
+
+    # Collect IDs of containers that are currently running in Docker.
+    try:
+        live_containers = docker_client.list_containers(all=False)  # type: ignore[attr-defined]
+        live_ids: set[str] = {ci.container_id for ci in live_containers}
+    except Exception as e:
+        logger.warning(
+            "Startup reconciliation skipped: could not list live containers",
+            error=str(e),
+        )
+        return 0
+
+    try:
+        pipeline_ids: list[str] = store.list_pipelines()  # type: ignore[attr-defined]
+    except Exception as e:
+        logger.warning(
+            "Startup reconciliation skipped: could not list pipelines",
+            error=str(e),
+        )
+        return 0
+
+    recovered = 0
+
+    for pipeline_id in pipeline_ids:
+        try:
+            pipeline = store.load_pipeline(pipeline_id)  # type: ignore[attr-defined]
+        except Exception as e:
+            logger.warning(
+                "Startup reconciliation: could not load pipeline",
+                pipeline_id=pipeline_id,
+                error=str(e),
+            )
+            continue
+
+        if pipeline.status != PipelineStatus.RUNNING:
+            continue
+
+        changed = False
+
+        for phase_key, phase_execution in pipeline.phases.items():
+            if phase_execution.status != PipelineStatus.RUNNING:
+                continue
+
+            for container_info in phase_execution.containers:
+                if container_info.status == ContainerStatus.RUNNING:
+                    if container_info.container_id not in live_ids:
+                        logger.warning(
+                            "Startup reconciliation: container missing, marking FAILED",
+                            pipeline_id=pipeline_id,
+                            phase=phase_key,
+                            container_id=container_info.container_id,
+                        )
+                        container_info.status = ContainerStatus.FAILED
+                        container_info.exit_code = -1
+                        container_info.exited_at = datetime.utcnow()
+                        changed = True
+
+            for agent in phase_execution.agents:
+                if agent.status == AgentExecutionStatus.RUNNING:
+                    if agent.container_id and agent.container_id not in live_ids:
+                        logger.warning(
+                            "Startup reconciliation: agent container missing, marking FAILED",
+                            pipeline_id=pipeline_id,
+                            phase=phase_key,
+                            agent_role=str(agent.role),
+                            container_id=agent.container_id,
+                        )
+                        agent.status = AgentExecutionStatus.FAILED
+                        agent.completed_at = datetime.utcnow()
+                        agent.error = (
+                            "Container not found at orchestrator startup — "
+                            "likely lost during a previous crash"
+                        )
+                        changed = True
+
+        if changed:
+            pipeline.status = PipelineStatus.FAILED
+            pipeline.error = (
+                "Pipeline marked FAILED at orchestrator startup: one or more agent "
+                "containers were not found. Restart via POST /pipelines/{id}/start."
+            )
+            try:
+                store.save_pipeline(pipeline)  # type: ignore[attr-defined]
+                recovered += 1
+                logger.warning(
+                    "Startup reconciliation: pipeline marked FAILED",
+                    pipeline_id=pipeline_id,
+                )
+            except Exception as e:
+                logger.warning(
+                    "Startup reconciliation: could not save pipeline",
+                    pipeline_id=pipeline_id,
+                    error=str(e),
+                )
+
+    return recovered