diff --git a/k8s/base/orchestrator-deployment.yaml b/k8s/base/orchestrator-deployment.yaml index e554bb1d3f..33f5748ded 100644 --- a/k8s/base/orchestrator-deployment.yaml +++ b/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 diff --git a/k8s/overlays/local/patches/orchestrator-volumes.yaml b/k8s/overlays/local/patches/orchestrator-volumes.yaml index a77ab358dc..05e32ac81d 100644 --- a/k8s/overlays/local/patches/orchestrator-volumes.yaml +++ b/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 diff --git a/orchestrator/routes/deployment.py b/orchestrator/routes/deployment.py index 94f5a701cc..055457dee8 100644 --- a/orchestrator/routes/deployment.py +++ b/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", "") + 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 diff --git a/orchestrator/routes/pipelines.py b/orchestrator/routes/pipelines.py index 31614b0794..bef3a2e70a 100644 --- a/orchestrator/routes/pipelines.py +++ b/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 diff --git a/orchestrator/state_store.py b/orchestrator/state_store.py index 2629414192..6624c1c818 100644 --- a/orchestrator/state_store.py +++ b/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) diff --git a/orchestrator/tests/test_deployment_routes.py b/orchestrator/tests/test_deployment_routes.py index f7d0e652e3..424c91c911 100644 --- a/orchestrator/tests/test_deployment_routes.py +++ b/orchestrator/tests/test_deployment_routes.py @@ -848,6 +848,193 @@ def test_session_store_rule_fires_on_gateway_dash_variant(self): assert len(fired) == 1 assert fired[0]["resource"] == "Deployment/gateway-canary" + def _orchestrator_doc(self, *, name="orchestrator", repos_vol, egg_state_vol): + """Orchestrator Deployment fixture with optional repos/egg-state volumes. + + Shapes the #3070 trap: the pipeline-state store (egg-state) must share + a persistence class with the repos whose .git holds its committed + history. Pass ``None`` for either volume to omit it. + """ + volumes = [v for v in (repos_vol, egg_state_vol) if v is not None] + return { + "kind": "Deployment", + "metadata": {"name": name}, + "spec": { + "template": { + "metadata": {"labels": {"app": "orchestrator"}}, + "spec": { + "containers": [{"name": "orch", "image": "egg-orchestrator:dev"}], + "volumes": volumes, + }, + } + }, + } + + def test_pipeline_state_store_not_persistent_triggers_error(self): + """hostPath repos + emptyDir egg-state is the #3070 trap.""" + from routes.deployment import _validate_deployment_docs + + docs = [ + self._orchestrator_doc( + repos_vol={"name": "repos", "hostPath": {"path": "/host/repos"}}, + egg_state_vol={"name": "egg-state", "emptyDir": {}}, + ) + ] + warnings = _validate_deployment_docs(docs, is_k3s=True) + fired = [w for w in warnings if w.get("rule") == "pipeline-state-store-not-persistent"] + assert len(fired) == 1 + assert fired[0]["severity"] == "error" + assert fired[0]["resource"] == "Deployment/orchestrator" + assert "#3070" in fired[0]["message"] + assert "emptyDir" in fired[0]["message"] + + def test_pipeline_state_store_rule_fires_when_egg_state_absent(self): + """hostPath repos + NO egg-state volume is the realistic overlay slip. + + Pre-#3070 the local overlay added the hostPath ``repos`` volume but + left the base's ``egg-state`` emptyDir alone; an overlay author who + omits the volume entirely lands /home/egg/.egg-state inside the + ``home`` emptyDir — same loss mode, so the rule must fire on absence + the same way it fires on emptyDir. + """ + from routes.deployment import _validate_deployment_docs + + docs = [ + self._orchestrator_doc( + repos_vol={"name": "repos", "hostPath": {"path": "/host/repos"}}, + egg_state_vol=None, + ) + ] + warnings = _validate_deployment_docs(docs, is_k3s=True) + fired = [w for w in warnings if w.get("rule") == "pipeline-state-store-not-persistent"] + assert len(fired) == 1 + assert fired[0]["severity"] == "error" + assert "not declared" in fired[0]["message"] or "no ``egg-state``" in fired[0]["message"] + + def test_pipeline_state_store_persistent_is_clean(self): + """hostPath repos + hostPath egg-state — the fixed shape — is clean.""" + from routes.deployment import _validate_deployment_docs + + docs = [ + self._orchestrator_doc( + repos_vol={"name": "repos", "hostPath": {"path": "/host/repos"}}, + egg_state_vol={"name": "egg-state", "hostPath": {"path": "/host/state"}}, + ) + ] + warnings = _validate_deployment_docs(docs, is_k3s=True) + fired = [w for w in warnings if w.get("rule") == "pipeline-state-store-not-persistent"] + assert fired == [] + + def test_pipeline_state_store_rule_silent_when_repos_not_persistent(self): + """All-emptyDir base/cloud shape has no asymmetry, so silent. + + With ephemeral repos nothing survives a pod recreation anyway — there + is no stale committed tip to silently rewind to — so the rule must not + fire (avoids false positives on stateless base deploys). + """ + from routes.deployment import _validate_deployment_docs + + docs = [ + self._orchestrator_doc( + repos_vol={"name": "repos", "emptyDir": {}}, + egg_state_vol={"name": "egg-state", "emptyDir": {}}, + ) + ] + warnings = _validate_deployment_docs(docs, is_k3s=True) + fired = [w for w in warnings if w.get("rule") == "pipeline-state-store-not-persistent"] + assert fired == [] + + def test_pipeline_state_store_rule_scoped_to_orchestrator(self): + """The rule must not fire on non-orchestrator deployments.""" + from routes.deployment import _validate_deployment_docs + + docs = [ + self._orchestrator_doc( + name="litellm-orchestrator", + repos_vol={"name": "repos", "hostPath": {"path": "/host/repos"}}, + egg_state_vol={"name": "egg-state", "emptyDir": {}}, + ) + ] + warnings = _validate_deployment_docs(docs, is_k3s=True) + fired = [w for w in warnings if w.get("rule") == "pipeline-state-store-not-persistent"] + assert fired == [] + + def test_pipeline_state_store_rule_clean_when_both_pvc_backed(self): + """PVC repos + PVC egg-state is clean (rule generalizes beyond hostPath). + + Counterpart to ``test_session_store_rule_clean_when_both_pvc_backed``: + locks in that the rule expresses "pipeline-state store has at least the + same persistence class as repos" via emptyDir-checking, not via + hostPath-checking — so PVC / NFS / CSI futures work without a code + change. Both volumes survive pod recreation; no asymmetry; silent. + """ + from routes.deployment import _validate_deployment_docs + + docs = [ + self._orchestrator_doc( + repos_vol={ + "name": "repos", + "persistentVolumeClaim": {"claimName": "repos-pvc"}, + }, + egg_state_vol={ + "name": "egg-state", + "persistentVolumeClaim": {"claimName": "state-pvc"}, + }, + ) + ] + warnings = _validate_deployment_docs(docs, is_k3s=True) + fired = [w for w in warnings if w.get("rule") == "pipeline-state-store-not-persistent"] + assert fired == [] + + def test_pipeline_state_store_rule_fires_when_pvc_repos_emptydir_state(self): + """PVC repos + emptyDir egg-state is the hypothetical cloud-overlay trap. + + Counterpart to ``test_session_store_rule_fires_when_pvc_worktrees_emptydir_state``: + same #3070 failure mode as hostPath/emptyDir — repos survive a pod + recreation (PVC) but the pipeline-state worktree doesn't (emptyDir). + The rule must fire here too — that's the whole point of expressing the + check against emptyDir rather than against the specific hostPath + backing the local overlay uses today. + """ + from routes.deployment import _validate_deployment_docs + + docs = [ + self._orchestrator_doc( + repos_vol={ + "name": "repos", + "persistentVolumeClaim": {"claimName": "repos-pvc"}, + }, + egg_state_vol={"name": "egg-state", "emptyDir": {}}, + ) + ] + warnings = _validate_deployment_docs(docs, is_k3s=True) + fired = [w for w in warnings if w.get("rule") == "pipeline-state-store-not-persistent"] + assert len(fired) == 1 + assert fired[0]["severity"] == "error" + + def test_pipeline_state_store_rule_fires_on_orchestrator_dash_variant(self): + """An ``orchestrator-canary`` deployment IS an orchestrator variant and must match. + + Counterpart to ``test_session_store_rule_fires_on_gateway_dash_variant``: + the exact-or-prefix scope is ``orchestrator`` exactly or + ``orchestrator-*``, so a canary / rollout variant still inherits the + invariant. Pairs with ``test_pipeline_state_store_rule_scoped_to_orchestrator``, + which pins that an unrelated ``litellm-orchestrator`` does NOT match. + """ + from routes.deployment import _validate_deployment_docs + + docs = [ + self._orchestrator_doc( + name="orchestrator-canary", + repos_vol={"name": "repos", "hostPath": {"path": "/host/repos"}}, + egg_state_vol={"name": "egg-state", "emptyDir": {}}, + ) + ] + warnings = _validate_deployment_docs(docs, is_k3s=True) + fired = [w for w in warnings if w.get("rule") == "pipeline-state-store-not-persistent"] + assert len(fired) == 1 + assert fired[0]["resource"] == "Deployment/orchestrator-canary" + # --------------------------------------------------------------------------- # prune_stale_worktrees diff --git a/orchestrator/tests/test_state_store.py b/orchestrator/tests/test_state_store.py index 175b601eda..f7886fcd12 100644 --- a/orchestrator/tests/test_state_store.py +++ b/orchestrator/tests/test_state_store.py @@ -297,6 +297,51 @@ def test_save_without_commit(self, state_store, mock_git): commit_calls = [c for c in mock_git.call_args_list if "commit" in c[0]] assert len(commit_calls) == 0 + def test_save_prompt_driven_pipeline_commits_by_default(self, state_store, mock_git): + """Prompt-driven pipelines (no issue_number) commit on every save. + + Regression test for #3070: the old gate committed prompt-driven + pipelines only at phase advance/completion, so one parked at a HITL + gate had no commit at all and vanished on pod recreation. + """ + pipeline = Pipeline( + id="pipeline-deadbeef", + issue_number=None, + repo="owner/repo", + branch="egg/pipeline-deadbeef", + prompt="free-text pipeline", + ) + pipeline.status = PipelineStatus.AWAITING_HUMAN + # _commit_state only commits when the staged diff is non-empty. + mock_git.side_effect = lambda *args, **kwargs: MagicMock( + stdout="abc1234\n", returncode=1 if args[0] == "diff" else 0 + ) + state_store.save_pipeline(pipeline) + + commit_calls = [c for c in mock_git.call_args_list if "commit" in c[0]] + assert len(commit_calls) == 1 + + def test_delete_prompt_driven_pipeline_commits_by_default(self, state_store, mock_git): + """Deletions commit regardless of pipeline origin (#3070 invariant).""" + pipeline = Pipeline( + id="pipeline-cafef00d", + issue_number=None, + repo="owner/repo", + branch="egg/pipeline-cafef00d", + prompt="free-text pipeline", + ) + state_store.save_pipeline(pipeline, commit=False) + mock_git.reset_mock() + # Deletion stages then commits when the diff reports staged changes. + mock_git.side_effect = lambda *args, **kwargs: MagicMock( + stdout="abc1234\n", returncode=1 if args[0] == "diff" else 0 + ) + + state_store.delete_pipeline("pipeline-cafef00d") + + commit_calls = [c for c in mock_git.call_args_list if "commit" in c[0]] + assert len(commit_calls) == 1 + class TestPipelineUpdate: """Tests for updating pipelines.""" @@ -1579,7 +1624,7 @@ def failing_git(*args, **kwargs): mock_git.side_effect = failing_git pipeline.status = PipelineStatus.RUNNING - path = state_store.save_pipeline(pipeline, force_commit=True) + path = state_store.save_pipeline(pipeline) # File must be valid JSON and loadable loaded = state_store.load_pipeline("issue-702") @@ -1597,7 +1642,7 @@ def test_save_pipeline_does_not_raise_on_commit_failure(self, state_store, mock_ ) with patch.object(state_store, "_commit_state", side_effect=GitOperationError("boom")): - path = state_store.save_pipeline(pipeline, force_commit=True) + path = state_store.save_pipeline(pipeline) assert path.exists() loaded = state_store.load_pipeline("issue-703")