Report actionable merge refusals and preserve recovery certainty

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={

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"

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

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

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,

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()