Add runtime container monitor and wave cycle cap

orchestrator/api.py

@@ -189,6 +189,23 @@ def main() -> None:
                 error=str(reconcile_err),
             )
 
+        # Start runtime container liveness monitor
+        try:
+            from container_monitor import (
+                create_pipeline_reconciliation_handler,
+                get_container_monitor,
+            )
+
+            monitor = get_container_monitor()
+            monitor.add_handler(create_pipeline_reconciliation_handler(repo_path))
+            monitor.start()
+            logger.info("Container monitor started for runtime liveness checks")
+        except Exception as monitor_err:
+            logger.warning(
+                "Container monitor startup failed",
+                error=str(monitor_err),
+            )
+
     if debug:
         # Use Flask's built-in server for development
         app.run(host=host, port=port, debug=True)

orchestrator/container_monitor.py

@@ -293,3 +293,144 @@ def get_container_monitor() -> ContainerMonitor:
     if _container_monitor is None:
         _container_monitor = ContainerMonitor()
     return _container_monitor
+
+
+def _reconcile_container_state(store: Any, container_info: ContainerInfo) -> bool:
+    """Update pipeline state for a single container that has exited.
+
+    Scans all RUNNING pipelines for a container matching the given
+    container_info and marks the container and its agent as FAILED.
+    If any changes are made, the pipeline itself is marked FAILED.
+
+    Uses per-pipeline locking (via ``get_pipeline_state_lock``) and
+    optimistic version checks (``expected_version``) to prevent race
+    conditions with concurrent state writers (e.g. agent signal handlers).
+
+    A container belongs to exactly one pipeline, so the function returns
+    after updating the first matching pipeline.
+
+    Args:
+        store: StateStore instance
+        container_info: Info about the exited/failed container
+
+    Returns:
+        True if any pipeline state was updated
+    """
+    from models import AgentExecutionStatus, PipelineStatus
+    from state_store import VersionConflictError, get_pipeline_state_lock
+
+    try:
+        pipeline_ids: list[str] = store.list_pipelines()
+    except Exception as e:
+        logger.warning(
+            "Runtime reconciliation: could not list pipelines",
+            error=str(e),
+        )
+        return False
+
+    for pipeline_id in pipeline_ids:
+        with get_pipeline_state_lock(pipeline_id):
+            try:
+                pipeline = store.load_pipeline(pipeline_id)
+            except Exception:
+                continue
+
+            if pipeline.status != PipelineStatus.RUNNING:
+                continue
+
+            changed = False
+
+            for phase_execution in pipeline.phases.values():
+                if phase_execution.status != PipelineStatus.RUNNING:
+                    continue
+
+                for ci in phase_execution.containers:
+                    if ci.container_id == container_info.container_id and ci.status == ContainerStatus.RUNNING:
+                        logger.warning(
+                            "Runtime reconciliation: container exited, marking FAILED",
+                            pipeline_id=pipeline_id,
+                            container_id=container_info.container_id[:12],
+                        )
+                        ci.status = ContainerStatus.FAILED
+                        ci.exit_code = container_info.exit_code if container_info.exit_code is not None else -1
+                        ci.exited_at = container_info.exited_at or datetime.utcnow()
+                        changed = True
+
+                for agent in phase_execution.agents:
+                    if (
+                        agent.status == AgentExecutionStatus.RUNNING
+                        and agent.container_id == container_info.container_id
+                    ):
+                        logger.warning(
+                            "Runtime reconciliation: agent container exited, marking FAILED",
+                            pipeline_id=pipeline_id,
+                            agent_role=str(agent.role),
+                            container_id=container_info.container_id[:12],
+                        )
+                        agent.status = AgentExecutionStatus.FAILED
+                        agent.completed_at = datetime.utcnow()
+                        agent.error = (
+                            "Container exited unexpectedly during execution — "
+                            "detected by runtime container monitor"
+                        )
+                        changed = True
+
+            if changed:
+                pipeline.status = PipelineStatus.FAILED
+                pipeline.error = (
+                    "Pipeline marked FAILED: agent container exited unexpectedly "
+                    "during execution. Restart via POST /pipelines/{id}/start."
+                )
+                try:
+                    store.save_pipeline(
+                        pipeline,
+                        expected_version=pipeline.version,
+                    )
+                    logger.warning(
+                        "Runtime reconciliation: pipeline marked FAILED",
+                        pipeline_id=pipeline_id,
+                    )
+                    return True
+                except VersionConflictError:
+                    logger.warning(
+                        "Runtime reconciliation: version conflict, skipping "
+                        "(concurrent writer updated pipeline)",
+                        pipeline_id=pipeline_id,
+                    )
+                    return False
+                except Exception as e:
+                    logger.error(
+                        "Runtime reconciliation: could not save pipeline",
+                        pipeline_id=pipeline_id,
+                        error=str(e),
+                    )
+                    return False
+
+    return False
+
+
+def create_pipeline_reconciliation_handler(repo_path: str) -> EventHandler:
+    """Create handler that updates pipeline state when containers exit.
+
+    The handler is invoked by the ContainerMonitor whenever a container
+    state change is detected. Only FAILED events (non-zero exit) trigger
+    reconciliation — STOPPED (exit code 0) represents a graceful exit
+    and should not mark pipelines as failed.
+
+    Args:
+        repo_path: Path to the repository (for StateStore access)
+
+    Returns:
+        Event handler function
+    """
+
+    def handler(event: ContainerEvent) -> None:
+        if event.event_type != ContainerEvent.FAILED:
+            return
+
+        from state_store import get_state_store
+
+        store = get_state_store(repo_path)
+        _reconcile_container_state(store, event.container_info)
+
+    return handler

orchestrator/multi_agent.py

@@ -510,17 +510,28 @@ def execute_all_waves(
         self,
         on_wave_complete: Callable[[AgentWave], None] | None = None,
         agent_prompts: dict[AgentRole, str] | None = None,
+        max_waves: int = 5,
     ) -> list[AgentWave]:
         """Execute all waves until completion or failure.
 
         Args:
             on_wave_complete: Optional callback after each wave
             agent_prompts: Role-to-prompt mapping (required when using spawn_fn)
+            max_waves: Safety cap on number of wave iterations (default: 5)
 
         Returns:
             List of completed waves
         """
+        waves_executed = 0
         while True:
+            if waves_executed >= max_waves:
+                logger.warning(
+                    "Max waves reached, stopping execution",
+                    pipeline_id=self.pipeline.id,
+                    max_waves=max_waves,
+                )
+                break
+
             wave = self.get_next_wave()
             if not wave:
                 break
@@ -529,6 +540,7 @@ def execute_all_waves(
                 wave,
                 agent_prompts=agent_prompts,
             )
+            waves_executed += 1
 
             if on_wave_complete:
                 on_wave_complete(completed)