diff --git a/gateway/README.md b/gateway/README.md index b39cc84db6..5e41e58c38 100644 --- a/gateway/README.md +++ b/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. diff --git a/gateway/gateway.py b/gateway/gateway.py index 08ef874a99..6e5cc4b40a 100644 --- a/gateway/gateway.py +++ b/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)") diff --git a/gateway/orchestrator_pipelines.py b/gateway/orchestrator_pipelines.py new file mode 100644 index 0000000000..3c7029664d --- /dev/null +++ b/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)) diff --git a/gateway/tests/test_gateway_shutdown.py b/gateway/tests/test_gateway_shutdown.py index 036e43d83f..ae0e55f2c0 100644 --- a/gateway/tests/test_gateway_shutdown.py +++ b/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, diff --git a/gateway/tests/test_orchestrator_pipelines.py b/gateway/tests/test_orchestrator_pipelines.py new file mode 100644 index 0000000000..9c9c6f09e0 --- /dev/null +++ b/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 diff --git a/gateway/tests/test_startup_k8s_guard.py b/gateway/tests/test_startup_k8s_guard.py index 7656d6e34b..02e8b8f16f 100644 --- a/gateway/tests/test_startup_k8s_guard.py +++ b/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"), diff --git a/gateway/tests/test_worktree_manager.py b/gateway/tests/test_worktree_manager.py index bbc1d97a2b..fcb6506b7e 100644 --- a/gateway/tests/test_worktree_manager.py +++ b/gateway/tests/test_worktree_manager.py @@ -284,6 +284,80 @@ def test_cleanup_skips_in_flight_worktrees(self, manager, temp_dirs): assert removed == 0 assert container_dir.exists() + def test_cleanup_preserves_active_pipeline_worktrees(self, manager, temp_dirs): + """Worktrees of live pipelines survive even with zero containers. + + Regression for #3070: a pipeline parked at a HITL gate has no + running containers and no sessions, so the container-derived active + set is empty — its worktrees (pipeline-level and per-agent) must be + preserved via the orchestrator-reported pipeline list instead. + """ + worktree_base, _ = temp_dirs + + preserved = [] + for name in ("pipeline-c978dac3", "pipeline-c978dac3-refiner"): + d = worktree_base / name + d.mkdir(parents=True) + (d / "webapp").mkdir() + preserved.append(d) + orphan = worktree_base / "pipeline-deadbeef-coder" + orphan.mkdir(parents=True) + (orphan / "webapp").mkdir() + + manager.cleanup_orphaned_worktrees(set(), active_pipeline_ids={"pipeline-c978dac3"}) + + for d in preserved: + assert d.exists(), f"{d.name} should have been preserved" + assert not orphan.exists() + + def test_cleanup_pipeline_anchor_is_delimiter_bound(self, manager, temp_dirs): + """``issue-302`` must not anchor ``issue-3023-*`` worktrees.""" + worktree_base, _ = temp_dirs + + near_miss = worktree_base / "issue-3023-coder" + near_miss.mkdir(parents=True) + (near_miss / "egg").mkdir() + + manager.cleanup_orphaned_worktrees(set(), active_pipeline_ids={"issue-302"}) + + assert not near_miss.exists() + + def test_cleanup_never_deletes_branches(self, manager, temp_dirs): + """Orphan sweeps must not delete branches (#3070). + + The sweep cannot know whether the branch's work was pushed; + deleting it strands commits as dangling. Branch deletion belongs + to the explicit per-container teardown paths only. + """ + worktree_base, _ = temp_dirs + + orphan = worktree_base / "pipeline-deadbeef" + orphan.mkdir(parents=True) + (orphan / "webapp").mkdir() + + with patch.object(manager, "remove_worktree", wraps=manager.remove_worktree) as spy: + manager.cleanup_orphaned_worktrees(set(), active_pipeline_ids=set()) + + assert spy.call_count >= 1 + for call in spy.call_args_list: + assert call.kwargs.get("delete_branch") is False + + def test_list_orphan_worktree_dirs_respects_pipeline_anchors(self, manager, temp_dirs): + """Dry-run listing mirrors the sweep's pipeline-anchor skip.""" + worktree_base, _ = temp_dirs + + live = worktree_base / "pipeline-c978dac3-overseer" + live.mkdir(parents=True) + orphan = worktree_base / "pipeline-deadbeef" + orphan.mkdir(parents=True) + + orphans = manager.list_orphan_worktree_dirs( + set(), active_pipeline_ids={"pipeline-c978dac3"} + ) + + assert str(orphan) in orphans + assert str(live) not in orphans + class TestGetActiveDockerContainers: """Tests for get_active_docker_containers helper.""" @@ -350,12 +424,36 @@ def test_with_active_containers(self, tmp_path): mock_instance.cleanup_orphaned_worktrees.return_value = 1 MockManager.return_value = mock_instance - removed = startup_cleanup(active_containers={"active-container"}) + removed = startup_cleanup( + active_containers={"active-container"}, + active_pipeline_ids=set(), + ) assert removed == 1 mock_instance.cleanup_orphaned_worktrees.assert_called_once_with( - {"active-container"}, None + {"active-container"}, None, active_pipeline_ids=set() ) + def test_skips_sweep_when_pipeline_liveness_unknown(self): + """No sweep without a verified pipeline list (#3070 fail-safe). + + ``active_pipeline_ids=None`` means the orchestrator could not be + reached — sweeping on container liveness alone is what deleted a + HITL-parked pipeline's worktree, contract, and branches in #3070. + """ + from worktree_manager import startup_cleanup + + with patch("worktree_manager.WorktreeManager") as MockManager: + mock_instance = MagicMock() + MockManager.return_value = mock_instance + + removed = startup_cleanup(active_containers=set(), active_pipeline_ids=None) + + assert removed == 0 + mock_instance.cleanup_orphaned_worktrees.assert_not_called() + # The safe maintenance passes still run. + mock_instance.prune_stale_worktrees.assert_called_once() + mock_instance.cleanup_orphaned_pack_files.assert_called_once() + def test_with_none_uses_docker(self): """Falls back to querying Docker when active_containers is None.""" from worktree_manager import startup_cleanup @@ -369,7 +467,7 @@ def test_with_none_uses_docker(self): "worktree_manager.get_active_docker_containers", return_value={"container-1"}, ): - removed = startup_cleanup(active_containers=None) + removed = startup_cleanup(active_containers=None, active_pipeline_ids=set()) assert removed == 0 def test_with_empty_set(self): @@ -381,7 +479,7 @@ def test_with_empty_set(self): mock_instance.cleanup_orphaned_worktrees.return_value = 3 MockManager.return_value = mock_instance - removed = startup_cleanup(active_containers=set()) + removed = startup_cleanup(active_containers=set(), active_pipeline_ids=set()) assert removed == 3 @@ -2647,7 +2745,7 @@ def test_calls_prune_after_orphan_cleanup(self): mock_instance.prune_stale_worktrees.return_value = 1 MockManager.return_value = mock_instance - startup_cleanup(active_containers=set()) + startup_cleanup(active_containers=set(), active_pipeline_ids=set()) mock_instance.cleanup_orphaned_worktrees.assert_called_once() mock_instance.prune_stale_worktrees.assert_called_once() @@ -2663,7 +2761,7 @@ def test_prune_failure_does_not_prevent_cleanup(self): MockManager.return_value = mock_instance # Should not raise - removed = startup_cleanup(active_containers=set()) + removed = startup_cleanup(active_containers=set(), active_pipeline_ids=set()) assert removed == 2 def test_calls_pack_cleanup_after_prune(self): @@ -2677,7 +2775,7 @@ def test_calls_pack_cleanup_after_prune(self): mock_instance.cleanup_orphaned_pack_files.return_value = (5, 1024000) MockManager.return_value = mock_instance - startup_cleanup(active_containers=set()) + startup_cleanup(active_containers=set(), active_pipeline_ids=set()) mock_instance.cleanup_orphaned_worktrees.assert_called_once() mock_instance.prune_stale_worktrees.assert_called_once() @@ -2694,7 +2792,7 @@ def test_pack_cleanup_failure_does_not_prevent_startup(self): mock_instance.cleanup_orphaned_pack_files.side_effect = RuntimeError("disk error") MockManager.return_value = mock_instance - removed = startup_cleanup(active_containers=set()) + removed = startup_cleanup(active_containers=set(), active_pipeline_ids=set()) assert removed == 1 diff --git a/gateway/tests/test_worktree_prune_route.py b/gateway/tests/test_worktree_prune_route.py index 97dbb5a677..4049b28b74 100644 --- a/gateway/tests/test_worktree_prune_route.py +++ b/gateway/tests/test_worktree_prune_route.py @@ -67,6 +67,18 @@ def _reset_prune_lock(): pass +@pytest.fixture(autouse=True) +def _verified_pipeline_liveness(): + """Stub the orchestrator's active-pipelines answer (#3070). + + The route refuses (503) when pipeline liveness cannot be verified, so + every pre-existing test runs with a verified-empty answer. Tests that + exercise the unverified path override the patch themselves. + """ + with patch.object(gateway, "fetch_active_pipeline_ids", return_value=set()) as mock: + yield mock + + class TestWorktreesPruneAuth: """Launcher auth is the only thing standing between an orchestrator- side compromise and wholesale worktree removal. Belt-and-suspenders @@ -137,6 +149,48 @@ def test_dry_run_with_no_orphans_returns_empty_plan(self, client, launcher_auth_ assert data["removed_count"] == 0 +class TestWorktreesPrunePipelineLiveness: + """Pipeline liveness is mandatory input to the sweep (#3070).""" + + def test_returns_503_when_liveness_unverifiable( + self, client, launcher_auth_headers, fake_manager + ): + """Orchestrator unreachable → refuse, never sweep blind.""" + with ( + patch.object(gateway, "get_worktree_manager", return_value=fake_manager), + patch.object(gateway, "fetch_active_pipeline_ids", return_value=None), + ): + response = client.post( + "/api/v1/worktrees/prune", + json={"dry_run": False}, + headers=launcher_auth_headers, + ) + assert response.status_code == 503 + fake_manager.cleanup_orphaned_worktrees.assert_not_called() + fake_manager.list_orphan_worktree_dirs.assert_not_called() + + def test_active_pipeline_ids_forwarded_to_sweep( + self, client, launcher_auth_headers, fake_manager + ): + """The orchestrator's answer reaches both enumeration and cleanup.""" + live = {"pipeline-c978dac3"} + with ( + patch.object(gateway, "get_worktree_manager", return_value=fake_manager), + patch.object(gateway, "fetch_active_pipeline_ids", return_value=live), + ): + response = client.post( + "/api/v1/worktrees/prune", + json={"dry_run": False}, + headers=launcher_auth_headers, + ) + assert response.status_code == 200 + _args, kwargs = fake_manager.cleanup_orphaned_worktrees.call_args + assert kwargs.get("active_pipeline_ids") == live + _largs, lkwargs = fake_manager.list_orphan_worktree_dirs.call_args + assert lkwargs.get("active_pipeline_ids") == live + assert response.get_json()["data"]["active_pipelines_count"] == 1 + + class TestWorktreesPruneMutation: """Non-dry-run path actually cleans up orphaned dirs.""" @@ -150,7 +204,7 @@ def test_dry_run_false_calls_cleanup( orphan_path.mkdir() fake_manager.list_orphan_worktree_dirs.return_value = [str(orphan_path)] - def _fake_cleanup(active_containers): + def _fake_cleanup(active_containers, active_pipeline_ids=None): orphan_path.rmdir() return 1 diff --git a/gateway/worktree_manager.py b/gateway/worktree_manager.py index f4a5bf46c2..4071cd565a 100644 --- a/gateway/worktree_manager.py +++ b/gateway/worktree_manager.py @@ -1616,10 +1616,32 @@ def list_worktrees_for_pipeline(self, pipeline_id: str) -> list[WorktreeInfo]: return results + @staticmethod + def _is_pipeline_anchored(container_id: str, active_pipeline_ids: set[str]) -> bool: + """True when *container_id* belongs to an active pipeline. + + Worktree dir names are either the pipeline ID itself (the + pipeline-level worktree) or ``{pipeline_id}-{suffix}`` (per-agent + worktrees, e.g. ``pipeline-c978dac3-refiner``, + ``issue-3023-slice-1-coder``). The delimiter anchor prevents + ``issue-302`` from matching ``issue-3023-*``. + + Intentionally looser than ``list_worktrees_for_pipeline``'s + ``{pid}-[a-z_]+`` regex (#1865): slice-scoped suffixes like + ``issue-3023-slice-1-coder`` contain digits and would be + falsely orphaned by the stricter pattern. Over-preserving on + one-active-pipeline-ID-prefixes-another is the right side of + the fail-safe principle here (#3070). + """ + return any( + container_id == pid or container_id.startswith(f"{pid}-") for pid in active_pipeline_ids + ) + def cleanup_orphaned_worktrees( self, active_containers: set[str], session_manager: Any | None = None, + active_pipeline_ids: set[str] | None = None, ) -> int: """ Remove worktrees for containers that no longer exist. @@ -1633,6 +1655,13 @@ def cleanup_orphaned_worktrees( Args: active_containers: Set of currently active container IDs session_manager: Optional SessionManager for session auto-commit on cleanup + active_pipeline_ids: IDs of non-terminal pipelines per the + orchestrator. Worktrees anchored to them are preserved even + with no live container — a pipeline parked at a HITL gate or + between phases has no containers and no sessions, but its + worktree holds the contract and any un-pushed work (#3070). + ``None`` means liveness could not be verified; callers must + not invoke this sweep in that case (see ``startup_cleanup``). Returns: Number of worktrees removed @@ -1652,6 +1681,12 @@ def cleanup_orphaned_worktrees( if container_id in active_containers: continue + # Skip worktrees belonging to a live pipeline, container or not. + if active_pipeline_ids and self._is_pipeline_anchored( + container_id, active_pipeline_ids + ): + continue + # Skip worktrees that this process just created. create_worktree # populates ``_active_worktrees[container_id]`` before returning, # so any per-agent worktree made during this gateway's lifetime @@ -1685,10 +1720,17 @@ def cleanup_orphaned_worktrees( error=str(e), ) - # Remove each worktree + # Remove each worktree. Never delete branches here: an orphan + # sweep cannot know whether the work they point at was pushed, + # and deleting them turns a recoverable mistake into data loss — + # in #3070 this left an operator-approved analysis reachable only + # as a dangling commit. Branch deletion stays with the explicit + # per-container teardown paths. for worktree in list(container_dir.iterdir()): if worktree.is_dir(): - result = self.remove_worktree(container_id, worktree.name, force=True) + result = self.remove_worktree( + container_id, worktree.name, force=True, delete_branch=False + ) if result.success: removed += 1 else: @@ -1842,12 +1884,18 @@ def git_worktree_prune_all(self) -> dict[str, list[str]]: result[repo_name] = paths return result - def list_orphan_worktree_dirs(self, active_containers: set[str]) -> list[str]: + def list_orphan_worktree_dirs( + self, + active_containers: set[str], + active_pipeline_ids: set[str] | None = None, + ) -> list[str]: """Return absolute paths of container dirs considered orphaned. A container dir under ``worktree_base`` is considered orphaned - when its name is not in *active_containers*. Each returned path - is first validated via :func:`Path.resolve` + + when its name is not in *active_containers* and it is not anchored + to an active pipeline (mirrors ``cleanup_orphaned_worktrees`` so + dry-run output matches what the sweep would do). Each returned + path is first validated via :func:`Path.resolve` + ``is_relative_to(self.worktree_base)`` to protect against symlink-based traversal. """ @@ -1879,6 +1927,8 @@ def list_orphan_worktree_dirs(self, active_containers: set[str]) -> list[str]: continue if child.name in active_containers: continue + if active_pipeline_ids and self._is_pipeline_anchored(child.name, active_pipeline_ids): + continue # Mirror the _active_worktrees guard from cleanup_orphaned_worktrees # so dry-run output accurately reflects what cleanup would skip. with self._lock: @@ -2064,7 +2114,10 @@ def cleanup_clean_worktree(self, container_id: str, repo_name: str) -> bool: return removal.success def cleanup_stale_pipeline_worktrees( - self, max_age_hours: int = 48, active_containers: set[str] | None = None + self, + max_age_hours: int = 48, + active_containers: set[str] | None = None, + active_pipeline_ids: set[str] | None = None, ) -> int: """Remove worktrees older than max_age_hours regardless of state. @@ -2073,12 +2126,24 @@ def cleanup_stale_pipeline_worktrees( TODO: Wire this into the orchestrator's maintenance loop. Currently only called from tests — not yet connected to production scheduling. + Whoever wires this up must pass ``active_pipeline_ids`` from + ``orchestrator_pipelines.fetch_active_pipeline_ids`` — otherwise a + long-parked HITL pipeline whose mtimes have aged past ``max_age_hours`` + becomes the next #3070 (an idle parked pipeline has no container and + no session activity; only the orchestrator-derived set distinguishes + it from a crashed leftover). Args: max_age_hours: Worktrees inactive for longer than this are removed. active_containers: Set of running container IDs. Worktrees with active containers are never deleted. If None, fetched via ``get_active_docker_containers()``. + active_pipeline_ids: IDs of non-terminal pipelines per the + orchestrator. Worktrees anchored to them are preserved even + when their mtimes look stale — mirrors + ``cleanup_orphaned_worktrees`` (#3070). ``None`` means + "skip the anchor check"; pass a verified set when wiring this + into production scheduling. Returns: Number of worktrees removed. @@ -2098,6 +2163,11 @@ def cleanup_stale_pipeline_worktrees( # Skip worktrees whose containers are still running. if entry.name in active_containers: continue + # Skip worktrees anchored to an active pipeline — a parked HITL + # pipeline has no container, so age-based deletion would strand + # its contract and un-pushed work (#3070). + if active_pipeline_ids and self._is_pipeline_anchored(entry.name, active_pipeline_ids): + continue try: # Use .git/index (updated on every commit/checkout) as the # staleness signal instead of walking the entire tree, which @@ -2140,8 +2210,15 @@ def cleanup_stale_pipeline_worktrees( if newest_mtime < cutoff: for repo_dir in entry.iterdir(): if repo_dir.is_dir(): + # Never delete branches in a periodic age sweep: + # the sweep can't know whether the work they + # point at was pushed, and deleting them turns a + # recoverable mistake into data loss. Branch + # deletion stays with explicit per-container + # teardown paths (mirrors + # ``cleanup_orphaned_worktrees``; #3070). removal_result = self.remove_worktree( - entry.name, repo_dir.name, force=True, delete_branch=True + entry.name, repo_dir.name, force=True, delete_branch=False ) if removal_result.success: removed += 1 @@ -2195,6 +2272,7 @@ def get_active_docker_containers() -> set[str]: def startup_cleanup( active_containers: set[str] | None = None, session_manager: Any | None = None, + active_pipeline_ids: set[str] | None = None, ) -> int: """ Clean up orphaned worktrees on gateway startup. @@ -2208,6 +2286,15 @@ def startup_cleanup( None, falls back to querying Docker (which may not be available inside the gateway container). session_manager: Optional SessionManager for session auto-commit on cleanup + active_pipeline_ids: IDs of non-terminal pipelines per the + orchestrator (``orchestrator_pipelines.fetch_active_pipeline_ids``). + ``None`` means pipeline liveness could not be verified — the + orphan sweep is SKIPPED entirely rather than run blind, because + container liveness alone cannot distinguish a crashed leftover + from a pipeline parked at a HITL gate (#3070; a redeploy swept + every worktree, contract included, with ``active_containers=0``). + Pass an empty set only when the orchestrator confirmed nothing + is active. Returns: Number of orphaned worktrees removed @@ -2216,12 +2303,25 @@ def startup_cleanup( if active_containers is None: active_containers = get_active_docker_containers() - logger.info( - "Running startup worktree cleanup", - active_containers=len(active_containers), - ) - - removed = manager.cleanup_orphaned_worktrees(active_containers, session_manager) + if active_pipeline_ids is None: + logger.error( + "Skipping orphaned-worktree sweep: pipeline liveness unknown " + "(orchestrator unreachable?); stale worktrees will accumulate " + "until the next startup or an operator-run prune", + active_containers=len(active_containers), + ) + removed = 0 + else: + logger.info( + "Running startup worktree cleanup", + active_containers=len(active_containers), + active_pipelines=len(active_pipeline_ids), + ) + removed = manager.cleanup_orphaned_worktrees( + active_containers, + session_manager, + active_pipeline_ids=active_pipeline_ids, + ) if removed > 0: logger.info(f"Cleaned up {removed} orphaned worktree(s)") diff --git a/orchestrator/tests/test_worktree_hitl.py b/orchestrator/tests/test_worktree_hitl.py index e20b8a4fc5..c6e17f1f2e 100644 --- a/orchestrator/tests/test_worktree_hitl.py +++ b/orchestrator/tests/test_worktree_hitl.py @@ -228,8 +228,11 @@ def test_removes_old_worktrees(self, tmp_path): ) assert removed == 1 + # delete_branch=False: an age-based sweep cannot tell whether the + # branch the worktree points at was pushed, so deleting it would + # strand reachable-only-as-dangling commits (#3070). mock_remove.assert_called_once_with( - "old-container", "myrepo", force=True, delete_branch=True + "old-container", "myrepo", force=True, delete_branch=False ) def test_preserves_recent_worktrees(self, tmp_path): @@ -317,3 +320,73 @@ def test_empty_base_returns_zero(self, tmp_path): manager = self._make_manager(tmp_path) removed = manager.cleanup_stale_pipeline_worktrees(active_containers=set()) assert removed == 0 + + def test_preserves_active_pipeline_worktrees(self, tmp_path): + """Should NOT remove worktrees anchored to an active pipeline, even when stale. + + A pipeline parked at a HITL gate has no container and no recent + mtime activity — only the orchestrator-derived active-pipeline + set distinguishes it from a crashed leftover (#3070). + """ + manager = self._make_manager(tmp_path) + + # Pipeline-level worktree (e.g. ``pipeline-c978dac3``) + pipeline_id = "pipeline-c978dac3" + container_dir = manager.worktree_base / pipeline_id + repo_dir = container_dir / "myrepo" + repo_dir.mkdir(parents=True) + + git_admin_dir = tmp_path / "git-admin" / "worktrees" / pipeline_id + git_admin_dir.mkdir(parents=True) + (repo_dir / ".git").write_text(f"gitdir: {git_admin_dir}\n") + (git_admin_dir / "index").touch() + (git_admin_dir / "HEAD").write_text("ref: refs/heads/egg/parked/work\n") + + # Set mtime to 72 hours ago — would normally be deleted + old_time = time.time() - (72 * 3600) + os.utime(str(container_dir), (old_time, old_time)) + os.utime(str(git_admin_dir / "index"), (old_time, old_time)) + os.utime(str(git_admin_dir / "HEAD"), (old_time, old_time)) + + with patch.object(manager, "remove_worktree") as mock_remove: + removed = manager.cleanup_stale_pipeline_worktrees( + max_age_hours=48, + active_containers=set(), + active_pipeline_ids={pipeline_id}, + ) + + assert removed == 0 + mock_remove.assert_not_called() + + def test_preserves_per_agent_active_pipeline_worktrees(self, tmp_path): + """Per-agent worktrees (``{pid}-{role}``) anchored to an active pipeline are preserved.""" + manager = self._make_manager(tmp_path) + + pipeline_id = "issue-3023" + # Per-agent worktree shape + per_agent_name = f"{pipeline_id}-coder" + container_dir = manager.worktree_base / per_agent_name + repo_dir = container_dir / "myrepo" + repo_dir.mkdir(parents=True) + + git_admin_dir = tmp_path / "git-admin" / "worktrees" / per_agent_name + git_admin_dir.mkdir(parents=True) + (repo_dir / ".git").write_text(f"gitdir: {git_admin_dir}\n") + (git_admin_dir / "index").touch() + (git_admin_dir / "HEAD").write_text("ref: refs/heads/egg/coder/work\n") + + # Stale mtimes + old_time = time.time() - (72 * 3600) + os.utime(str(container_dir), (old_time, old_time)) + os.utime(str(git_admin_dir / "index"), (old_time, old_time)) + os.utime(str(git_admin_dir / "HEAD"), (old_time, old_time)) + + with patch.object(manager, "remove_worktree") as mock_remove: + removed = manager.cleanup_stale_pipeline_worktrees( + max_age_hours=48, + active_containers=set(), + active_pipeline_ids={pipeline_id}, + ) + + assert removed == 0 + mock_remove.assert_not_called()