fix(orchestrator): commit pipeline state on every save and host-persist the state store (#3070)

k8s/base/orchestrator-deployment.yaml

@@ -163,10 +163,17 @@ spec:
             # emptyDir at /home/egg makes $HOME writable (needed for
             # .gitconfig, .egg-worktrees, sharing/notifications, etc.)
             # while keeping the rest of the rootfs read-only.
+            # No dedicated ``egg-state`` volume here: /home/egg/.egg-state
+            # falls inside the ``home`` emptyDir (same lifecycle), matching
+            # the gateway base. Overlays that host-persist ``repos`` must add
+            # a persistent ``egg-state`` volume or pipeline records saved
+            # since the last state-branch commit are lost on pod recreation —
+            # enforced by the ``pipeline-state-store-not-persistent``
+            # validation rule (#3070). Declaring it emptyDir here instead
+            # would break the overlay's strategic-merge add (emptyDir and
+            # hostPath would merge into one invalid two-type volume).
             - name: home
               mountPath: /home/egg
-            - name: egg-state
-              mountPath: /home/egg/.egg-state
             - name: tmp
               mountPath: /tmp
             # #2528: read-only mount of repositories.yaml so the
@@ -187,8 +194,6 @@ spec:
       volumes:
         - name: home
           emptyDir: {}
-        - name: egg-state
-          emptyDir: {}
         - name: tmp
           emptyDir: {}
         # #2528: project the repositories.yaml key out of the shared

k8s/overlays/local/patches/orchestrator-volumes.yaml

@@ -56,6 +56,28 @@ spec:
               mountPath: /home/egg/repos
             - name: worktrees
               mountPath: /home/egg/.egg-worktrees
+            # Pipeline-state store (orchestrator/state_store.py keeps the
+            # ``egg/pipeline-state`` worktrees under
+            # /home/egg/.egg-state/pipeline-worktree*). MUST be host-backed
+            # because repos above are: the state branch's *commits* live in
+            # each repo's .git (host-persistent), but the worktree holds any
+            # state saved between commits. With an emptyDir here, a pod
+            # recreation rebuilds the worktree from the last committed branch
+            # tip and silently drops everything newer — in #3070 that erased
+            # every in-flight prompt-driven pipeline (get_status 404) because
+            # their records had never been committed. Same host dir the
+            # gateway mounts for its session store; the two use disjoint
+            # subpaths (sessions/ vs pipeline-worktree*/), so they don't
+            # collide. Cross-pod ownership note: the gateway's entrypoint
+            # chowns ``/home/egg/.egg-state`` to ``HOST_UID:HOST_GID`` on
+            # startup, which is what lets the orchestrator (``runAsUser:
+            # 1000``) write to the shared dir — local-dev convention is
+            # ``HOST_UID=1000``, so this aligns; a host deploying with
+            # ``HOST_UID != 1000`` would hit EACCES here. Cloud overlays
+            # moving to PVCs (the future intent rule 7 / rule 6 already
+            # accommodate) sidestep the chown coupling entirely.
+            - name: egg-state
+              mountPath: /home/egg/.egg-state
             - name: secrets
               mountPath: /secrets
               readOnly: true
@@ -68,6 +90,10 @@ spec:
           hostPath:
             path: ${EGG_HOST_HOME}/.egg-worktrees
             type: DirectoryOrCreate
+        - name: egg-state
+          hostPath:
+            path: ${EGG_HOST_HOME}/.egg-state
+            type: DirectoryOrCreate
         - name: secrets
           secret:
             secretName: gateway-secrets

orchestrator/routes/deployment.py

@@ -926,6 +926,59 @@ def _validate_deployment_docs(docs: list[dict[str, Any]], *, is_k3s: bool) -> li
             ),
         )
 
+    # Rule 7: orchestrator pipeline-state store must share the repos'
+    # persistence lifetime (#3070). The orchestrator's StateStore keeps the
+    # ``egg/pipeline-state`` worktrees under
+    # /home/egg/.egg-state/pipeline-worktree* (the ``egg-state`` volume).
+    # The state branch's *commits* live in each repo's .git on the ``repos``
+    # volume, but the worktree's working files hold anything saved since the
+    # last commit. When an overlay gives ``repos`` a volume that survives pod
+    # recreation while ``egg-state`` is ephemeral (``emptyDir`` or absent —
+    # in which case /home/egg/.egg-state falls inside the ``home`` emptyDir),
+    # a pod recreation rebuilds each state worktree from the last committed
+    # branch tip and silently drops everything newer: in #3070 every
+    # in-flight pipeline whose record had no commit yet simply vanished
+    # (get_status 404, absent from list_tasks). Like rule 6, the check is
+    # expressed against ``emptyDir`` rather than ``hostPath`` so PVC/NFS/CSI
+    # backings satisfy it, and it self-gates on ``repos`` being persistent so
+    # it stays silent on all-emptyDir base/cloud deploys.
+    for dep in deployments:
+        name = dep.get("metadata", {}).get("name", "<unknown>")
+        if name != "orchestrator" and not name.startswith("orchestrator-"):
+            continue
+        vols = _deployment_volumes(dep)
+        repos_vol = next((v for v in vols if (v or {}).get("name") == "repos"), None)
+        if repos_vol is None or "emptyDir" in repos_vol:
+            continue
+        egg_state_vol = next((v for v in vols if (v or {}).get("name") == "egg-state"), None)
+        egg_state_persistent = egg_state_vol is not None and "emptyDir" not in egg_state_vol
+        if egg_state_persistent:
+            continue
+        if egg_state_vol is None:
+            detail = (
+                "orchestrator repos survive pod recreation but no "
+                "``egg-state`` volume is declared, so /home/egg/.egg-state "
+                "falls inside the ``home`` emptyDir and the pipeline-state "
+                "worktree is ephemeral"
+            )
+        else:
+            detail = (
+                "orchestrator repos survive pod recreation but its "
+                "pipeline-state store (egg-state volume, "
+                "/home/egg/.egg-state) is an emptyDir and does not"
+            )
+        _warn(
+            warnings,
+            rule="pipeline-state-store-not-persistent",
+            severity="error",
+            resource=f"Deployment/{name}",
+            message=(
+                f"{detail}; an orchestrator pod recreation will rebuild the "
+                "state worktree from the last committed branch tip and "
+                "silently lose any pipeline state saved since (#3070)"
+            ),
+        )
+
     return warnings
 
 

orchestrator/routes/pipelines.py

@@ -23133,7 +23133,7 @@ def _health_monitor_poll(monitor, stop_event: threading.Event, interval: float =
                 with get_pipeline_state_lock(pipeline_id):
                     pipeline = store.load_pipeline(pipeline_id)
                     pipeline.status = PipelineStatus.COMPLETE
-                    store.save_pipeline(pipeline, force_commit=(pipeline.issue_number is None))
+                    store.save_pipeline(pipeline)
 
                 # Report pipeline completion to collaborator
                 report_pipeline_status(
@@ -23189,7 +23189,7 @@ def _health_monitor_poll(monitor, stop_event: threading.Event, interval: float =
                 pipeline.current_phase = next_phase
                 pipeline.run_epoch = datetime.now(UTC)
                 # ``updated_at`` is unconditionally set by ``StateStore.save_pipeline``.
-                store.save_pipeline(pipeline, force_commit=(pipeline.issue_number is None))
+                store.save_pipeline(pipeline)
 
             # Drop the previous phase's in-memory consensus tracker and
             # message-store entries (#2502).  The other phase-transition

orchestrator/state_store.py

@@ -675,18 +675,15 @@ def save_pipeline(
         commit: bool = True,
         message: str | None = None,
         expected_version: int | None = None,
-        force_commit: bool = False,
     ) -> Path:
         """Save pipeline state to disk with optimistic locking.
 
         Args:
             pipeline: Pipeline state to save
-            commit: Whether to commit the change (ignored for local pipelines unless force_commit=True)
+            commit: Whether to commit the change
             message: Commit message (auto-generated if not provided)
             expected_version: If provided, checks that current version matches
                               before saving (optimistic locking)
-            force_commit: If True, commit even for local pipelines. Use at phase
-                          boundaries to ensure state is persisted to git.
 
         Returns:
             Path to saved file
@@ -731,11 +728,12 @@ def save_pipeline(
                 pass
             raise
 
-        # For prompt-driven pipelines (no issue_number), only commit if force_commit
-        # is True (phase boundaries). For issue pipelines, always commit when commit=True.
-        is_prompt_driven = pipeline.issue_number is None
-        should_commit = commit and (not is_prompt_driven or force_commit)
-        if should_commit:
+        # Commit unconditionally: the on-disk file alone is not durable. The
+        # state worktree may sit on a pod-lifetime volume, so any save that
+        # skips the commit exists only until the next pod recreation — that is
+        # how prompt-driven pipelines parked at a HITL gate vanished in #3070
+        # (the old gate committed them only on phase advance/completion).
+        if commit:
             try:
                 self._commit_state(pipeline, message)
             except GitOperationError:
@@ -1093,15 +1091,13 @@ def delete_pipeline(
         self,
         pipeline_id: str,
         commit: bool = True,
-        force_commit: bool = False,
         cleanup_lock: bool = True,
     ) -> None:
         """Delete a pipeline.
 
         Args:
             pipeline_id: Pipeline ID to delete
-            commit: Whether to commit the deletion (ignored for local unless force_commit)
-            force_commit: If True, commit deletion even for local pipelines
+            commit: Whether to commit the deletion
             cleanup_lock: Whether to release the per-pipeline state lock.
                 Set to False when called from within a lock (e.g. create_pipeline
                 replacing a terminal pipeline) to avoid removing the lock while
@@ -1121,11 +1117,10 @@ def delete_pipeline(
         if cleanup_lock:
             release_pipeline_state_lock(pipeline_id)
 
-        # For local pipelines, only commit if force_commit is True
-        # For issue pipelines, always commit when commit=True
-        is_local = pipeline_id.startswith("local-")
-        should_commit = commit and (not is_local or force_commit)
-        if should_commit:
+        # Commit unconditionally (same durability invariant as save_pipeline,
+        # #3070): a deletion that only happens on disk resurrects the pipeline
+        # on the next pod recreation.
+        if commit:
             wt = self.worktree
             with self._git_op():
                 self._run_git("add", rel_path, cwd=wt)