Fix #1813: restore post-consensus stall recovery wiring

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

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()