Fix #2411: tolerant startup reconciliation + start_pipeline MCP verb

docs/reference/orchestrator-cli.md

@@ -234,10 +234,11 @@ egg-orch anchor validate
 
 ## Phase Management MCP Tools
 
-Four MCP tools expose phase management operations for pipeline recovery and manual intervention, eliminating the need for raw `curl` calls during stuck pipeline scenarios (see [#1570](https://github.com/jwbron/egg/issues/1570) for motivation).
+Five MCP tools expose phase- and pipeline-level recovery operations, eliminating the need for raw `curl` calls during stuck pipeline scenarios (see [#1570](https://github.com/jwbron/egg/issues/1570) for motivation; pipeline-level recovery added in [#2411](https://github.com/jwbron/egg/issues/2411)).
 
 | MCP Tool | REST Endpoint | Description |
 |----------|---------------|-------------|
+| `start_pipeline` | `POST /pipelines/{id}/start` | Recover a non-RUNNING pipeline (FAILED, AWAITING_HUMAN with all decisions resolved, or PENDING — the route has no early-return for PENDING). **Unconditionally** resets the current phase to PENDING (clears `containers`, `agents`, `artifacts` regardless of whether the records are verifiably stale), bumps `run_epoch`, sets `pipeline.status = RUNNING`, and re-launches the `_run_pipeline` thread. **Distinct from `start_phase`** — targets pipeline-level state. Use for the FAILED + RUNNING-phase combo that startup reconciliation can produce. Cancel any live pods first (`cancel_task(cleanup=true)`) if record drift caused a false-positive FAILED — see #2420 for the tracking issue on adding a defensive route-level guard |
 | `advance_phase` | `POST /pipelines/{id}/phase` | Advance pipeline to a target phase. With `force=true`, stops running containers first to prevent SIGTERM cascading. When leaving the plan phase, automatically populates the contract from the plan draft |
 | `start_phase` | `POST /pipelines/{id}/phase/start` | Mark the current phase RUNNING. Does **not** spawn agents — agent spawning is driven by the `_run_pipeline` loop. Use for operator recovery when a phase needs to be re-marked RUNNING |
 | `complete_phase` | `POST /pipelines/{id}/phase/complete` | Mark a phase COMPLETE. Does **not** advance the pipeline — call `advance_phase` next. Response includes `current_phase` (unchanged) and `next_phase` (suggested transition). Returns 409 if unresolved HITL decisions exist; pass `force=true` to abandon them |
@@ -247,6 +248,7 @@ Four MCP tools expose phase management operations for pipeline recovery and manu
 
 All tools require `task_id` (the pipeline ID). Additional parameters:
 
+- **`start_pipeline`**: No additional parameters.
 - **`advance_phase`**: `target_phase` (string, required) — the phase to advance to (e.g., `"plan"`, `"implement"`, `"pr"`). `force` (boolean, optional, default `false`) — skip validation and stop running containers before advancing. **Important:** When `force=true`, containers from the current phase are stopped before the transition to prevent their SIGTERM signals from being misinterpreted as failures in the new phase. When the current phase is `plan`, `advance_phase` automatically runs the contract populate step (parsing the plan's `yaml-tasks` appendix into the contract (phases, tasks, and `contract.pr` metadata)), so a separate `populate_contract` call is not needed for plan→implement transitions.
 - **`start_phase`**: No additional parameters.
 - **`complete_phase`**: `artifacts` (object, optional) — phase completion artifacts to store (e.g., commit SHAs, PR URLs).
@@ -282,6 +284,14 @@ Note: reason codes are present in the raw HTTP response. The MCP handler layer d
 # 1. Check current state
 egg-orch phase get <pipeline-id>
 
+# 1a. If pipeline is FAILED with the current phase still RUNNING (a state startup
+#     reconciliation can produce on partial agent-state loss after an orch
+#     restart — see #2411), use start_pipeline. It resets the failed phase to
+#     PENDING and re-launches the runner.
+# Via MCP tool: start_pipeline(task_id="<id>")
+# Via REST:
+curl -X POST http://egg-orchestrator:9849/api/v1/pipelines/<id>/start
+
 # 2. Force-advance past a stuck phase (stops running containers first)
 # Via MCP tool: advance_phase(task_id="<id>", target_phase="implement", force=true)
 # Via REST:

orchestrator/mcp_tools.py

@@ -762,6 +762,55 @@ def _is_timeout_error(exc: BaseException) -> bool:
             "required": ["task_id", "target_phase"],
         },
     },
+    {
+        "name": "start_pipeline",
+        "description": (
+            "Recover a non-RUNNING pipeline by calling "
+            "``POST /api/v1/pipelines/{id}/start``.  Targets pipeline-level "
+            "state — distinct from ``start_phase``, which only flips the "
+            "current phase.  Intended for the FAILED + RUNNING-phase combo "
+            "that startup reconciliation can produce (#2411): the route "
+            "**unconditionally** resets the failed phase to PENDING "
+            "(clears ``containers``, ``agents``, ``artifacts`` regardless "
+            "of whether the records are verifiably stale), bumps "
+            "``run_epoch``, sets ``pipeline.status = RUNNING``, and "
+            "re-launches the ``_run_pipeline`` thread.  Also handles "
+            "AWAITING_HUMAN recovery when all decisions are resolved, and "
+            "starts PENDING pipelines (no early-return for PENDING in the "
+            "route).\n\n"
+            "Note: the reset is unconditional — there is no programmatic "
+            "check for live pods.  If pods labeled to the pipeline still "
+            "exist (e.g. orch restarted but pods are healthy), they will "
+            "be orphaned.  This footgun applies to **all** non-RUNNING "
+            "states the route accepts, not just the FAILED + RUNNING-phase "
+            "combo: AWAITING_HUMAN-with-resolved-decisions also flows "
+            "through a phase-reset branch (when the resolution is "
+            "request_changes / change_approach), and startup "
+            "reconciliation's AWAITING_HUMAN→FAILED transition runs "
+            "*before* the new live-pod safety net (so the live-pod-orphan "
+            "case can apply on the AWAITING_HUMAN recovery path too).  "
+            "Use ``cancel_task(cleanup=true)`` first or rely on the "
+            "running orchestrator's reconciliation if the pipeline is "
+            "genuinely alive.  See #2420 for the tracking issue on "
+            "adding a defensive route-level guard.\n\n"
+            "Error responses include a machine-readable ``reason`` code "
+            "(#1939). Note: reason codes are only visible to direct HTTP "
+            "callers; the MCP handler layer does not yet surface them.\n"
+            "- 409 — pipeline already RUNNING / COMPLETE / CANCELLED, or "
+            "AWAITING_HUMAN with pending decisions\n"
+            "- ``invalid_pipeline_id`` (400), ``pipeline_not_found`` (404)"
+        ),
+        "inputSchema": {
+            "type": "object",
+            "properties": {
+                "task_id": {
+                    "type": "string",
+                    "description": "Pipeline/task ID",
+                },
+            },
+            "required": ["task_id"],
+        },
+    },
     {
         "name": "start_phase",
         "description": (
@@ -1069,6 +1118,7 @@ def handle_tool_call(self, tool_name: str, arguments: dict[str, Any]) -> dict[st
             "restart_agent": self._handle_restart_agent,
             "restart_phase": self._handle_restart_phase,
             "advance_phase": self._handle_advance_phase,
+            "start_pipeline": self._handle_start_pipeline,
             "start_phase": self._handle_start_phase,
             "complete_phase": self._handle_complete_phase,
             "populate_contract": self._handle_populate_contract,
@@ -2588,6 +2638,21 @@ def _handle_advance_phase(self, args: dict[str, Any]) -> dict[str, Any]:
             result["failed_containers"] = failed_containers
         return result
 
+    def _handle_start_pipeline(self, args: dict[str, Any]) -> dict[str, Any]:
+        """Recover a non-RUNNING pipeline (#2411).
+
+        Targets the pipeline-level recovery route ``POST
+        /api/v1/pipelines/{id}/start``.  See the ``start_pipeline`` tool
+        definition in :data:`PIPELINE_TOOLS` for the full contract,
+        including the FAILED + RUNNING-phase combo from startup
+        reconciliation that this verb exists to recover from.
+        """
+        task_id = quote(args["task_id"], safe="")
+        return self._make_request(
+            f"/api/v1/pipelines/{task_id}/start",
+            method="POST",
+        )
+
     def _handle_start_phase(self, args: dict[str, Any]) -> dict[str, Any]:
         """Start execution of the current phase."""
         task_id = quote(args["task_id"], safe="")

orchestrator/startup_reconciliation.py

@@ -26,15 +26,26 @@ def get_logger(name: str, **kwargs) -> logging.Logger:  # type: ignore[misc]
 
 logger = get_logger("orchestrator.startup_reconciliation")
 
+# K8s label that scopes a pod to a specific pipeline.  Imported from
+# ``kubernetes_client`` so the literal lives in exactly one place.  Safe at
+# import time because ``kubernetes_client`` only imports the ``kubernetes``
+# pip package inside method bodies, so this module stays importable even
+# in test environments that don't install it.
+from kubernetes_client import LABEL_PIPELINE_ID as _LABEL_PIPELINE_ID
+
 
 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.
+    pipeline that shows status=RUNNING, the reconciler queries k8s for live
+    pods labeled ``egg.pipeline.id=<id>``.  If any pods are alive for the
+    pipeline, the pipeline is left RUNNING — record drift between the
+    persisted in-memory state and the new orch process's view of the pods is
+    expected after a restart and is reconciled by the running orchestrator,
+    not at startup (#2411).  Only when the pipeline has zero live pods do we
+    fall back to the older "any stale record fails the pipeline" behavior so
+    that genuinely orphaned pipelines still surface as FAILED.
 
     Args:
         store: StateStore instance (already bound to the correct repo path).
@@ -170,6 +181,47 @@ def reconcile_stale_containers(store: object, docker_client: object) -> int:
                 )
             continue
 
+        # Query k8s for pods labeled to this pipeline.  When any are alive
+        # the pipeline is not dead — even if individual ``container_id``s in
+        # the persisted state don't match the new orch process's view of
+        # the pods (e.g. a pod was recreated and its uid changed, or the
+        # record was written before the latest pod uid was observed),
+        # treat that drift as a problem the running orchestrator will
+        # reconcile naturally instead of terminating the pipeline at
+        # startup (#2411).
+        #
+        # On failure we fail-safe (leave the pipeline RUNNING and skip).
+        # In practice both queries route through the same
+        # ``KubernetesClient.list_containers`` → ``list_namespaced_pod`` —
+        # if the global query at line 63 already succeeded, a per-pipeline
+        # failure is rare enough that the safe choice is to defer to the
+        # running orchestrator's reconciliation rather than risk repeating
+        # the #2411 false-positive on misbehaving clusters.  The genuinely
+        # orphaned case (zero live pods) is rare and surfaceable elsewhere
+        # — drift cases are the active concern here.
+        try:
+            pipeline_live_containers = docker_client.list_containers(  # type: ignore[attr-defined]
+                labels={_LABEL_PIPELINE_ID: pipeline_id},
+            )
+            pipeline_live_ids: set[str] = {ci.container_id for ci in pipeline_live_containers}
+        except Exception as e:
+            logger.warning(
+                "Startup reconciliation: pipeline-scoped container query failed, "
+                "leaving pipeline RUNNING (deferring to running orchestrator's "
+                "reconciliation rather than risk a #2411-style false positive)",
+                pipeline_id=pipeline_id,
+                error=str(e),
+            )
+            continue
+
+        if pipeline_live_ids:
+            logger.info(
+                "Startup reconciliation: pipeline has live pods, leaving RUNNING",
+                pipeline_id=pipeline_id,
+                live_pod_count=len(pipeline_live_ids),
+            )
+            continue
+
         for container_info in phase_execution.containers:
             if container_info.status == ContainerStatus.RUNNING:
                 if container_info.container_id not in live_ids:

orchestrator/tests/test_mcp_tools.py

@@ -840,6 +840,7 @@ def test_all_tools_registered(self, handler):
             "restart_agent",
             "restart_phase",
             "advance_phase",
+            "start_pipeline",
             "start_phase",
             "complete_phase",
             "populate_contract",
@@ -874,6 +875,59 @@ def test_checkpoint_tools_have_repo_property(self):
             assert "owner/repo" in props["repo"]["description"]
 
 
+class TestStartPipeline:
+    """Tests for the start_pipeline MCP tool (#2411)."""
+
+    def test_tool_definition_exists(self):
+        from mcp_tools import PIPELINE_TOOLS
+
+        tool_names = [t["name"] for t in PIPELINE_TOOLS]
+        assert "start_pipeline" in tool_names
+
+    def test_tool_definition_requires_task_id(self):
+        from mcp_tools import PIPELINE_TOOLS
+
+        tool = next(t for t in PIPELINE_TOOLS if t["name"] == "start_pipeline")
+        schema = tool["inputSchema"]
+        assert "task_id" in schema.get("required", [])
+
+    def test_calls_pipeline_start_endpoint(self, handler):
+        """start_pipeline should POST to /pipelines/{id}/start, not /phase/start."""
+        with patch.object(handler, "_make_request") as mock_req:
+            mock_req.return_value = {"success": True, "data": {"status": "running"}}
+            handler.handle_tool_call("start_pipeline", {"task_id": "issue-2411"})
+
+        mock_req.assert_called_once()
+        endpoint = mock_req.call_args[0][0]
+        assert "/api/v1/pipelines/issue-2411/start" in endpoint
+        assert "/phase/" not in endpoint
+        assert mock_req.call_args.kwargs.get("method") == "POST"
+
+    def test_url_encodes_task_id(self, handler):
+        """task_id with reserved characters should be URL-encoded."""
+        with patch.object(handler, "_make_request") as mock_req:
+            mock_req.return_value = {"success": True}
+            handler.handle_tool_call("start_pipeline", {"task_id": "issue-1/odd"})
+
+        endpoint = mock_req.call_args[0][0]
+        assert "issue-1%2Fodd" in endpoint
+
+    def test_distinct_from_start_phase(self, handler):
+        """start_pipeline and start_phase must hit different endpoints."""
+        with patch.object(handler, "_make_request") as mock_req:
+            mock_req.return_value = {"success": True}
+
+            handler.handle_tool_call("start_pipeline", {"task_id": "issue-99"})
+            pipeline_endpoint = mock_req.call_args[0][0]
+
+            handler.handle_tool_call("start_phase", {"task_id": "issue-99"})
+            phase_endpoint = mock_req.call_args[0][0]
+
+        assert pipeline_endpoint != phase_endpoint
+        assert pipeline_endpoint.endswith("/start")
+        assert phase_endpoint.endswith("/phase/start")
+
+
 class TestValidateConfig:
     def test_valid_config(self, handler):
         result = handler.handle_tool_call(