Fix restart_phase wedge on sliced implement phases

orchestrator/routes/pipelines.py

@@ -185,6 +185,7 @@ def get_repo_checks(repo: str) -> list[dict[str, str]]:  # type: ignore[misc]
     )
     from ..kubernetes_client import (
         LABEL_PIPELINE_ID,
+        LABEL_SLICE_ID,
         JobOperationError,
         KubernetesClientError,
         PodNotFoundError,
@@ -239,6 +240,7 @@ def get_repo_checks(repo: str) -> list[dict[str, str]]:  # type: ignore[misc]
     )
     from kubernetes_client import (  # type: ignore
         LABEL_PIPELINE_ID,
+        LABEL_SLICE_ID,
         JobOperationError,
         KubernetesClientError,
         PodNotFoundError,
@@ -1013,6 +1015,45 @@ def _count_live_pods_for_pipeline(pipeline_id: str, *, quiet: bool = False) -> i
         return None
 
 
+def _slice_agents_alive(spawner: Any, pipeline_id: str, slice_id: str) -> bool:
+    """Check if any live agents exist for a slice (#2914).
+
+    Returns ``True`` if at least one pod labeled with the pipeline and
+    slice IDs is in a live state (Pending/Creating/Running). Returns
+    ``False`` if zero live pods or if the label query fails — the
+    conservative default forces re-spawn rather than risking a wedge.
+
+    Caller contract: callers must have already torn down stale cohorts
+    with foreground propagation (e.g. ``restart_phase`` step 4 calls
+    ``remove_agent_container(force=True)``). A pod whose Job is being
+    deleted but is still in its termination grace period still reports
+    ``phase=Running`` (``kubernetes_client.py`` maps Running → RUNNING
+    without a Terminating-specific status), so without foreground
+    teardown the helper can false-positive against terminating pods
+    and wedge again. The ``spawner`` is taken as a parameter (rather
+    than fetched via ``_get_spawner``) so tests can inject a stub
+    directly, paralleling how ``_classify_non_complete_slice``
+    receives ``gateway``.
+    """
+    try:
+        pods = spawner.backend.list_containers(
+            labels={
+                LABEL_PIPELINE_ID: pipeline_id,
+                LABEL_SLICE_ID: slice_id,
+            },
+        )
+        live_count = sum(1 for p in pods if p.status in _LIVE_POD_STATUSES)
+        return live_count > 0
+    except Exception as e:  # noqa: BLE001
+        logger.warning(
+            "Slice liveness check failed; treating as not-alive to force re-spawn (#2914)",
+            pipeline_id=pipeline_id,
+            slice_id=slice_id,
+            error=str(e),
+        )
+        return False
+
+
 def _guard_live_pods_or_force(
     pipeline_id: str,
     force: bool,
@@ -16085,6 +16126,7 @@ def _bootstrap_check_one(slice_obj: Any) -> tuple[str, bool]:
     bootstrap_consensus_complete: list[str] = []
     bootstrap_blocked: list[str] = []
     bootstrap_corrupt: list[str] = []
+    bootstrap_reclassified_fresh: list[str] = []  # resume-but-dead → fresh (#2914)
     layer_b_marked_complete = set(bootstrap_merged)
     for s in layer_b_candidates:
         if s.id in layer_b_marked_complete:
@@ -16117,8 +16159,21 @@ def _bootstrap_check_one(slice_obj: Any) -> tuple[str, bool]:
             bootstrap_consensus_complete.append(s.id)
             continue
         if classification == "resume":
-            scheduler.mark_spawned(s.id)
-            bootstrap_resumed.append(s.id)
+            # Verify agents are actually live before marking as spawned (#2914).
+            # On restart_phase, agents were torn down but contract still shows
+            # IN_PROGRESS with commits — we must not mark_spawned when cohort
+            # is absent, or the pipeline wedges with no agents running.
+            if _slice_agents_alive(spawner, pipeline_id, s.id):
+                scheduler.mark_spawned(s.id)
+                bootstrap_resumed.append(s.id)
+            else:
+                logger.warning(
+                    "Layer-C resume classification but no live agents; "
+                    "treating as fresh to force re-spawn (#2914)",
+                    pipeline_id=pipeline_id,
+                    slice_id=s.id,
+                )
+                bootstrap_reclassified_fresh.append(s.id)
             continue
         if classification == "blocked":
             bootstrap_blocked.append(s.id)
@@ -16135,20 +16190,33 @@ def _bootstrap_check_one(slice_obj: Any) -> tuple[str, bool]:
             already_complete_on_contract=bootstrap_complete,
             detected_merged_on_origin=bootstrap_merged,
         )
-    if bootstrap_resumed or bootstrap_consensus_complete or bootstrap_blocked or bootstrap_corrupt:
+    if (
+        bootstrap_resumed
+        or bootstrap_consensus_complete
+        or bootstrap_blocked
+        or bootstrap_corrupt
+        or bootstrap_reclassified_fresh
+    ):
         # NOTE: include ``bootstrap_blocked`` in the gate (reviewer_code
         # v3 NACK fix) — a bootstrap pass whose only Layer-C activity is
         # BLOCKED slices was previously suppressing the audit-trail line
         # entirely. Case-4 escalation still fires, but operators need
         # the structured "we saw a blocked slice" log to spot
         # pending-HITL backlogs without grepping for the side-effect.
+        #
+        # Also include ``bootstrap_reclassified_fresh`` (#2914) — resume-
+        # classified slices that were re-verified against k8s and found
+        # to have no live agents. Surfacing the reclassification here
+        # gives operators a structured audit trail for the
+        # ``restart_phase``-recovery path.
         logger.info(
             "Slice bootstrap reconciliation classified non-COMPLETE slices (slice-4 TASK-4-4)",
             pipeline_id=pipeline_id,
             resumed=bootstrap_resumed,
             consensus_complete_unrecorded=bootstrap_consensus_complete,
             blocked=bootstrap_blocked,
             corrupt=bootstrap_corrupt,
+            reclassified_fresh=bootstrap_reclassified_fresh,
         )
     # Case 5 — escalate via HITL so the pipeline pauses until the
     # operator picks an option (reviewer_contract / reviewer_code v1

orchestrator/tests/test_slice_4_restart_hardening.py

@@ -1093,6 +1093,147 @@ def test_probe_failure_classifies_fresh(self) -> None:
         assert result == "fresh"
 
 
+# ---------------------------------------------------------------------------
+# #2914: _slice_agents_alive() — k8s probe for restart-phase resume guard
+#
+# The fix for #2914 adds a runtime check that prevents the bootstrap
+# reconciler from calling scheduler.mark_spawned() when no live agents
+# exist. Without this, restart_phase on a sliced implement wedges the
+# pipeline: the scheduler thinks the slice is RUNNING but no containers
+# are present, so no signals can arrive and the slice never completes.
+#
+# This helper must be defensive:
+# - Returns False (force fresh re-spawn) on any k8s API error
+# - Returns False when zero pods match the slice labels
+# - Returns True only when at least one pod is in a live state
+# - Filters by both pipeline_id AND slice_id labels (not just pipeline)
+# ---------------------------------------------------------------------------
+
+
+class TestSliceAgentsAlive:
+    """Exercise _slice_agents_alive() against a stubbed spawner backend.
+
+    The helper takes ``spawner`` as a parameter (paralleling how
+    ``_classify_non_complete_slice`` takes ``gateway``) so tests inject
+    a stub directly without patching ``routes.pipelines._get_spawner``.
+    """
+
+    @staticmethod
+    def _make_container_info(container_id: str, status):
+        from models import ContainerInfo
+
+        return ContainerInfo(
+            container_id=container_id,
+            container_name=f"egg-{container_id}",
+            status=status,
+        )
+
+    def _make_spawner(self, returned_pods):
+        """Build a spawner stub whose backend.list_containers yields
+        the given pods."""
+        backend = MagicMock()
+        backend.list_containers.return_value = returned_pods
+        spawner = MagicMock()
+        spawner.backend = backend
+        return spawner
+
+    def test_true_when_running_pod_exists(self):
+        """At least one RUNNING pod → slice is live, resume is safe."""
+        from models import ContainerStatus
+        from routes.pipelines import _slice_agents_alive
+
+        pods = [
+            self._make_container_info("p1", ContainerStatus.RUNNING),
+        ]
+
+        spawner = self._make_spawner(pods)
+        assert _slice_agents_alive(spawner, "pipeline-x", "slice-1") is True
+
+    def test_true_when_pending_pod_exists(self):
+        """PENDING pod (still scheduling) → slice is live, don't re-spawn."""
+        from models import ContainerStatus
+        from routes.pipelines import _slice_agents_alive
+
+        pods = [
+            self._make_container_info("p1", ContainerStatus.PENDING),
+        ]
+
+        spawner = self._make_spawner(pods)
+        assert _slice_agents_alive(spawner, "pipeline-x", "slice-1") is True
+
+    def test_true_when_creating_pod_exists(self):
+        """CREATING pod (Job→Pod transition) → slice is live, don't re-spawn.
+
+        ``_LIVE_POD_STATUSES`` (``models.LIVE_POD_STATUSES``) includes
+        CREATING because k8s Jobs pass through it on their way to
+        Running. Without this branch, a slice mid-spawn would be
+        misclassified as dead and double-spawned. (reviewer suggestion 2
+        on #2916: same shape as the RUNNING/PENDING tests.)
+        """
+        from models import ContainerStatus
+        from routes.pipelines import _slice_agents_alive
+
+        pods = [
+            self._make_container_info("p1", ContainerStatus.CREATING),
+        ]
+
+        spawner = self._make_spawner(pods)
+        assert _slice_agents_alive(spawner, "pipeline-x", "slice-1") is True
+
+    def test_false_when_no_pods(self):
+        """Zero pods → slice is dead, force fresh re-spawn."""
+        from routes.pipelines import _slice_agents_alive
+
+        spawner = self._make_spawner([])
+        assert _slice_agents_alive(spawner, "pipeline-x", "slice-1") is False
+
+    def test_false_when_only_terminal_pods(self):
+        """Only EXITED/FAILED pods (post-restart_phase cleanup) → slice is dead."""
+        from models import ContainerStatus
+        from routes.pipelines import _slice_agents_alive
+
+        pods = [
+            self._make_container_info("p1", ContainerStatus.EXITED),
+            self._make_container_info("p2", ContainerStatus.FAILED),
+        ]
+
+        spawner = self._make_spawner(pods)
+        assert _slice_agents_alive(spawner, "pipeline-x", "slice-1") is False
+
+    def test_false_on_k8s_api_error(self):
+        """Defensive: k8s API error → assume dead, force re-spawn."""
+        from routes.pipelines import _slice_agents_alive
+
+        backend = MagicMock()
+        backend.list_containers.side_effect = RuntimeError("k8s unreachable")
+        spawner = MagicMock()
+        spawner.backend = backend
+
+        assert _slice_agents_alive(spawner, "pipeline-x", "slice-1") is False
+
+    def test_filters_by_pipeline_and_slice_labels(self):
+        """Helper must query with both labels to avoid false-positive on
+        a different slice in the same pipeline."""
+        from models import ContainerStatus
+        from routes.pipelines import _slice_agents_alive
+
+        backend = MagicMock()
+        backend.list_containers.return_value = [
+            self._make_container_info("p1", ContainerStatus.RUNNING),
+        ]
+        spawner = MagicMock()
+        spawner.backend = backend
+
+        _slice_agents_alive(spawner, "pipeline-x", "slice-2")
+
+        # Verify the label selector included both pipeline and slice
+        call_kwargs = backend.list_containers.call_args.kwargs
+        assert "labels" in call_kwargs
+        labels = call_kwargs["labels"]
+        assert labels["egg.pipeline.id"] == "pipeline-x"
+        assert labels["egg.slice.id"] == "slice-2"
+
+
 class TestSliceHasPendingDecision:
     """``_slice_has_pending_decision`` — the helper that detects a
     BLOCKED slice without any pending HITL on the contract (case-4

orchestrator/tests/test_slice_run_loop_integration.py

@@ -1289,6 +1289,115 @@ def test_bootstrap_detection_failure_falls_through(self) -> None:
         spawner.gateway.create_slice_pr.assert_called_once()
 
 
+# ---------------------------------------------------------------------------
+# #2914 — restart_phase resume guard: bootstrap must verify pods are alive
+# before mark_spawned, or the pipeline wedges with no agents running.
+# ---------------------------------------------------------------------------
+
+
+class TestBootstrapResumeAliveGuard:
+    """#2914 — Layer-C resume classification must re-verify against k8s.
+
+    The bootstrap classifier returns ``"resume"`` for IN_PROGRESS slices
+    with commits on the integration branch. Without the alive-guard,
+    that flips the scheduler slice to RUNNING via ``mark_spawned`` —
+    correct under normal restart, but wedging after ``restart_phase``
+    tore the container cohort down (contract still shows IN_PROGRESS +
+    commits but no pods exist). The guard re-queries the spawner; if
+    no live pods are found the slice is treated as fresh so the run
+    loop re-yields READY and respawns a new cohort.
+
+    These tests close the integration gap that
+    ``TestSliceAgentsAlive`` (unit) cannot — exercising the actual
+    call site at ``routes/pipelines.py:_bootstrap_check_one``'s
+    Layer-C ``resume`` branch end-to-end through
+    ``_run_implement_phase_slices``.
+    """
+
+    def _make_spawner(self, *, live_pods: list[Any]) -> MagicMock:
+        """Build a spawner whose backend.list_containers returns the
+        given pods and whose gateway probe reports the integration
+        branch has commits on origin (classifier → ``"resume"``)."""
+        spawner = MagicMock()
+        spawner.gateway = MagicMock()
+        spawner.gateway.create_slice_pr.return_value = "https://example/pr/1"
+        spawner.gateway.is_slice_branch_merged_into_parent.return_value = False
+        # Non-None SHA → classifier sees commits on origin → "resume".
+        spawner.gateway.get_remote_branch_sha.return_value = "deadbeef" * 5
+        backend = MagicMock()
+        backend.list_containers.return_value = live_pods
+        spawner.backend = backend
+        return spawner
+
+    @staticmethod
+    def _make_container_info(container_id: str, status: Any) -> Any:
+        from models import ContainerInfo
+
+        return ContainerInfo(
+            container_id=container_id,
+            container_name=f"egg-{container_id}",
+            status=status,
+        )
+
+    def test_resume_with_no_live_pods_respawns_fresh(self) -> None:
+        """Classifier → ``"resume"`` but ``list_containers`` returns []
+        → slice must run through the regular path (i.e.
+        ``_run_concurrent_phase`` IS called). Without the guard,
+        ``mark_spawned`` would have flipped the slice to RUNNING and
+        ``iter_ready`` would never re-yield it, leaving the pipeline
+        wedged."""
+        pipeline = _make_pipeline()
+        slice1 = _make_slice("slice-1", tasks=[_make_task("task-1-1")])
+        slice1.status = SliceStatus.IN_PROGRESS
+        contract = _make_contract(slices=[slice1])
+
+        with (
+            patch("egg_contracts.loader.load_contract", return_value=contract),
+            patch("egg_contracts.loader.save_contract"),
+            patch("routes.pipelines._start_stacked_pr_reconciler") as mock_start_recon,
+            patch(
+                "routes.pipelines._run_concurrent_phase", return_value=(0, "ok")
+            ) as mock_run_phase,
+            patch("orchestrator.peer_consensus.remove_peer_consensus_tracker"),
+            # No tracker → classifier's consensus_complete=False
+            # branch → "resume".
+            patch(
+                "routes.pipelines._lookup_peer_consensus_tracker_or_none",
+                return_value=None,
+            ),
+        ):
+            mock_start_recon.return_value = (MagicMock(), threading.Event())
+            spawner = self._make_spawner(live_pods=[])
+            exit_code, _ = _run_implement_phase_slices(
+                pipeline_id=pipeline.id,
+                pipeline=pipeline,
+                spawner=spawner,
+                repo_volumes={},
+                gateway_mode="public",
+                repos=["owner/repo"],
+                sandbox_env={},
+                store=MagicMock(),
+                certs_volume=None,
+                worktree_repo_path=Path("/tmp/x"),
+            )
+        assert exit_code == 0
+        # The fix: with no live pods, the slice must still spawn fresh.
+        invoked = {c.kwargs["slice_id"] for c in mock_run_phase.call_args_list}
+        assert invoked == {"slice-1"}, (
+            "resume-classified slice with no live pods must be respawned — "
+            "without the #2914 guard, mark_spawned would wedge the pipeline"
+        )
+        # Liveness probe must have actually fired for the slice (label
+        # selector includes the slice id, not just the pipeline).
+        list_calls = spawner.backend.list_containers.call_args_list
+        slice_labelled_calls = [
+            c for c in list_calls if c.kwargs.get("labels", {}).get("egg.slice.id") == "slice-1"
+        ]
+        assert slice_labelled_calls, (
+            "_slice_agents_alive must call list_containers with the slice label"
+        )
+
+
 # ---------------------------------------------------------------------------
 # Coder fixes for reviewer_code_holistic v1 NACK (now regression guards)
 # ---------------------------------------------------------------------------