diff --git a/orchestrator/kubernetes_monitor.py b/orchestrator/kubernetes_monitor.py index 3ac23d6fe7..15ab2cf1b8 100644 --- a/orchestrator/kubernetes_monitor.py +++ b/orchestrator/kubernetes_monitor.py @@ -215,8 +215,11 @@ def _check_pod(self, pod_info: ContainerInfo) -> None: def _run_runtime_tick_checks(self) -> None: """Fire RUNTIME_TICK health checks on all running pipelines. - Called when container state changes are detected. Requires that - ``set_health_check_runner`` has been called to wire the runner. + Called from ``_check_pod`` on container state changes + and from ``_reconciliation_sweep`` on the periodic interval so the + post-consensus stall recovery still runs for pipelines where no + pods are transitioning. Requires that ``set_health_check_runner`` + has been called to wire the runner. """ runner = getattr(self, "_health_check_runner", None) if runner is None: @@ -246,7 +249,8 @@ def _run_runtime_tick_checks(self) -> None: docker_client=self.k8s_client, state_store=store, ) - runner.run(ctx, HealthTrigger.RUNTIME_TICK) + results = runner.run(ctx, HealthTrigger.RUNTIME_TICK) + self._handle_consensus_stall_recovery(results, pipeline, store) except Exception as e: logger.debug( "RUNTIME_TICK check failed for pipeline", @@ -549,6 +553,13 @@ def _reconciliation_sweep(self) -> None: agent_role=str(agent.role), ) + # Fire RUNTIME_TICK checks every sweep so pipelines where no pods + # are transitioning (e.g. all agents quietly polling post-BRC + # consensus) still exercise the consensus-stall recovery path. + # Without this, a stuck post-consensus pipeline never recovers + # because _check_pod is the only other call site. (#1813) + self._run_runtime_tick_checks() + def stop(self) -> None: """Stop the monitor and periodic reconciliation.""" stopped_any = False diff --git a/orchestrator/tests/test_kubernetes_monitor.py b/orchestrator/tests/test_kubernetes_monitor.py index e786155937..9f446b678d 100644 --- a/orchestrator/tests/test_kubernetes_monitor.py +++ b/orchestrator/tests/test_kubernetes_monitor.py @@ -1040,3 +1040,92 @@ def test_terminal_status_without_exited_at_not_reconciled(self, monitor, mock_k8 monitor._reconciliation_sweep() store.save_pipeline.assert_not_called() + + +# --------------------------------------------------------------------------- +# TestRuntimeTickConsensusStallWiring (#1813) +# --------------------------------------------------------------------------- + + +class TestRuntimeTickConsensusStallWiring: + """Regression tests for #1813: RUNTIME_TICK results must drive stall recovery. + + #1692 (Docker→k8s migration) dropped the call that forwarded health-check + results to _handle_consensus_stall_recovery and moved the only trigger from + the per-poll cycle to pod state transitions only. A pipeline stuck in + post-consensus with no pods transitioning was left with no recovery path. + """ + + def _make_running_pipeline(self): + from models import ( + Pipeline, + PipelinePhase, + PipelineStatus, + ) + + pipeline = Pipeline( + id="issue-1759-v3", + issue_number=1759, + repo="owner/repo", + status=PipelineStatus.RUNNING, + current_phase=PipelinePhase.PLAN, + phases={}, + ) + store = MagicMock() + store.list_pipelines.return_value = [pipeline.id] + store.load_pipeline.return_value = pipeline + store.repo_path = "/tmp/repo" + return pipeline, store + + def test_runtime_tick_forwards_results_to_stall_recovery(self, monitor): + """_run_runtime_tick_checks must pass runner results to the recovery handler.""" + pipeline, store = self._make_running_pipeline() + monitor._reconciliation_stores = [store] + + mock_results = [MagicMock(check_name="consensus_stall")] + mock_runner = MagicMock() + mock_runner.run.return_value = mock_results + monitor.set_health_check_runner(mock_runner) + + with patch.object(monitor, "_handle_consensus_stall_recovery") as mock_recovery: + monitor._run_runtime_tick_checks() + + mock_recovery.assert_called_once() + args = mock_recovery.call_args.args + assert args[0] is mock_results + assert args[1] is pipeline + assert args[2] is store + + def test_runtime_tick_skips_non_running_pipelines(self, monitor): + """Pipelines that are not RUNNING must not be passed to the recovery handler.""" + from models import PipelineStatus + + pipeline, store = self._make_running_pipeline() + pipeline.status = PipelineStatus.COMPLETE + monitor._reconciliation_stores = [store] + + mock_runner = MagicMock() + mock_runner.run.return_value = [] + monitor.set_health_check_runner(mock_runner) + + with patch.object(monitor, "_handle_consensus_stall_recovery") as mock_recovery: + monitor._run_runtime_tick_checks() + + mock_runner.run.assert_not_called() + mock_recovery.assert_not_called() + + def test_reconciliation_sweep_invokes_runtime_tick(self, monitor, mock_k8s_client): + """_reconciliation_sweep must fire _run_runtime_tick_checks every sweep. + + Previously only _check_pod called it; this left pipelines + with no pod churn unable to exercise the stall-recovery path. + """ + mock_k8s_client.list_containers.return_value = [] + mock_k8s_client.list_jobs.return_value = [] + + # No reconciliation stores needed — we're asserting the hook fires + # regardless of pipeline content. + with patch.object(monitor, "_run_runtime_tick_checks") as mock_tick: + monitor._reconciliation_sweep() + + mock_tick.assert_called_once()