diff --git a/orchestrator/gateway_client.py b/orchestrator/gateway_client.py index 01014b17e4..b2816aeefe 100644 --- a/orchestrator/gateway_client.py +++ b/orchestrator/gateway_client.py @@ -1693,6 +1693,124 @@ def _sha_is_ancestor( ) return False + def is_slice_branch_merged_into_parent( + self, + pipeline_id: str, + repo_path: str, + *, + integration_branch: str, + parent_branch: str, + agent_role: str = "coder", + mode: Literal["public", "private"] = "public", + ) -> bool: + """Return True iff the slice integration branch's tip on origin is + already reachable from ``parent_branch``'s tip on origin. + + This is the #2549 "slice already merged" signal: after the slice's + PR is merged into the parent, the integration branch's old tip is + an ancestor of the parent's new tip. The inverse direction of the + #2512 restart-recovery check — and the case that previously caused + ``create_slice_integration_branch`` to fall through to a non-fast- + forward push and fail the slice (and cascade-fail the phase). + + Returns False on any of: + + * Either branch is missing on origin (nothing to compare against). + * The integration branch tip equals the parent tip (``==`` is + neither "merged" nor "diverged"; just a no-op state — let the + regular create path handle it as a fast-forward no-op). + * The ancestry check itself fails (gateway down, missing object + after a flaky fetch). In that case we return False so the + caller falls through to the existing create path rather than + silently skipping the slice. + + The transport mirrors :meth:`create_slice_integration_branch`: + a single synthetic launcher-authenticated session shared across + ls-remote, fetch, and the merge-base call. + """ + if not integration_branch or not parent_branch: + return False + if integration_branch == parent_branch: + return False + + temp_container_id = ( + f"{pipeline_id}-slice-merged-check-{integration_branch.replace('/', '-')}" + ) + session_token: str | None = None + try: + session = self.register_session( + container_id=temp_container_id, + container_ip=self.self_ip, + mode=mode, + pipeline_id=pipeline_id, + agent_role=agent_role, + branch=integration_branch, + synthetic=True, + ) + session_token = session.session_token + + parent_sha = self.get_remote_branch_sha( + pipeline_id, + repo_path, + f"refs/heads/{parent_branch}", + mode=mode, + bearer_token=session_token, + ) + existing_sha = self.get_remote_branch_sha( + pipeline_id, + repo_path, + f"refs/heads/{integration_branch}", + mode=mode, + bearer_token=session_token, + ) + if not parent_sha or not existing_sha: + return False + if parent_sha == existing_sha: + return False + + # Both refs must be locally reachable for ``merge-base + # --is-ancestor`` to evaluate without errors. Best-effort: + # if either fetch fails the merge-base call will return + # False (missing object → returncode != 0) and we degrade + # to "not merged", which matches the safe default. + self.fetch_branch( + pipeline_id, + repo_path, + args=[f"+refs/heads/{parent_branch}:refs/remotes/origin/{parent_branch}"], + mode=mode, + bearer_token=session_token, + ) + self.fetch_branch( + pipeline_id, + repo_path, + args=[f"+refs/heads/{integration_branch}:refs/remotes/origin/{integration_branch}"], + mode=mode, + bearer_token=session_token, + ) + + return self._sha_is_ancestor( + pipeline_id, + repo_path, + existing_sha, + parent_sha, + bearer_token=session_token, + ) + except Exception as exc: # noqa: BLE001 + logger.warning( + "is_slice_branch_merged_into_parent: gateway request failed", + pipeline_id=pipeline_id, + integration_branch=integration_branch, + parent_branch=parent_branch, + error=str(exc), + ) + return False + finally: + if session_token: + try: + self.delete_session(session_token) + except Exception: + pass + def create_slice_integration_branch( self, pipeline_id: str, diff --git a/orchestrator/routes/pipelines.py b/orchestrator/routes/pipelines.py index d5be4613a4..59624b6903 100644 --- a/orchestrator/routes/pipelines.py +++ b/orchestrator/routes/pipelines.py @@ -12319,15 +12319,13 @@ def _contract_loader() -> Any: except Exception: # noqa: BLE001 return None - reconciler_thread, reconciler_stop = _start_stacked_pr_reconciler( - pipeline_id, - _contract_loader, - spawner.gateway, - pipeline, - worktree_repo_path=worktree_repo_path, - repo=getattr(pipeline, "repo", None), - ) - + # #2549 reviewer note: defer starting the stacked-PR reconciler + # until after the bootstrap reconciliation pass so an unhandled + # exception during bootstrap (e.g. a hard ImportError of + # ``SliceStatus`` or a programming error in the pass) cannot leak + # the daemon thread. The reconciler does not depend on bootstrap + # state, so its start is safe to move after the pass; the existing + # ``finally`` at the bottom of the run loop owns its teardown. aggregate_logs: list[str] = [] overall_exit = 0 poll_interval = 5.0 @@ -12337,6 +12335,170 @@ def _contract_loader() -> Any: except ImportError: import global_slice_admit # type: ignore[no-redef] + try: + from orchestrator.peer_consensus import ( + remove_peer_consensus_tracker, + ) + except ImportError: + from peer_consensus import ( # type: ignore[no-redef] + remove_peer_consensus_tracker, + ) + + try: + from state_store import get_pipeline_state_lock + except ImportError: + from orchestrator.state_store import ( # type: ignore[no-redef] + get_pipeline_state_lock, + ) + + from egg_contracts.models import SliceStatus + + def _persist_slice_status_complete(slice_id: str) -> None: + """Mark ``slice_id`` as ``SliceStatus.COMPLETE`` on the contract. + + #2549 — durable record of slice completion. The + ``Slice.status`` field has had a ``COMPLETE`` value since + the original schema, but until #2549 nothing wrote it; the + #2470 ``restart_agent`` reader at line 2653 was effectively + dead code. This helper closes that gap so: + + * Subsequent ``start_pipeline`` calls see merged slices as + COMPLETE on the contract and skip them in the bootstrap + reconciliation pass below — no GitHub round-trip needed. + * The #2470 ``restart_agent`` parent-complete fallback + finally has a real signal to read. + + Best-effort: if the lock or save fails, the in-memory + scheduler state still reflects completion and the slice + won't run this pass; the next start_pipeline will + re-detect via the merged-detection helper. + """ + try: + with get_pipeline_state_lock(pipeline_id): + contract_local = load_contract(pipeline_id, worktree_repo_path) + for s in contract_local.slices: + if s.id == slice_id: + s.status = SliceStatus.COMPLETE + break + save_contract(contract_local, worktree_repo_path) + except Exception as save_err: # noqa: BLE001 + logger.warning( + "Failed to persist slice.status=COMPLETE", + pipeline_id=pipeline_id, + slice_id=slice_id, + error=str(save_err), + ) + + # #2549 — bootstrap reconciliation pass. Before the run loop picks up + # any slices, fold in two sources of "this slice is already done" + # state that the scheduler (a pure rebuild from ``contract.slices``) + # cannot see on its own: + # + # (A) Slices that the contract already records as + # ``SliceStatus.COMPLETE``. Once #2549 starts writing this + # field on success, future restarts can trust it directly + # without a GitHub round-trip. + # + # (B) Slices whose integration branch on origin is reachable from + # their parent's tip — i.e. their PR has been merged. This + # handles the literal #2549 repro (operator merges slice-1's + # PR, runs ``start_pipeline`` to resume) AND any slice whose + # completion was committed before #2549's writer landed. On a + # hit, also persist (A) so subsequent restarts hit the cheap + # path. + # + # Without this pass, the scheduler would yield every slice as READY + # on its first ``iter_ready`` tick and ``create_slice_integration_ + # branch`` would attempt to push parent_sha onto an existing slice + # ref whose tip is now an ancestor of parent — a non-fast-forward + # rejection that previously failed the slice and cascaded the + # phase. Both layers (A+B) are best-effort: a failure in this pass + # silently falls through to the existing run loop, preserving the + # pre-#2549 behaviour as the floor. + bootstrap_complete: list[str] = [] + bootstrap_merged: list[str] = [] + + # Layer (A): cheap, no I/O. Trust contract-recorded COMPLETE status. + layer_b_candidates = [] + for s in slices: + if s.status == SliceStatus.COMPLETE: + scheduler.record_complete(s.id) + bootstrap_complete.append(s.id) + continue + layer_b_candidates.append(s) + + # Layer (B): origin-side detection for slices not yet recorded as + # COMPLETE on the contract. Each helper call uses its own synthetic + # gateway session, so we parallelise across slices to keep startup + # latency bounded as forests grow. Cap workers so a large forest + # doesn't burst against the gateway. + if pipeline.repo and layer_b_candidates: + + def _bootstrap_check_one(slice_obj: Any) -> tuple[str, bool]: + # Prefer the parent branch the slice was actually forked + # off of (recorded by ``_run_one_slice_inner``). Falls back + # to the dependency-derived parent for slices that never + # made it through ``_run_one_slice_inner`` (e.g. fresh + # contract on first run). Both should agree today, but a + # future re-plan that mutates ``dependencies`` post-creation + # would diverge — preferring the recorded value future- + # proofs the check. + if slice_obj.parent_branch_at_creation: + parent_branch_for_check = slice_obj.parent_branch_at_creation + elif slice_obj.dependencies: + parent_branch_for_check = f"{issue_branch}/{slice_obj.dependencies[0]}" + else: + parent_branch_for_check = pipeline_branch + integration_branch_for_check = f"{issue_branch}/{slice_obj.id}" + try: + merged = spawner.gateway.is_slice_branch_merged_into_parent( + pipeline_id, + str(worktree_repo_path), + integration_branch=integration_branch_for_check, + parent_branch=parent_branch_for_check, + agent_role="coder", + mode=gateway_mode, # type: ignore[arg-type] + ) + except Exception as detect_err: # noqa: BLE001 + logger.warning( + "Bootstrap merged-detection raised; treating slice as not-merged", + pipeline_id=pipeline_id, + slice_id=slice_obj.id, + error=str(detect_err), + ) + return slice_obj.id, False + return slice_obj.id, bool(merged) + + max_workers = min(len(layer_b_candidates), 8) + with concurrent.futures.ThreadPoolExecutor( + max_workers=max_workers, + thread_name_prefix=f"slice-bootstrap-{pipeline_id}", + ) as bootstrap_pool: + results = list(bootstrap_pool.map(_bootstrap_check_one, layer_b_candidates)) + + for slice_id, already_merged in results: + if already_merged: + scheduler.record_complete(slice_id) + _persist_slice_status_complete(slice_id) + bootstrap_merged.append(slice_id) + + if bootstrap_complete or bootstrap_merged: + logger.info( + "Slice bootstrap reconciliation marked slices complete", + pipeline_id=pipeline_id, + already_complete_on_contract=bootstrap_complete, + detected_merged_on_origin=bootstrap_merged, + ) + + reconciler_thread, reconciler_stop = _start_stacked_pr_reconciler( + pipeline_id, + _contract_loader, + spawner.gateway, + pipeline, + worktree_repo_path=worktree_repo_path, + repo=getattr(pipeline, "repo", None), + ) + try: while not scheduler.all_done(): # 1. Snapshot ready slices for this tick. @@ -12376,21 +12538,6 @@ def _contract_loader() -> Any: # on the scheduler from inside ``_run_one_slice`` so the # cascade machinery sees the same wall-clock as the run # loop. - try: - from orchestrator.peer_consensus import ( - remove_peer_consensus_tracker, - ) - except ImportError: - from peer_consensus import ( # type: ignore[no-redef] - remove_peer_consensus_tracker, - ) - - try: - from state_store import get_pipeline_state_lock - except ImportError: - from orchestrator.state_store import ( # type: ignore[no-redef] - get_pipeline_state_lock, - ) def _run_one_slice(slice_id: str, parent_slice_id: str | None) -> tuple[int, str]: # Release the global-admission slot when the slice @@ -12430,6 +12577,54 @@ def _run_one_slice_inner(slice_id: str, parent_slice_id: str | None) -> tuple[in error=str(save_err), ) + # #2549 race protection: a slice's PR can be merged + # between the bootstrap reconciliation pass and this + # spawn (e.g. operator merges slice-1 while slice-2 is + # still queued). When that happens, the integration + # branch's old tip is reachable from the parent's new + # tip, and the create-branch push below would be + # rejected as non-fast-forward (cascading the slice and + # its descendants to FAILED). Detect that case here and + # skip directly to COMPLETE — same effect as the + # bootstrap reconciliation pass, just on a slice that + # transitioned during the run. + if pipeline.repo: + try: + already_merged = spawner.gateway.is_slice_branch_merged_into_parent( + pipeline_id, + str(worktree_repo_path), + integration_branch=integration_branch, + parent_branch=parent_branch, + agent_role="coder", + mode=gateway_mode, # type: ignore[arg-type] + ) + except Exception as detect_err: # noqa: BLE001 + logger.warning( + "Slice merged-detection raised; treating as not-merged", + pipeline_id=pipeline_id, + slice_id=slice_id, + error=str(detect_err), + ) + already_merged = False + if already_merged: + logger.info( + "Slice already merged into parent on origin — skipping spawn (#2549)", + pipeline_id=pipeline_id, + slice_id=slice_id, + integration_branch=integration_branch, + parent_branch=parent_branch, + ) + scheduler.record_complete(slice_id) + _persist_slice_status_complete(slice_id) + try: + remove_peer_consensus_tracker(pipeline_id, slice_id) + except Exception: # noqa: BLE001 + pass + return 0, ( + f"slice {slice_id}: already merged into " + f"{parent_branch} on origin — skipped" + ) + # #2137 TASK-4-2: create the slice integration branch # on origin BEFORE spawning containers. Push # ``parent_branch:refs/heads/integration_branch`` @@ -12663,6 +12858,7 @@ def _run_one_slice_inner(slice_id: str, parent_slice_id: str | None) -> tuple[in ) scheduler.record_complete(slice_id) + _persist_slice_status_complete(slice_id) try: remove_peer_consensus_tracker(pipeline_id, slice_id) except Exception: # noqa: BLE001 diff --git a/orchestrator/tests/test_create_slice_integration_branch.py b/orchestrator/tests/test_create_slice_integration_branch.py index 9aa7ab0308..7f5eef2dbb 100644 --- a/orchestrator/tests/test_create_slice_integration_branch.py +++ b/orchestrator/tests/test_create_slice_integration_branch.py @@ -726,6 +726,243 @@ def fake_make_request(endpoint, method=None, data=None, **kwargs): assert fetch_calls[0][0] == ("+refs/heads/egg/issue-1:refs/remotes/origin/egg/issue-1") +class TestIsSliceBranchMergedIntoParent: + """#2549 — detect whether a slice's PR has already merged into its + parent. This is the inverse of the #2512 restart-recovery check: + when ``existing_sha`` (slice tip on origin) is reachable from + ``parent_sha`` (parent tip on origin), the slice's commits are + already in the parent and any attempt to (re)create the slice's + integration branch via ``parent_sha:refs/heads/`` would be + rejected as non-fast-forward. + + The bootstrap reconciliation pass and the run-loop race-protection + check in ``routes/pipelines._run_implement_phase_slices`` both rely + on this signal — a False from here lets the slice run normally; a + True short-circuits the slice to COMPLETE. + """ + + def _setup_remotes(self, parent_sha: str | None, existing_sha: str | None): + def fake_get_remote_branch_sha(pipeline_id, repo_path, ref, **kwargs): + if ref.endswith("/slice-1"): + return existing_sha + return parent_sha + + return fake_get_remote_branch_sha + + def test_returns_true_when_slice_tip_is_ancestor_of_parent(self, gateway_client): + """The literal #2549 repro: slice-1 PR was merged into the + work branch; the slice-1 ref still exists on origin at its + pre-merge tip, and the work tip now has the merge commit on + top. ``existing_sha`` is reachable from ``parent_sha`` → + merged → True.""" + parent_sha = "f3c16e3b" * 5 # work tip after merge + existing_sha = "ea591ec1" * 5 # pre-merge slice-1 tip + + merge_base_calls: list[dict] = [] + + def fake_make_request(endpoint, method=None, data=None, **kwargs): + if endpoint == "/api/v1/git/execute": + merge_base_calls.append(dict(data or {})) + # existing IS reachable from parent → returncode 0 → True + return {"success": True, "data": {"returncode": 0}} + return {"success": True, "data": {}} + + with ( + patch.object(gateway_client, "register_session", return_value=_session_info()), + patch.object(gateway_client, "delete_session", return_value=True), + patch.object(gateway_client, "fetch_branch", return_value=True), + patch.object( + gateway_client, + "get_remote_branch_sha", + side_effect=self._setup_remotes(parent_sha, existing_sha), + ), + patch.object(gateway_client, "_make_request", side_effect=fake_make_request), + ): + merged = gateway_client.is_slice_branch_merged_into_parent( + "issue-2474-v2", + "/repo", + integration_branch="egg/issue-2474-v2/slice-1", + parent_branch="egg/issue-2474-v2/work", + ) + + assert merged is True + assert len(merge_base_calls) == 1 + mb = merge_base_calls[0] + assert mb["args"] == ["--is-ancestor", existing_sha, parent_sha], ( + "ancestry direction is the inverse of #2512: existing must be " + "ancestor of parent, signalling 'slice merged into parent'" + ) + + def test_returns_false_when_slice_tip_diverged_from_parent(self, gateway_client): + """Genuinely diverged history (slice has commits parent doesn't, + or vice versa) → not merged. Caller falls through to the regular + create path so origin's rejection (if any) surfaces normally.""" + parent_sha = "deadbeef" * 5 + existing_sha = "feedface" * 5 + + def fake_make_request(endpoint, method=None, data=None, **kwargs): + if endpoint == "/api/v1/git/execute": + # not-ancestor → returncode 1 + raise GatewayError( + "git merge-base failed", + status_code=500, + details={"returncode": 1, "stdout": "", "stderr": ""}, + ) + return {"success": True, "data": {}} + + with ( + patch.object(gateway_client, "register_session", return_value=_session_info()), + patch.object(gateway_client, "delete_session", return_value=True), + patch.object(gateway_client, "fetch_branch", return_value=True), + patch.object( + gateway_client, + "get_remote_branch_sha", + side_effect=self._setup_remotes(parent_sha, existing_sha), + ), + patch.object(gateway_client, "_make_request", side_effect=fake_make_request), + ): + merged = gateway_client.is_slice_branch_merged_into_parent( + "p", + "/repo", + integration_branch="egg/issue-1/slice-1", + parent_branch="egg/issue-1", + ) + + assert merged is False + + def test_returns_false_when_integration_branch_absent(self, gateway_client): + """First-run / branch-deleted case: ``ls-remote`` returns no + SHA for the integration branch → can't be merged → False. + Crucially does NOT run merge-base (no SHA to compare).""" + parent_sha = "abc12345" * 5 + + merge_base_calls: list[dict] = [] + + def fake_make_request(endpoint, method=None, data=None, **kwargs): + if endpoint == "/api/v1/git/execute": + merge_base_calls.append(dict(data or {})) + return {"success": True, "data": {}} + + with ( + patch.object(gateway_client, "register_session", return_value=_session_info()), + patch.object(gateway_client, "delete_session", return_value=True), + patch.object(gateway_client, "fetch_branch", return_value=True), + patch.object( + gateway_client, + "get_remote_branch_sha", + side_effect=self._setup_remotes(parent_sha, existing_sha=None), + ), + patch.object(gateway_client, "_make_request", side_effect=fake_make_request), + ): + merged = gateway_client.is_slice_branch_merged_into_parent( + "p", + "/repo", + integration_branch="egg/issue-1/slice-1", + parent_branch="egg/issue-1", + ) + + assert merged is False + assert merge_base_calls == [], ( + "must not run merge-base when one of the SHAs is unresolvable" + ) + + def test_returns_false_when_parent_branch_absent(self, gateway_client): + """If the parent branch can't be resolved on origin we have + nothing to compare against — return False rather than guess.""" + existing_sha = "feedface" * 5 + + with ( + patch.object(gateway_client, "register_session", return_value=_session_info()), + patch.object(gateway_client, "delete_session", return_value=True), + patch.object(gateway_client, "fetch_branch", return_value=True), + patch.object( + gateway_client, + "get_remote_branch_sha", + side_effect=self._setup_remotes(parent_sha=None, existing_sha=existing_sha), + ), + patch.object(gateway_client, "_make_request", return_value={"success": True}), + ): + merged = gateway_client.is_slice_branch_merged_into_parent( + "p", + "/repo", + integration_branch="egg/issue-1/slice-1", + parent_branch="egg/issue-1", + ) + + assert merged is False + + def test_returns_false_when_branches_equal(self, gateway_client): + """Tips equal → no-op state, neither merged nor diverged. Let + the caller fall through to the regular fast-forward no-op path.""" + sha = "cafebabe" * 5 + + merge_base_calls: list[dict] = [] + + def fake_make_request(endpoint, method=None, data=None, **kwargs): + if endpoint == "/api/v1/git/execute": + merge_base_calls.append(dict(data or {})) + return {"success": True, "data": {}} + + with ( + patch.object(gateway_client, "register_session", return_value=_session_info()), + patch.object(gateway_client, "delete_session", return_value=True), + patch.object(gateway_client, "fetch_branch", return_value=True), + patch.object(gateway_client, "get_remote_branch_sha", return_value=sha), + patch.object(gateway_client, "_make_request", side_effect=fake_make_request), + ): + merged = gateway_client.is_slice_branch_merged_into_parent( + "p", + "/repo", + integration_branch="egg/issue-1/slice-1", + parent_branch="egg/issue-1", + ) + + assert merged is False + assert merge_base_calls == [], "no merge-base when tips are equal" + + def test_session_cleaned_up_on_success_and_failure(self, gateway_client): + """The synthetic session must be deleted via ``delete_session`` + on both the success path and any exception path — symmetric + with ``create_slice_integration_branch``.""" + parent_sha = "deadbeef" * 5 + existing_sha = "feedface" * 5 + + delete_calls: list = [] + + def _delete(token): + delete_calls.append(token) + return True + + # Force an exception in merge-base so we exercise the failure path. + def fake_make_request(endpoint, method=None, data=None, **kwargs): + raise RuntimeError("kaboom") + + with ( + patch.object( + gateway_client, "register_session", return_value=_session_info("merged-tok") + ), + patch.object(gateway_client, "delete_session", side_effect=_delete), + patch.object(gateway_client, "fetch_branch", return_value=True), + patch.object( + gateway_client, + "get_remote_branch_sha", + side_effect=self._setup_remotes(parent_sha, existing_sha), + ), + patch.object(gateway_client, "_make_request", side_effect=fake_make_request), + ): + merged = gateway_client.is_slice_branch_merged_into_parent( + "p", + "/repo", + integration_branch="egg/issue-1/slice-1", + parent_branch="egg/issue-1", + ) + + assert merged is False + assert delete_calls == ["merged-tok"], ( + "synthetic session must be cleaned up even when the call raises" + ) + + class TestShaIsAncestor: """Unit tests for the ``_sha_is_ancestor`` helper that backs the #2512 restart-recovery detection.""" diff --git a/orchestrator/tests/test_slice_run_loop_integration.py b/orchestrator/tests/test_slice_run_loop_integration.py index 7c186f0fa1..2ac09c4d1e 100644 --- a/orchestrator/tests/test_slice_run_loop_integration.py +++ b/orchestrator/tests/test_slice_run_loop_integration.py @@ -439,6 +439,11 @@ def _make_spawner(self) -> MagicMock: spawner = MagicMock() spawner.gateway = MagicMock() spawner.gateway.create_slice_pr.return_value = "https://example/pr/1" + # #2549 — bootstrap reconciliation + run-loop race-protection both + # call this gateway helper. Default to False so existing tests + # exercise the spawn-and-run path; merged-detection tests set it + # to True explicitly. + spawner.gateway.is_slice_branch_merged_into_parent.return_value = False return spawner def _make_loader_save_pair(self, contract: Contract) -> tuple[MagicMock, MagicMock]: @@ -942,6 +947,342 @@ def test_single_slice_path_skips_pr_when_repo_unset(self) -> None: spawner.gateway.create_slice_pr.assert_not_called() +# --------------------------------------------------------------------------- +# #2549 — already-merged-slice detection (bootstrap + race protection) +# --------------------------------------------------------------------------- + + +class TestSliceMergedDetection: + """#2549 — orchestrator must skip slices whose PR has already merged. + + Live repro: pipeline ``issue-2474-v2`` slice-1 merged → operator + ran ``start_pipeline`` to resume from slice-2 → orchestrator tried + to recreate slice-1's integration branch → push rejected as + non-fast-forward → slice-1 cascade-failed slices 2-5 in 5 seconds. + + Two layers cover the failure: + + * **Bootstrap reconciliation** runs once before the slice run loop + starts. Folds in (A) ``Slice.status == COMPLETE`` from prior + run's contract write and (B) gateway-detected + ``is_slice_branch_merged_into_parent`` for slices the contract + doesn't know about yet (e.g. pipelines whose merge happened + before the writer landed). + + * **Run-loop race protection** runs at slice spawn. Catches the + narrow window where a slice's PR is merged after bootstrap but + before the slice's wave executes. + + Both layers persist ``slice.status = SliceStatus.COMPLETE`` on + the contract so subsequent restarts go through the cheap + contract-only path. + """ + + def _make_spawner(self) -> MagicMock: + spawner = MagicMock() + spawner.gateway = MagicMock() + spawner.gateway.create_slice_pr.return_value = "https://example/pr/1" + spawner.gateway.is_slice_branch_merged_into_parent.return_value = False + return spawner + + def test_bootstrap_skips_slice_marked_complete_on_contract(self) -> None: + """(A) — Slice already marked COMPLETE on the contract is + skipped without calling ``is_slice_branch_merged_into_parent`` + (cheap path: trust the contract, no GitHub round-trip).""" + pipeline = _make_pipeline() + slice1 = _make_slice("slice-1", tasks=[_make_task("task-1-1")]) + slice1.status = SliceStatus.COMPLETE # prior run wrote this on success + slice2 = _make_slice("slice-2", deps=["slice-1"], tasks=[_make_task("task-2-1")]) + contract = _make_contract(slices=[slice1, slice2]) + + with ( + patch("egg_contracts.loader.load_contract", return_value=contract), + patch("egg_contracts.loader.save_contract"), + patch("routes.pipelines._start_stacked_pr_reconciler") as mock_start_recon, + patch( + "routes.pipelines._run_concurrent_phase", return_value=(0, "ok") + ) as mock_run_phase, + patch("orchestrator.peer_consensus.remove_peer_consensus_tracker"), + ): + mock_start_recon.return_value = (MagicMock(), threading.Event()) + spawner = self._make_spawner() + exit_code, _ = _run_implement_phase_slices( + pipeline_id=pipeline.id, + pipeline=pipeline, + spawner=spawner, + repo_volumes={}, + gateway_mode="public", + repos=["owner/repo"], + sandbox_env={}, + store=MagicMock(), + certs_volume=None, + worktree_repo_path=Path("/tmp/x"), + ) + assert exit_code == 0 + # Only slice-2 ran — slice-1 was trusted from the contract. + invoked = {c.kwargs["slice_id"] for c in mock_run_phase.call_args_list} + assert invoked == {"slice-2"}, ( + "slice-1 must be skipped at bootstrap when its contract status is COMPLETE" + ) + # No PR opened for slice-1 (it's already done). + pr_slice_ids = [ + c.kwargs["slice_id"] for c in spawner.gateway.create_slice_pr.call_args_list + ] + assert "slice-1" not in pr_slice_ids + assert "slice-2" in pr_slice_ids + # Step (A) trusts the contract — no GitHub round-trip for the COMPLETE slice. + merged_calls_for_slice1 = [ + c + for c in spawner.gateway.is_slice_branch_merged_into_parent.call_args_list + if c.kwargs.get("integration_branch", "").endswith("/slice-1") + ] + assert merged_calls_for_slice1 == [], ( + "step (A) must skip the GitHub-side merged-detection when contract " + "already records COMPLETE" + ) + + def test_bootstrap_detects_merged_slice_on_origin(self) -> None: + """(B) — slice still PENDING on contract but merged on origin + (the literal #2549 repro). Bootstrap detects via + ``is_slice_branch_merged_into_parent``, marks the slice + complete, persists ``status=COMPLETE``, and the run loop + proceeds with slice-2 alone.""" + pipeline = _make_pipeline() + slice1 = _make_slice("slice-1", tasks=[_make_task("task-1-1")]) + slice2 = _make_slice("slice-2", deps=["slice-1"], tasks=[_make_task("task-2-1")]) + contract = _make_contract(slices=[slice1, slice2]) + + # The contract write under the lock loads + saves; mock the + # save to capture what status got persisted. + save_calls: list[Contract] = [] + + def _capture_save(c: Contract, _path: Any) -> None: + save_calls.append(c) + + # Slice-1 is the merged slice; slice-2 is not. + def _merged_side_effect(*_args: Any, **kwargs: Any) -> bool: + return kwargs.get("integration_branch", "").endswith("/slice-1") + + with ( + patch("egg_contracts.loader.load_contract", return_value=contract), + patch("egg_contracts.loader.save_contract", side_effect=_capture_save), + patch("routes.pipelines._start_stacked_pr_reconciler") as mock_start_recon, + patch( + "routes.pipelines._run_concurrent_phase", return_value=(0, "ok") + ) as mock_run_phase, + patch("orchestrator.peer_consensus.remove_peer_consensus_tracker"), + ): + mock_start_recon.return_value = (MagicMock(), threading.Event()) + spawner = self._make_spawner() + spawner.gateway.is_slice_branch_merged_into_parent.side_effect = _merged_side_effect + exit_code, _ = _run_implement_phase_slices( + pipeline_id=pipeline.id, + pipeline=pipeline, + spawner=spawner, + repo_volumes={}, + gateway_mode="public", + repos=["owner/repo"], + sandbox_env={}, + store=MagicMock(), + certs_volume=None, + worktree_repo_path=Path("/tmp/x"), + ) + assert exit_code == 0 + # slice-1 detected as merged → not run; slice-2 runs normally. + invoked = {c.kwargs["slice_id"] for c in mock_run_phase.call_args_list} + assert invoked == {"slice-2"}, ( + "slice-1 must be skipped at bootstrap when origin shows it merged" + ) + # No agent spawn or PR creation for slice-1. + pr_slice_ids = [ + c.kwargs["slice_id"] for c in spawner.gateway.create_slice_pr.call_args_list + ] + assert "slice-1" not in pr_slice_ids + # Status persisted to contract so future restarts hit the cheap path. + assert slice1.status == SliceStatus.COMPLETE, ( + "step (B) must persist slice.status=COMPLETE so subsequent restarts " + "skip the GitHub round-trip" + ) + + def test_bootstrap_does_nothing_when_pipeline_repo_unset(self) -> None: + """No ``pipeline.repo`` (e.g. local-only test pipeline) → step + (B) is skipped (no remote to query). Step (A) still applies + because it's a pure contract read — covered here by a + ``status=COMPLETE`` slice that must be skipped without any + gateway round-trip.""" + pipeline = _make_pipeline() + pipeline.repo = None + # slice-1 was completed on a prior run (step (A) — contract + # already records COMPLETE). slice-2 still has work to do and + # must run; step (B) cannot help here because there's no + # remote to query. + slice1 = _make_slice("slice-1", tasks=[_make_task("task-1-1")]) + slice1.status = SliceStatus.COMPLETE + slice2 = _make_slice("slice-2", deps=["slice-1"], tasks=[_make_task("task-2-1")]) + contract = _make_contract(slices=[slice1, slice2]) + + with ( + patch("egg_contracts.loader.load_contract", return_value=contract), + patch("egg_contracts.loader.save_contract"), + patch("routes.pipelines._start_stacked_pr_reconciler") as mock_start_recon, + patch( + "routes.pipelines._run_concurrent_phase", return_value=(0, "ok") + ) as mock_run_phase, + patch("orchestrator.peer_consensus.remove_peer_consensus_tracker"), + ): + mock_start_recon.return_value = (MagicMock(), threading.Event()) + spawner = self._make_spawner() + spawner.gateway.is_slice_branch_merged_into_parent.return_value = True + _run_implement_phase_slices( + pipeline_id=pipeline.id, + pipeline=pipeline, + spawner=spawner, + repo_volumes={}, + gateway_mode="public", + repos=["owner/repo"], + sandbox_env={}, + store=MagicMock(), + certs_volume=None, + worktree_repo_path=Path("/tmp/x"), + ) + # Step (B) skipped wholesale when pipeline.repo is None — we + # have no remote to query against. + spawner.gateway.is_slice_branch_merged_into_parent.assert_not_called() + # Step (A) still applies: slice-1 (already COMPLETE on the + # contract) is skipped; slice-2 runs normally. + invoked = {c.kwargs["slice_id"] for c in mock_run_phase.call_args_list} + assert invoked == {"slice-2"}, ( + "step (A) must trust the contract even when pipeline.repo is None" + ) + + def test_run_loop_race_skip_when_slice_merges_after_bootstrap(self) -> None: + """Race: bootstrap saw slice as PENDING (not merged); slice's + PR merges before the wave runs. ``_run_one_slice_inner`` must + re-check before push and skip cleanly — no agent spawn, no + slice-PR creation, no integration-branch push.""" + pipeline = _make_pipeline() + slice1 = _make_slice("slice-1", tasks=[_make_task("task-1-1")]) + contract = _make_contract(slices=[slice1]) + + # First call (bootstrap): not merged. Second call (race + # protection in _run_one_slice_inner): merged. + merged_call_count = {"n": 0} + + def _merged_side_effect(*_args: Any, **_kwargs: Any) -> bool: + merged_call_count["n"] += 1 + return merged_call_count["n"] >= 2 + + with ( + patch("egg_contracts.loader.load_contract", return_value=contract), + patch("egg_contracts.loader.save_contract"), + patch("routes.pipelines._start_stacked_pr_reconciler") as mock_start_recon, + patch( + "routes.pipelines._run_concurrent_phase", return_value=(0, "ok") + ) as mock_run_phase, + patch("orchestrator.peer_consensus.remove_peer_consensus_tracker"), + ): + mock_start_recon.return_value = (MagicMock(), threading.Event()) + spawner = self._make_spawner() + spawner.gateway.is_slice_branch_merged_into_parent.side_effect = _merged_side_effect + exit_code, logs = _run_implement_phase_slices( + pipeline_id=pipeline.id, + pipeline=pipeline, + spawner=spawner, + repo_volumes={}, + gateway_mode="public", + repos=["owner/repo"], + sandbox_env={}, + store=MagicMock(), + certs_volume=None, + worktree_repo_path=Path("/tmp/x"), + ) + assert exit_code == 0 + # No agent spawn — race-protection caught it after bootstrap missed it. + mock_run_phase.assert_not_called() + # No integration-branch push, no slice-PR creation. + spawner.gateway.create_slice_integration_branch.assert_not_called() + spawner.gateway.create_slice_pr.assert_not_called() + # Sanity: detection helper was actually called twice + # (bootstrap + race protection). + assert merged_call_count["n"] >= 2 + + def test_successful_slice_persists_status_complete_to_contract(self) -> None: + """Once a slice reaches PR-creation success, its + ``status=COMPLETE`` must land on the contract. Subsequent + restarts then skip via step (A) without a GitHub round-trip.""" + pipeline = _make_pipeline() + slice1 = _make_slice("slice-1", tasks=[_make_task("task-1-1")]) + contract = _make_contract(slices=[slice1]) + + with ( + patch("egg_contracts.loader.load_contract", return_value=contract), + patch("egg_contracts.loader.save_contract"), + patch("routes.pipelines._start_stacked_pr_reconciler") as mock_start_recon, + patch("routes.pipelines._run_concurrent_phase", return_value=(0, "ok")), + patch("orchestrator.peer_consensus.remove_peer_consensus_tracker"), + ): + mock_start_recon.return_value = (MagicMock(), threading.Event()) + spawner = self._make_spawner() + exit_code, _ = _run_implement_phase_slices( + pipeline_id=pipeline.id, + pipeline=pipeline, + spawner=spawner, + repo_volumes={}, + gateway_mode="public", + repos=["owner/repo"], + sandbox_env={}, + store=MagicMock(), + certs_volume=None, + worktree_repo_path=Path("/tmp/x"), + ) + assert exit_code == 0 + assert slice1.status == SliceStatus.COMPLETE, ( + "successful slice run must persist status=COMPLETE on the contract — " + "the durable signal that lets future restarts skip via step (A)" + ) + + def test_bootstrap_detection_failure_falls_through(self) -> None: + """``is_slice_branch_merged_into_parent`` raising must not + block the run loop — it's best-effort. The slice runs + through the regular path and the orchestrator tolerates the + gateway transient.""" + pipeline = _make_pipeline() + slice1 = _make_slice("slice-1", tasks=[_make_task("task-1-1")]) + contract = _make_contract(slices=[slice1]) + + with ( + patch("egg_contracts.loader.load_contract", return_value=contract), + patch("egg_contracts.loader.save_contract"), + patch("routes.pipelines._start_stacked_pr_reconciler") as mock_start_recon, + patch( + "routes.pipelines._run_concurrent_phase", return_value=(0, "ok") + ) as mock_run_phase, + patch("orchestrator.peer_consensus.remove_peer_consensus_tracker"), + ): + mock_start_recon.return_value = (MagicMock(), threading.Event()) + spawner = self._make_spawner() + spawner.gateway.is_slice_branch_merged_into_parent.side_effect = RuntimeError( + "gateway transient" + ) + exit_code, _ = _run_implement_phase_slices( + pipeline_id=pipeline.id, + pipeline=pipeline, + spawner=spawner, + repo_volumes={}, + gateway_mode="public", + repos=["owner/repo"], + sandbox_env={}, + store=MagicMock(), + certs_volume=None, + worktree_repo_path=Path("/tmp/x"), + ) + assert exit_code == 0 + # Detection raised → bootstrap and race-protection both treat + # as not-merged → slice runs the regular path. + assert mock_run_phase.call_count == 1 + spawner.gateway.create_slice_pr.assert_called_once() + + # --------------------------------------------------------------------------- # Coder fixes for reviewer_code_holistic v1 NACK (now regression guards) # --------------------------------------------------------------------------- @@ -997,6 +1338,7 @@ def _track_create_pr(*args: Any, **kwargs: Any) -> str: side_effect=_track_create_branch ) spawner.gateway.create_slice_pr = MagicMock(side_effect=_track_create_pr) + spawner.gateway.is_slice_branch_merged_into_parent = MagicMock(return_value=False) _run_implement_phase_slices( pipeline_id=pipeline.id, @@ -1284,6 +1626,11 @@ def _make_spawner(self) -> MagicMock: spawner = MagicMock() spawner.gateway = MagicMock() spawner.gateway.create_slice_pr.return_value = "https://example/pr/1" + # #2549 — bootstrap reconciliation + run-loop race-protection both + # call this gateway helper. Default to False so existing tests + # exercise the spawn-and-run path; merged-detection tests set it + # to True explicitly. + spawner.gateway.is_slice_branch_merged_into_parent.return_value = False return spawner def test_release_called_on_consensus_path(self) -> None: @@ -1539,6 +1886,7 @@ def _track_run_phase(*args: Any, **kwargs: Any) -> tuple[int, str]: side_effect=_track_create_branch ) spawner.gateway.create_slice_pr = MagicMock(return_value="https://example/pr/1") + spawner.gateway.is_slice_branch_merged_into_parent = MagicMock(return_value=False) _run_implement_phase_slices( pipeline_id=pipeline.id, @@ -1652,6 +2000,7 @@ def _capture(*args: Any, **kwargs: Any) -> bool: spawner.gateway = MagicMock() spawner.gateway.create_slice_integration_branch = MagicMock(side_effect=_capture) spawner.gateway.create_slice_pr = MagicMock(return_value="https://example/pr/1") + spawner.gateway.is_slice_branch_merged_into_parent = MagicMock(return_value=False) _run_implement_phase_slices( pipeline_id=pipeline.id,