fix(gateway): gate worktree cleanup on pipeline liveness, never delete branches in orphan sweeps (#3070)

gateway/README.md

@@ -536,6 +536,10 @@ Both methods clear all in-memory config caches so the next access re-reads from
 
 **Note:** `GATEWAY_TRUSTED_USERS` is read from the process environment, which is fixed at container start time. Changing trusted users requires a container restart — SIGHUP will re-read the same environment value.
 
+## Local Development
+
+On gateway startup the background worktree-cleanup thread polls the orchestrator at `EGG_ORCHESTRATOR_URL` (default `http://egg-orchestrator:9849`) for the active-pipeline set before sweeping; if the orchestrator never answers it waits up to `EGG_CLEANUP_ORCHESTRATOR_WAIT_SECONDS` (default `600`s — sized for a redeploy where both pods boot together) before logging at ERROR and skipping the sweep (#3070). Running the gateway against a non-existent orchestrator (developer laptop, isolated test container) therefore spins for the full 10 minutes by default — set `EGG_CLEANUP_ORCHESTRATOR_WAIT_SECONDS=0` to disable the wait (single attempt; sweep is then skipped immediately on first failure, which is the right behavior for local dev where there's nothing to preserve).
+
 ## Design Decisions
 
 1. **No merge capability**: Gateway does not expose a merge endpoint. Human must merge via GitHub UI. This maintains the existing safety model.

gateway/gateway.py

@@ -170,6 +170,10 @@
     )
     from .jira_search import extract_search_projects
     from .mode_gate import require_private_mode
+    from .orchestrator_pipelines import (
+        fetch_active_pipeline_ids,
+        wait_for_active_pipeline_ids,
+    )
     from .phase_filter import (
         OperationType,
         PipelinePhase,
@@ -324,6 +328,10 @@
         extract_search_projects,
     )
     from mode_gate import require_private_mode  # type: ignore[no-redef, import-untyped]
+    from orchestrator_pipelines import (  # type: ignore[no-redef, import-untyped]
+        fetch_active_pipeline_ids,
+        wait_for_active_pipeline_ids,
+    )
     from phase_filter import (  # type: ignore[no-redef, import-untyped]
         OperationType,
         PipelinePhase,
@@ -8066,8 +8074,11 @@ def worktrees_prune() -> tuple[Response, int] | Response:
     that would be removed but does not mutate the filesystem. When
     false, removes them using the existing ``cleanup_orphaned_worktrees``
     helper.  The active-container set is derived from the session
-    manager (plus an opportunistic ``docker ps`` fallback) so a live
-    pipeline's worktree is never mistaken for an orphan.
+    manager (plus an opportunistic ``docker ps`` fallback), and worktrees
+    anchored to a non-terminal pipeline (per the orchestrator) are
+    preserved even with no live container — a pipeline parked at a HITL
+    gate has neither (#3070). Returns 503 when pipeline liveness cannot
+    be verified.
 
     Proxied from the orchestrator's
     ``/api/v1/deployment/prune-worktrees`` endpoint (#1759).
@@ -8083,15 +8094,32 @@ def worktrees_prune() -> tuple[Response, int] | Response:
     if not _worktree_prune_lock.acquire(timeout=60):
         return make_error("Another worktree prune is in progress", status_code=409)
     try:
+        # Pipeline liveness is required, not best-effort: a parked pipeline
+        # has no containers or sessions, so the container-derived set alone
+        # would mark its worktree an orphan (#3070). This endpoint is proxied
+        # from the orchestrator, so it is normally up; refuse rather than
+        # sweep blind if it cannot answer.
+        active_pipeline_ids = fetch_active_pipeline_ids()
+        if active_pipeline_ids is None:
+            return make_error(
+                "Cannot verify pipeline liveness (orchestrator unreachable); "
+                "refusing to prune worktrees",
+                status_code=503,
+            )
+
         active_container_ids = _collect_active_container_ids()
         git_prune_report = manager.git_worktree_prune_all()
-        orphan_dirs = manager.list_orphan_worktree_dirs(active_containers=active_container_ids)
+        orphan_dirs = manager.list_orphan_worktree_dirs(
+            active_containers=active_container_ids,
+            active_pipeline_ids=active_pipeline_ids,
+        )
 
         removed_count = 0
         removed_paths: list[str] = []
         if not dry_run and orphan_dirs:
             removed_count = manager.cleanup_orphaned_worktrees(
                 active_containers=active_container_ids,
+                active_pipeline_ids=active_pipeline_ids,
             )
             # Any orphan we enumerated that no longer exists on disk
             # was removed by the helper.
@@ -8111,6 +8139,7 @@ def worktrees_prune() -> tuple[Response, int] | Response:
                 "git_worktree_prune": git_prune_report,
                 "orphan_dirs_count": len(orphan_dirs),
                 "active_containers_count": len(active_container_ids),
+                "active_pipelines_count": len(active_pipeline_ids),
                 "removed_count": removed_count,
             },
         )
@@ -8122,6 +8151,7 @@ def worktrees_prune() -> tuple[Response, int] | Response:
                 "git_worktree_prune": git_prune_report,
                 "orphan_dirs": orphan_dirs,
                 "active_containers_count": len(active_container_ids),
+                "active_pipelines_count": len(active_pipeline_ids),
                 "removed_count": removed_count,
                 "removed_paths": removed_paths,
             },
@@ -10167,9 +10197,19 @@ def main() -> None:
     # could serve any requests. See: https://github.com/jwbron/egg/issues/1400
     def _background_worktree_cleanup() -> None:
         try:
+            # Container liveness alone cannot distinguish a crashed leftover
+            # from a pipeline parked at a HITL gate (no containers, no
+            # sessions — that's its normal state). Ask the orchestrator which
+            # pipelines are live before sweeping; on a redeploy it may still
+            # be booting, so poll up to the configured deadline. If it never
+            # answers, startup_cleanup skips the sweep (fail-safe) — see
+            # #3070, where a blind sweep with active_containers=0 deleted a
+            # parked pipeline's worktree, contract, and branches.
+            active_pipeline_ids = wait_for_active_pipeline_ids()
             orphans_removed = startup_cleanup(
                 active_containers=active_container_ids,
                 session_manager=get_session_manager(),
+                active_pipeline_ids=active_pipeline_ids,
             )
             if orphans_removed > 0:
                 logger.info(f"Startup cleanup removed {orphans_removed} orphaned worktree(s)")

gateway/orchestrator_pipelines.py

@@ -0,0 +1,128 @@
+"""HTTP client for the orchestrator's pipeline listing.
+
+Worktree cleanup must not equate "no live container" with "orphaned
+worktree": a pipeline parked at a HITL gate (or between phases) has no
+running containers and no sessions, yet its worktree holds the contract
+and any un-pushed work. In #3070 a redeploy ran startup cleanup with
+``active_containers=0`` and force-removed every worktree — including a
+pipeline whose refine analysis had just been operator-approved.
+
+``fetch_active_pipeline_ids`` asks the orchestrator which pipelines are
+non-terminal so cleanup can preserve their worktrees regardless of
+container liveness. Failure returns ``None`` (never an empty set) so
+callers can distinguish "verified nothing active" from "could not
+verify" and fail safe by skipping deletion entirely.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import time
+from urllib.error import HTTPError, URLError
+from urllib.request import Request, urlopen
+
+from egg_logging import get_logger
+
+logger = get_logger("gateway.orchestrator_pipelines")
+
+_DEFAULT_ORCHESTRATOR_URL = "http://egg-orchestrator:9849"
+_DEFAULT_FETCH_TIMEOUT_SECONDS = 15
+# Startup cleanup runs in a background thread, so a long wait is cheap;
+# a redeploy restarts both pods and the orchestrator's cold boot
+# (image pull + startup reconciliation) can take minutes.
+_DEFAULT_MAX_WAIT_SECONDS = 600
+_DEFAULT_POLL_INTERVAL_SECONDS = 5.0
+
+
+def _orchestrator_url() -> str:
+    return os.environ.get("EGG_ORCHESTRATOR_URL", _DEFAULT_ORCHESTRATOR_URL).rstrip("/")
+
+
+def fetch_active_pipeline_ids(
+    *,
+    timeout: float = _DEFAULT_FETCH_TIMEOUT_SECONDS,
+) -> set[str] | None:
+    """Return the IDs of all non-terminal pipelines, or ``None`` on failure.
+
+    Queries ``GET /api/v1/pipelines?active_only=true``. ``None`` (as
+    opposed to an empty set) means the answer is unknown — the caller
+    must not treat it as "no active pipelines".
+    """
+    url = f"{_orchestrator_url()}/api/v1/pipelines?active_only=true"
+    req = Request(url, headers={"Accept": "application/json"}, method="GET")
+    try:
+        with urlopen(req, timeout=timeout) as resp:
+            body = resp.read().decode("utf-8")
+    except (HTTPError, URLError, TimeoutError) as exc:
+        logger.warning(
+            "Could not fetch active pipelines from orchestrator",
+            url=url,
+            error=str(exc),
+        )
+        return None
+    except Exception as exc:  # pragma: no cover - defensive
+        logger.warning(
+            "Unexpected error fetching active pipelines",
+            url=url,
+            error=str(exc),
+        )
+        return None
+
+    try:
+        parsed = json.loads(body)
+        pipelines = parsed["data"]["pipelines"]
+        ids = {p["id"] for p in pipelines if p.get("id")}
+    except (json.JSONDecodeError, KeyError, TypeError) as exc:
+        logger.warning(
+            "Malformed active-pipelines response from orchestrator",
+            url=url,
+            error=str(exc),
+        )
+        return None
+
+    logger.info("Fetched active pipelines from orchestrator", count=len(ids))
+    return ids
+
+
+def wait_for_active_pipeline_ids(
+    *,
+    max_wait_seconds: float | None = None,
+    poll_interval_seconds: float = _DEFAULT_POLL_INTERVAL_SECONDS,
+) -> set[str] | None:
+    """Poll the orchestrator until it answers or the deadline passes.
+
+    Intended for gateway startup, where the orchestrator pod may still be
+    booting (on a redeploy both restart together). Returns the active
+    pipeline-ID set on success, ``None`` once ``max_wait_seconds`` is
+    exhausted. ``EGG_CLEANUP_ORCHESTRATOR_WAIT_SECONDS`` overrides the
+    default deadline; ``0`` disables waiting (single attempt).
+    """
+    if max_wait_seconds is None:
+        try:
+            max_wait_seconds = float(
+                os.environ.get(
+                    "EGG_CLEANUP_ORCHESTRATOR_WAIT_SECONDS",
+                    str(_DEFAULT_MAX_WAIT_SECONDS),
+                )
+            )
+        except ValueError:
+            max_wait_seconds = _DEFAULT_MAX_WAIT_SECONDS
+
+    deadline = time.monotonic() + max_wait_seconds
+    attempt = 0
+    while True:
+        attempt += 1
+        ids = fetch_active_pipeline_ids()
+        if ids is not None:
+            return ids
+        remaining = deadline - time.monotonic()
+        if remaining <= 0:
+            logger.error(
+                "Orchestrator did not answer active-pipelines query before "
+                "deadline; worktree cleanup cannot verify pipeline liveness",
+                attempts=attempt,
+                max_wait_seconds=max_wait_seconds,
+            )
+            return None
+        time.sleep(min(poll_interval_seconds, remaining))

gateway/tests/test_gateway_shutdown.py

@@ -16,6 +16,7 @@ def _run_gateway_main():
         patch.object(gateway, "get_session_manager") as mock_sm,
         patch.object(gateway, "get_active_docker_containers", return_value=set()),
         patch.object(gateway, "startup_cleanup", return_value=0),
+        patch.object(gateway, "wait_for_active_pipeline_ids", return_value=set()),
         patch.object(gateway, "get_launcher_secret", return_value="secret"),
         patch.object(gateway, "serve"),
         patch("signal.signal") as mock_signal,

gateway/tests/test_orchestrator_pipelines.py

@@ -0,0 +1,122 @@
+"""Tests for gateway/orchestrator_pipelines.py (#3070).
+
+The client's one hard invariant: failure is ``None``, never an empty
+set, so worktree cleanup can tell "verified nothing active" apart from
+"could not verify" and fail safe.
+"""
+
+from __future__ import annotations
+
+import io
+import json
+from unittest.mock import patch
+from urllib.error import URLError
+
+from orchestrator_pipelines import (
+    fetch_active_pipeline_ids,
+    wait_for_active_pipeline_ids,
+)
+
+
+def _raw_response(body_bytes: bytes) -> io.BytesIO:
+    """Build a stub urlopen response from the raw bytes the server would return.
+
+    Mirrors the ``with urlopen(...) as resp: resp.read().decode("utf-8")``
+    chain the client uses: the context-manager protocol is satisfied,
+    and ``read()`` returns the same bytes the network would.
+    """
+    body = io.BytesIO(body_bytes)
+    body.__enter__ = lambda *a: body  # type: ignore[method-assign]
+    body.__exit__ = lambda *a: False  # type: ignore[method-assign]
+    return body
+
+
+def _response(payload: dict) -> io.BytesIO:
+    return _raw_response(json.dumps(payload).encode("utf-8"))
+
+
+class TestFetchActivePipelineIds:
+    def test_returns_ids_on_success(self):
+        payload = {
+            "success": True,
+            "data": {
+                "pipelines": [
+                    {"id": "pipeline-c978dac3", "status": "awaiting_human"},
+                    {"id": "issue-3023", "status": "running"},
+                ]
+            },
+        }
+        with patch("orchestrator_pipelines.urlopen", return_value=_response(payload)) as mock:
+            ids = fetch_active_pipeline_ids()
+
+        assert ids == {"pipeline-c978dac3", "issue-3023"}
+        url = mock.call_args[0][0].full_url
+        assert url.endswith("/api/v1/pipelines?active_only=true")
+
+    def test_returns_empty_set_when_no_active_pipelines(self):
+        payload = {"success": True, "data": {"pipelines": []}}
+        with patch("orchestrator_pipelines.urlopen", return_value=_response(payload)):
+            assert fetch_active_pipeline_ids() == set()
+
+    def test_returns_none_on_network_error(self):
+        with patch("orchestrator_pipelines.urlopen", side_effect=URLError("refused")):
+            assert fetch_active_pipeline_ids() is None
+
+    def test_returns_none_on_malformed_body(self):
+        # Use the same response-shape helper as the success path so the
+        # full read()/decode() chain is exercised — only the *bytes* the
+        # server returns differ.
+        with patch("orchestrator_pipelines.urlopen", return_value=_raw_response(b"not json")):
+            assert fetch_active_pipeline_ids() is None
+
+    def test_returns_none_on_missing_data_key(self):
+        with patch(
+            "orchestrator_pipelines.urlopen",
+            return_value=_response({"success": True, "data": {}}),
+        ):
+            assert fetch_active_pipeline_ids() is None
+
+    def test_url_from_env(self, monkeypatch):
+        monkeypatch.setenv("EGG_ORCHESTRATOR_URL", "http://orch.test:9849/")
+        payload = {"success": True, "data": {"pipelines": []}}
+        with patch("orchestrator_pipelines.urlopen", return_value=_response(payload)) as mock:
+            fetch_active_pipeline_ids()
+        assert mock.call_args[0][0].full_url.startswith("http://orch.test:9849/api")
+
+
+class TestWaitForActivePipelineIds:
+    def test_returns_immediately_on_success(self):
+        with patch(
+            "orchestrator_pipelines.fetch_active_pipeline_ids",
+            return_value={"issue-1"},
+        ) as mock:
+            assert wait_for_active_pipeline_ids(max_wait_seconds=60) == {"issue-1"}
+        assert mock.call_count == 1
+
+    def test_retries_until_success(self):
+        with (
+            patch(
+                "orchestrator_pipelines.fetch_active_pipeline_ids",
+                side_effect=[None, None, set()],
+            ) as mock,
+            patch("orchestrator_pipelines.time.sleep") as sleep,
+        ):
+            assert wait_for_active_pipeline_ids(max_wait_seconds=60) == set()
+        assert mock.call_count == 3
+        assert sleep.call_count == 2
+
+    def test_returns_none_at_deadline(self):
+        with (
+            patch("orchestrator_pipelines.fetch_active_pipeline_ids", return_value=None),
+            patch(
+                "orchestrator_pipelines.time.monotonic",
+                side_effect=[0.0, 100.0],
+            ),
+        ):
+            assert wait_for_active_pipeline_ids(max_wait_seconds=50) is None
+
+    def test_deadline_from_env(self, monkeypatch):
+        monkeypatch.setenv("EGG_CLEANUP_ORCHESTRATOR_WAIT_SECONDS", "0")
+        with patch("orchestrator_pipelines.fetch_active_pipeline_ids", return_value=None) as mock:
+            assert wait_for_active_pipeline_ids() is None
+        assert mock.call_count == 1

gateway/tests/test_startup_k8s_guard.py

@@ -45,6 +45,7 @@ def test_k8s_orchestrator_url_guard(k8s_host, orch_url, should_exit):
         patch.object(gateway, "get_session_manager") as mock_sm,
         patch.object(gateway, "get_active_docker_containers", return_value=set()),
         patch.object(gateway, "startup_cleanup", return_value=0),
+        patch.object(gateway, "wait_for_active_pipeline_ids", return_value=set()),
         patch.object(gateway, "get_launcher_secret", return_value="secret"),
         patch.object(gateway, "serve"),
         patch("signal.signal"),