diff --git a/control_plane/http_app.py b/control_plane/http_app.py index 8f14e947a..0dd87e1bc 100644 --- a/control_plane/http_app.py +++ b/control_plane/http_app.py @@ -370,6 +370,7 @@ from control_plane.merge_train_github import ( GitHubMergeTrainClient, MergeTrainGitHubError, + MergeTrainGitHubMergeRejectedError, MergeTrainGitHubStaleHeadError, UrllibMergeTrainGitHubTransport, ) @@ -5748,6 +5749,23 @@ def merge_train_github_stale_state_response( def merge_train_github_request_failed_response( *, trace_id: str, error: MergeTrainGitHubError ) -> JSONResponse: + if isinstance(error, MergeTrainGitHubMergeRejectedError): + return JSONResponse( + status_code=409, + content={ + "status": "rejected", + "trace_id": trace_id, + "error": { + "code": "github_merge_rejected", + "message": "GitHub refused the guarded pull-request merge; inspect the refusal diagnosis before retrying.", + }, + "details": { + "github_status_code": error.status_code, + "pull_request_number": error.pull_request_number, + "refusal_diagnosis": error.refusal_diagnosis, + }, + }, + ) return JSONResponse( status_code=502, content={ diff --git a/control_plane/merge_train_controller_run_once.py b/control_plane/merge_train_controller_run_once.py index 45f64428f..25f46b61c 100644 --- a/control_plane/merge_train_controller_run_once.py +++ b/control_plane/merge_train_controller_run_once.py @@ -62,6 +62,7 @@ from control_plane.merge_train_github import ( GitHubMergeTrainClient, MergeTrainGitHubError, + MergeTrainGitHubMergeRejectedError, MergeTrainGitHubStaleHeadError, MergeTrainGitHubTransport, UrllibMergeTrainGitHubTransport, @@ -2833,6 +2834,12 @@ def _controller_result_reconciliation_detail( def _controller_exception_reconciliation_detail(error: Exception) -> str: + if isinstance(error, MergeTrainGitHubMergeRejectedError): + return ( + "operator_required:pull_request_head_behind_base" + if error.refusal_diagnosis == "head_behind_base" + else "operator_required:github_merge_rejected" + ) if isinstance(error, MergeTrainGitHubError): if error.status_code is None or error.status_code >= 500: return "retryable:github_request_failed" diff --git a/control_plane/merge_train_github.py b/control_plane/merge_train_github.py index 514ae65de..fd32463dd 100644 --- a/control_plane/merge_train_github.py +++ b/control_plane/merge_train_github.py @@ -64,6 +64,24 @@ class MergeTrainGitHubStaleHeadError(MergeTrainGitHubError): """Raised when GitHub state no longer matches guarded merge evidence.""" +class MergeTrainGitHubMergeRejectedError(MergeTrainGitHubError): + """A conclusive merge refusal with bounded, separately observed diagnosis.""" + + def __init__(self, *, pull_request_number: int, head_behind_base: bool) -> None: + self.pull_request_number = pull_request_number + self.refusal_diagnosis = "head_behind_base" if head_behind_base else "unconfirmed" + diagnosis = ( + "The same PR head is behind its base; refresh the branch and wait for fresh checks " + "before submitting it to the train again." + if head_behind_base + else "Reread the PR's merge requirements before another attempt." + ) + super().__init__( + f"GitHub refused to merge PR #{pull_request_number} (HTTP 405). {diagnosis}", + status_code=405, + ) + + HistoricalCompletionProofStatus = Literal["unsupported", "indeterminate"] HistoricalCompletionProofReason = Literal[ "plan_invalid", @@ -830,7 +848,7 @@ def update_progress( "Pull request head tree moved outside the batch landing plan.", status_code=409, ) - self._validate_open_landing_pull_request( + head_behind_base = self._validate_open_landing_pull_request( repository_path=repository_path, entry=entry, expected_base_ref=landing_plan.base_branch, @@ -846,6 +864,12 @@ def update_progress( observed_pull_request_state="open", observed_at=recorded_at, ) + if head_behind_base: + raise MergeAdmissionDeniedError( + f"PR #{entry.pull_request_number} is behind its base; refresh the branch and " + "wait for fresh checks before submitting it to the train again.", + reason_code="pull_request_head_behind_base", + ) if checkpoint is not None: checkpoint( _validated_model_update( @@ -1308,7 +1332,7 @@ def _validate_open_landing_pull_request( expected_base_ref: str, expected_base_sha: str, expected_base_tree_sha: str, - ) -> None: + ) -> bool: pull_request = _json_object( self.transport.request( method="GET", @@ -1354,6 +1378,7 @@ def _validate_open_landing_pull_request( raise MergeTrainGitHubStaleHeadError( "Target base branch moved outside the batch landing plan.", status_code=409 ) + return pull_request.get("mergeable_state") == "behind" def add_pull_request_label( self, *, repository: str, pull_request_number: int, label: str @@ -1389,14 +1414,36 @@ def merge_pull_request( merge_method: MergeTrainMergeMethod, ) -> str: repository_path = _repository_path(repository) - payload = self.transport.request( - method="PUT", - path=f"/repos/{repository_path}/pulls/{pull_request_number}/merge", - body={ - "sha": _required_value(head_sha, "Pull request head SHA is required."), - "merge_method": merge_method, - }, - ) + expected_head_sha = _required_value(head_sha, "Pull request head SHA is required.") + try: + payload = self.transport.request( + method="PUT", + path=f"/repos/{repository_path}/pulls/{pull_request_number}/merge", + body={"sha": expected_head_sha, "merge_method": merge_method}, + ) + except MergeTrainGitHubError as error: + if error.status_code != 405: + raise + head_behind_base = False + try: + observed = self.transport.request( + method="GET", + path=f"/repos/{repository_path}/pulls/{pull_request_number}", + ) + except Exception: # noqa: BLE001 - diagnosis cannot erase the confirmed merge refusal + observed = None + if isinstance(observed, dict): + head = observed.get("head") + head_behind_base = ( + observed.get("number") == pull_request_number + and observed.get("state") == "open" + and isinstance(head, dict) + and head.get("sha") == expected_head_sha + and observed.get("mergeable_state") == "behind" + ) + raise MergeTrainGitHubMergeRejectedError( + pull_request_number=pull_request_number, head_behind_base=head_behind_base + ) from error if not isinstance(payload, dict): raise MergeTrainGitHubError( "GitHub merge response must be a JSON object.", status_code=None diff --git a/docs/merge-admission.md b/docs/merge-admission.md index b2b36e836..048f41013 100644 --- a/docs/merge-admission.md +++ b/docs/merge-admission.md @@ -72,6 +72,23 @@ outcome is also effect-unknown. Neither state permits another provider attempt. Reconciliation observes GitHub first and appends a successor outcome; it never rewrites history or repeats an ambiguous mutation. +A GitHub merge-endpoint HTTP 405 remains a conclusive rejected outcome. The +adapter makes one read of the same PR to diagnose whether its unchanged, open +head is now behind its base. The operator response retains the attempt's trace, +PR number, and provider status, with a branch-refresh instruction only when that +read proves the condition. A failed, malformed, closed, or changed-head read +leaves the diagnosis unconfirmed and asks the operator to reread merge +requirements. Raw provider response bodies are not copied into this diagnosis. +The refusal returns HTTP 409 `github_merge_rejected`, not an upstream-outage +retry instruction; no second merge is attempted by the diagnostic read. +Before each subsequent merge, the existing PR read also blocks an observed +`behind` head before new admission or provider mutation, after reconciling an +earlier unresolved attempt from that same unchanged open-head/base proof. This includes resumed +partial batches: the first landed entry stays recorded, and the behind entry +does not accumulate repeated admissions and rejected merge calls while GitHub +continues to report that condition. Unknown mergeability alone does not prove +the refusal's cause. + The `already_contained_no_provider_effect` landed reason is a successful no-op, not evidence of a merge request. It records `provider_effect_attempted=false`, the observed PR lifecycle and exact unchanged base/head identities, with no diff --git a/tests/test_http_app_merge_train.py b/tests/test_http_app_merge_train.py index 6d5df4b01..494b89498 100644 --- a/tests/test_http_app_merge_train.py +++ b/tests/test_http_app_merge_train.py @@ -29,6 +29,7 @@ ) from control_plane.merge_train import MergeTrainDryRunSnapshot from control_plane.merge_train_controller_run_once import MERGE_TRAIN_CONTROLLER_ACTIVE_ACTION +from control_plane.merge_train_github import MergeTrainGitHubMergeRejectedError from control_plane.service_auth import ( BearerIdentityConfig, LaunchplaneAuthzPolicy, @@ -2259,6 +2260,97 @@ def land_batch_candidate(self, **kwargs: Any) -> Any: self.assertEqual(stale_record.landing_plan.entries[0].merge_commit_sha, "d" * 40) self.assertTrue(stale_record.source.startswith("service:controller:stale-landing:")) + async def test_later_merge_refusal_exposes_diagnosis_and_preserves_first_landing(self) -> None: + class PartialLandingThenRefused(_FakeMergeTrainGitHubClient): + def land_batch_candidate(self, **kwargs: Any) -> Any: + plan = kwargs["landing_plan"] + first, second = plan.entries + merged = first.model_copy( + update={ + "status": "merged", + "merge_commit_sha": "d" * 40, + "merge_commit_tree_sha": "e" * 40, + } + ) + progress = plan.model_copy(update={"entries": (merged, second)}) + record = kwargs["checkpoint"](progress, merged, "entry_merged") + kwargs["admission_guard"].update_landing_plan_record(record) + kwargs["provider_checkpoint"](progress, second) + raise MergeTrainGitHubMergeRejectedError( + pull_request_number=second.pull_request_number, head_behind_base=True + ) + + with ( + TemporaryDirectory() as temporary_directory_name, + patch.dict("os.environ", {"GH_TOKEN": "token"}, clear=True), + ): + state_dir = Path(temporary_directory_name) / "state" + _seed_merge_train_policy(state_dir) + store = FilesystemRecordStore(state_dir=state_dir) + app = create_launchplane_fastapi_app( + verifier=_StubVerifier(_merge_train_service_identity()), + authz_policy=_merge_train_service_policy(), + record_store_factory=lambda: store, + ) + request_payload = { + "schema_version": 1, + "repository": "cbusillo/sellyouroutboard", + "base_branch": "main", + "mutate": True, + } + with ( + patch( + "control_plane.merge_train_github.GitHubMergeTrainSnapshotReader", + _FakeExpandedMergeTrainSnapshotReader, + ), + patch( + "control_plane.merge_train_controller_run_once.GitHubMergeTrainClient", + _FakeMergeTrainGitHubClient, + ), + ): + for _ in range(4): + await _post_merge_train_controller_run_once(app, request_payload) + with patch( + "control_plane.merge_train_controller_run_once.GitHubMergeTrainClient", + PartialLandingThenRefused, + ): + response = await _post_merge_train_controller_run_once(app, request_payload) + state = store.list_merge_train_controller_state_records( + repository="cbusillo/sellyouroutboard", base_branch="main", limit=1 + )[0] + progress_record = next( + record + for record in store.list_merge_train_batch_landing_plan_records( + repository="cbusillo/sellyouroutboard", base_branch="main" + ) + if record.source + == f"service:controller:landing-progress:{response.json()['trace_id']}" + ) + + self.assertEqual(response.status_code, 409) + payload = response.json() + self.assertEqual(payload["error"]["code"], "github_merge_rejected") + self.assertEqual( + payload["details"], + { + "github_status_code": 405, + "pull_request_number": 2, + "refusal_diagnosis": "head_behind_base", + }, + ) + self.assertIn("inspect the refusal diagnosis", payload["error"]["message"]) + self.assertIn(payload["trace_id"], progress_record.source) + self.assertEqual( + [ + (entry.status, entry.merge_commit_sha) + for entry in progress_record.landing_plan.entries + ], + [("merged", "d" * 40), ("planned", "")], + ) + self.assertEqual( + state.reconciliation_detail, "operator_required:pull_request_head_behind_base" + ) + async def test_admission_block_recovers_stuck_pre_provider_reconciliation(self) -> None: with ( TemporaryDirectory() as temporary_directory_name, diff --git a/tests/test_merge_admission_records.py b/tests/test_merge_admission_records.py index 75ed0fb7d..c2495a171 100644 --- a/tests/test_merge_admission_records.py +++ b/tests/test_merge_admission_records.py @@ -41,6 +41,7 @@ MergeAdmissionReconciliationRequiredError, ) from control_plane.merge_train_admission import build_merge_train_controller_status_read_model +from control_plane.merge_train_github import MergeTrainGitHubMergeRejectedError from control_plane.storage.filesystem import FilesystemRecordStore from control_plane.storage.postgres import PostgresRecordStore from tests.test_merge_readiness import ( @@ -938,6 +939,29 @@ def test_scenario_23_rejection_requires_fresh_admission_for_retry(self) -> None: self.assertNotEqual(first.admission_id, second.admission_id) self.assertEqual(second.attempt_sequence, 2) + def test_diagnosed_refusal_preserves_provider_rejection_and_attempt_trace(self) -> None: + guard = self._guard() + admission = self._admit(guard) + error = MergeTrainGitHubMergeRejectedError( + pull_request_number=admission.pull_request_number, head_behind_base=True + ) + + outcome = guard.record_provider_failure( + admission=admission, error=error, observed_at="2026-08-11T03:02:00Z" + ) + stored = self.store.list_merge_landing_outcome_records(admission_id=admission.admission_id)[ + 0 + ] + + self.assertEqual(stored, outcome) + self.assertEqual(stored.status, "rejected") + self.assertEqual(stored.provider_status_code, 405) + self.assertTrue(stored.provider_effect_attempted) + self.assertTrue(stored.provider_conclusive_rejection) + self.assertTrue(admission.source.endswith(guard.trace_id)) + self.assertTrue(stored.source.endswith(guard.trace_id)) + self.assertIn("behind its base", stored.provider_message) + if __name__ == "__main__": unittest.main() diff --git a/tests/test_merge_train_github.py b/tests/test_merge_train_github.py index 0a8098ecc..e5eb600b9 100644 --- a/tests/test_merge_train_github.py +++ b/tests/test_merge_train_github.py @@ -1,6 +1,8 @@ import unittest from types import SimpleNamespace +from typing import cast from email.message import Message +from http.client import IncompleteRead from unittest.mock import patch from urllib.error import HTTPError @@ -32,9 +34,11 @@ from control_plane.merge_train_github import GitHubMergeTrainClient from control_plane.merge_train_github import GitHubMergeTrainSnapshotReader from control_plane.merge_train_github import MergeTrainGitHubError +from control_plane.merge_train_github import MergeTrainGitHubMergeRejectedError from control_plane.merge_train_github import MergeTrainGitHubStaleHeadError from control_plane.merge_train_github import RecordingMergeTrainGitHubTransport from control_plane.merge_train_github import UrllibMergeTrainGitHubTransport +from control_plane.merge_admission import GuardedMergeAdmission, MergeAdmissionDeniedError from control_plane.merge_train_structural_provenance import ( evaluate_merge_train_structural_candidate, ) @@ -98,6 +102,7 @@ def __init__(self) -> None: self.admit_calls: list[dict[str, object]] = [] self.landed_calls: list[dict[str, object]] = [] self.reconcile_required_calls: list[dict[str, object]] = [] + self.no_effect_reconciliations = 0 def admit(self, **kwargs: object) -> object: self.admit_calls.append(kwargs) @@ -116,7 +121,7 @@ def reconcile_existing_landed(self, **_: object) -> None: return None def reconcile_existing_no_effect(self, **_: object) -> None: - return None + self.no_effect_reconciliations += 1 def update_landing_plan(self, _: MergeTrainBatchLandingPlan) -> None: return None @@ -218,6 +223,87 @@ def test_merge_pull_request_requires_merge_response_sha(self) -> None: merge_method="merge", ) + def test_merge_refusal_diagnoses_only_the_same_open_head_without_retry(self) -> None: + observed = { + "number": 42, + "state": "open", + "head": {"sha": "head-42"}, + "mergeable_state": "behind", + } + for changes, diagnosis in ( + ({}, "head_behind_base"), + ({"head": {"sha": "new-head"}}, "unconfirmed"), + ({"state": "closed"}, "unconfirmed"), + ({"number": 43}, "unconfirmed"), + ({"mergeable_state": "blocked"}, "unconfirmed"), + ): + with self.subTest(changes=changes): + transport = RecordingMergeTrainGitHubTransport( + responses=( + MergeTrainGitHubError("private provider detail", status_code=405), + observed | changes, + ) + ) + client = GitHubMergeTrainClient(transport=transport) + + with self.assertRaises(MergeTrainGitHubMergeRejectedError) as caught: + client.merge_pull_request( + repository="example/train-repo", + pull_request_number=42, + head_sha="head-42", + merge_method="merge", + ) + + self.assertEqual(caught.exception.status_code, 405) + self.assertEqual(caught.exception.refusal_diagnosis, diagnosis) + self.assertNotIn("private provider detail", str(caught.exception)) + self.assertEqual( + [(request.method, request.path) for request in transport.requests], + [ + ("PUT", "/repos/example/train-repo/pulls/42/merge"), + ("GET", "/repos/example/train-repo/pulls/42"), + ], + ) + + def test_merge_refusal_read_failure_does_not_erase_conclusive_rejection(self) -> None: + observations: tuple[object, ...] = ( + None, + [], + {"head": []}, + MergeTrainGitHubError("read failed", status_code=503), + IncompleteRead(b"private response fragment"), + ) + for observation in observations: + with self.subTest(observation=observation): + transport = RecordingMergeTrainGitHubTransport( + responses=(MergeTrainGitHubError("refused", status_code=405), observation) + ) + with self.assertRaises(MergeTrainGitHubMergeRejectedError) as caught: + GitHubMergeTrainClient(transport=transport).merge_pull_request( + repository="example/train-repo", + pull_request_number=42, + head_sha="head-42", + merge_method="merge", + ) + + self.assertEqual(caught.exception.status_code, 405) + self.assertEqual(caught.exception.refusal_diagnosis, "unconfirmed") + self.assertEqual(len(transport.requests), 2) + + def test_merge_transport_ambiguity_is_not_diagnosed_as_a_refusal(self) -> None: + error = MergeTrainGitHubError("unavailable", status_code=503) + transport = RecordingMergeTrainGitHubTransport(responses=(error,)) + with self.assertRaises(MergeTrainGitHubError) as caught: + GitHubMergeTrainClient(transport=transport).merge_pull_request( + repository="example/train-repo", + pull_request_number=42, + head_sha="head-42", + merge_method="merge", + ) + + self.assertIs(caught.exception, error) + self.assertEqual(len(transport.requests), 1) + def test_comment_pull_request_posts_issue_comment(self) -> None: transport = RecordingMergeTrainGitHubTransport( responses=({"html_url": "https://github.com/example/repo/pull/11#issuecomment-1"},) @@ -1456,6 +1542,50 @@ def test_lagging_projection_confirmation_rejects_foreign_identity_before_second_ self.assertEqual(progress[-1].entries[0].status, "merged") self.assertEqual(progress[-1].entries[0].merge_commit_sha, "merge-sha-1") + def test_later_refusal_is_not_retried_when_resumed_head_is_behind(self) -> None: + progress: list[MergeTrainBatchLandingPlan] = [] + guard = _PermissiveMergeAdmissionGuard() + behind = _landing_pull_request(2, base_sha="merge-sha-1") | {"mergeable_state": "behind"} + transport = RecordingMergeTrainGitHubTransport( + responses=( + _github_branch(sha="base-main"), + *_normal_landing_responses(1, "base-main", "merge-sha-1"), + _github_branch(sha="merge-sha-1"), + _git_commit("head-2", "tree-head-2"), + _landing_pull_request(2, base_sha="merge-sha-1") | {"mergeable_state": "unknown"}, + MergeTrainGitHubError("provider detail", status_code=405), + behind, + ) + ) + with self.assertRaises(MergeTrainGitHubMergeRejectedError) as refused: + GitHubMergeTrainClient(transport=transport).land_batch_candidate( + landing_plan=_landing_plan(), + admission_guard=cast(GuardedMergeAdmission, guard), + checkpoint=lambda plan, _entry, _phase: progress.append(plan), + ) + self.assertEqual(refused.exception.pull_request_number, 2) + self.assertEqual(refused.exception.refusal_diagnosis, "head_behind_base") + self.assertEqual(len(guard.admit_calls), 2) + self.assertEqual(progress[-1].entries[0].merge_commit_sha, "merge-sha-1") + + resumed = RecordingMergeTrainGitHubTransport( + responses=( + _github_branch(sha="merge-sha-1"), + *_already_merged_responses(1, "base-main", "merge-sha-1", "identical"), + _github_branch(sha="merge-sha-1"), + _git_commit("head-2", "tree-head-2"), + behind, + ) + ) + with self.assertRaises(MergeAdmissionDeniedError) as blocked: + GitHubMergeTrainClient(transport=resumed).land_batch_candidate( + landing_plan=progress[-1], admission_guard=cast(GuardedMergeAdmission, guard) + ) + self.assertEqual(blocked.exception.reason_code, "pull_request_head_behind_base") + self.assertEqual(len(guard.admit_calls), 2) + self.assertEqual(guard.no_effect_reconciliations, 3) + self.assertTrue(all(request.method == "GET" for request in resumed.requests)) + def test_land_batch_candidate_merges_original_prs_in_order(self) -> None: landing_plan = _landing_plan() checkpoints: list[tuple[str, int, int]] = []