Merge pull request 'fix(mcp): derive checks requirement from live branch protection (Closes #751)' (#752) from fix/issue-751-checks-assessor into master

`gitea_assess_pr_sync_status` could permanently block a merge-ready PR whose
head commit had no status contexts.

Gitea's combined commit-status endpoint reports `state: pending` both when a
check is executing and when the status-context collection is empty. The
wrapper took that `state` verbatim, and `checks_required` defaulted to True
with no production path ever setting it, so "no CI configured" was
indistinguishable from "CI is running" and waiting could never resolve it.

Root cause spans the whole dataflow, not one call site:

* the wrapper read only the combined `state` and never counted contexts;
  its `checks_status="none"` fallback was unreachable for any truthy state
* `pr_sync_status.assess_pr_sync_status` defaulted `checks_required=True`
* the MCP wrapper exposed no derived `checks_required` and never passed one
* the branch-protection reader fetched the payload carrying
  `enable_status_check` / `status_check_contexts` and discarded both

Changes:

* `pr_sync_status.py`: add `classify_commit_checks` plus explicit
  CHECKS_* classifications (success / failure / pending / none /
  not_required / missing_required / unknown). The combined state is
  recorded for observability but is never evidence that CI is executing.
  Aggregation is fail-closed and newest-wins per context.
* `pr_sync_status.py`: rewrite the merge_now checks gate to consume the
  derived `checks_required`, distinguish the new classifications with
  precise blocker reasons, and fail closed on unrecognized vocabulary.
  The previous gate let any value outside its fixed vocabulary fall
  through to merge_now.
* `gitea_mcp_server.py`: add `_branch_protection_policy` deriving both the
  current-base rule and the status-check requirement from one live read;
  `_branch_protection_requires_current_base` is retained as a thin
  accessor with unchanged semantics. Add `_commit_checks_snapshot` which
  returns the context collection alongside the combined state.
* `gitea_mcp_server.py`: wire the derived `checks_required` into the
  production assessment path and report `checks_evidence`.

`checks_required` is always derived from live evidence and is deliberately
not a caller-supplied input, so no session can declare checks optional
without proof. Live classification also outranks a caller-supplied
`checks_status`, which can no longer mask a real required-check failure.
An unreadable protection policy or status collection stays fail-closed.

Approval, current-head, current-base, mergeability, conflict, role, lease
and merge-authorization gates are unchanged.

Tests: `tests/test_issue_751_checks_assessor.py` (40 tests).

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>

gitea_mcp_server.py

@@ -15367,46 +15367,151 @@ def _count_commits_behind(
     return None
 
 
+def _matching_protection_rule(protections: Any, branch: str) -> dict | None:
+    """Select the branch-protection rule governing ``branch`` (#751)."""
+    if not isinstance(protections, list):
+        return None
+    for rule in protections:
+        if not isinstance(rule, dict):
+            continue
+        name = (rule.get("branch_name") or rule.get("rule_name") or "").strip()
+        # Exact match or glob-ish contains for common patterns.
+        if name == branch or name in ("*", f"{branch}"):
+            return rule
+    # Fallback: any protection that mentions the base branch.
+    for rule in protections:
+        if not isinstance(rule, dict):
+            continue
+        name = (rule.get("branch_name") or rule.get("rule_name") or "").strip()
+        if name and (branch in name or name.endswith(branch)):
+            return rule
+    return None
+
+
+def _branch_protection_policy(
+    base_url: str,
+    auth: dict,
+    *,
+    base_branch: str,
+) -> dict:
+    """Read the live branch-protection policy for ``base_branch`` (#751).
+
+    Exposes both the current-base rule (``block_on_outdated_branch``) and the
+    status-check requirement (``enable_status_check`` / ``status_check_contexts``)
+    from the same payload, so the checks assessor can tell "no checks required"
+    apart from "required checks pending".
+
+    ``determinable`` is False only when the policy genuinely could not be read
+    (missing branch or API failure) — never merely because no rule exists. A
+    successful read that finds no protection for the branch is authoritative
+    evidence that status checks are not required.
+    """
+    policy: dict[str, Any] = {
+        "determinable": False,
+        "protection_found": False,
+        "requires_current_base": None,
+        "checks_enabled": None,
+        "required_contexts": [],
+        "base_branch": (base_branch or "").strip() or None,
+    }
+    branch = (base_branch or "").strip()
+    if not branch:
+        return policy
+
+    def _apply(rule: dict) -> None:
+        policy["protection_found"] = True
+        if "block_on_outdated_branch" in rule:
+            policy["requires_current_base"] = bool(
+                rule.get("block_on_outdated_branch")
+            )
+        if "enable_status_check" in rule:
+            policy["checks_enabled"] = bool(rule.get("enable_status_check"))
+        contexts = rule.get("status_check_contexts")
+        if isinstance(contexts, list):
+            policy["required_contexts"] = [
+                str(ctx).strip() for ctx in contexts if str(ctx or "").strip()
+            ]
+
+    try:
+        protections = api_request(
+            "GET", f"{base_url}/branch_protections", auth
+        )
+        rule = _matching_protection_rule(protections, branch)
+        if rule is not None:
+            _apply(rule)
+        if not policy["protection_found"]:
+            # Branch payload may embed effective protection.
+            br = api_request("GET", f"{base_url}/branches/{branch}", auth) or {}
+            prot = br.get("protection") or br.get("effective_branch_protection") or {}
+            if isinstance(prot, dict) and prot:
+                _apply(prot)
+        policy["determinable"] = True
+    except Exception:
+        # Genuine read failure — leave determinable False so callers fail closed.
+        return policy
+
+    if not policy["protection_found"]:
+        # Authoritative absence: no protection governs the branch, so no status
+        # check is required by policy.
+        policy["checks_enabled"] = False
+    elif policy["checks_enabled"] is None:
+        # Protection exists but omits the status-check field entirely: Gitea
+        # only enforces contexts when the flag is set, so absence means off.
+        policy["checks_enabled"] = False
+
+    return policy
+
+
 def _branch_protection_requires_current_base(
     base_url: str,
     auth: dict,
     *,
     base_branch: str,
 ) -> bool | None:
-    """Read Gitea branch protection ``block_on_outdated_branch`` when present."""
-    branch = (base_branch or "").strip()
-    if not branch:
-        return None
+    """Read Gitea branch protection ``block_on_outdated_branch`` when present.
+
+    Thin accessor over :func:`_branch_protection_policy`; return semantics are
+    unchanged (``None`` when the rule is absent or unreadable).
+    """
+    policy = _branch_protection_policy(base_url, auth, base_branch=base_branch)
+    return policy.get("requires_current_base")
+
+
+def _commit_checks_snapshot(
+    base_url: str,
+    auth: dict,
+    *,
+    sha: str,
+) -> dict:
+    """Read the combined commit status *and its context collection* (#751).
+
+    The combined ``state`` alone is not decisive: Gitea reports ``pending`` for
+    a commit with an empty status-context collection, which is indistinguishable
+    from executing CI unless the collection itself is inspected.
+    """
+    snapshot: dict[str, Any] = {
+        "determinable": False,
+        "combined_state": None,
+        "statuses": [],
+    }
+    head = (sha or "").strip()
+    if not head:
+        return snapshot
     try:
-        # Prefer the named protection rule matching the base branch.
-        protections = api_request(
-            "GET", f"{base_url}/branch_protections", auth
-        ) or []
-        if isinstance(protections, list):
-            for rule in protections:
-                if not isinstance(rule, dict):
-                    continue
-                name = (rule.get("branch_name") or rule.get("rule_name") or "").strip()
-                # Exact match or glob-ish contains for common patterns.
-                if name == branch or name in ("*", f"{branch}"):
-                    if "block_on_outdated_branch" in rule:
-                        return bool(rule.get("block_on_outdated_branch"))
-            # Fallback: any protection that mentions the base branch.
-            for rule in protections:
-                if not isinstance(rule, dict):
-                    continue
-                name = (rule.get("branch_name") or rule.get("rule_name") or "").strip()
-                if branch in name or name.endswith(branch):
-                    if "block_on_outdated_branch" in rule:
-                        return bool(rule.get("block_on_outdated_branch"))
-        # Branch payload may embed effective protection.
-        br = api_request("GET", f"{base_url}/branches/{branch}", auth) or {}
-        prot = br.get("protection") or br.get("effective_branch_protection") or {}
-        if isinstance(prot, dict) and "block_on_outdated_branch" in prot:
-            return bool(prot.get("block_on_outdated_branch"))
+        payload = api_request(
+            "GET", f"{base_url}/commits/{head}/status", auth
+        )
     except Exception:
-        return None
-    return None
+        return snapshot
+    if payload is None:
+        return snapshot
+    snapshot["determinable"] = True
+    if not isinstance(payload, dict):
+        return snapshot
+    snapshot["combined_state"] = (payload.get("state") or "").strip().lower() or None
+    statuses = payload.get("statuses")
+    snapshot["statuses"] = statuses if isinstance(statuses, list) else []
+    return snapshot
 
 
 def _prove_author_ownership_for_pr(
@@ -15651,11 +15756,14 @@ def gitea_assess_pr_sync_status(
             # Fail closed on unknown behind-count only when SHAs differ.
             commits_behind = 0 if pr_head_sha == base_head_sha else None
 
+    # Single live read of the base-branch protection policy; it carries both
+    # the current-base rule and the status-check requirement (#751).
+    protection_policy = _branch_protection_policy(
+        base, auth, base_branch=base_branch
+    )
     if branch_protection_requires_current_base is None:
-        branch_protection_requires_current_base = (
-            _branch_protection_requires_current_base(
-                base, auth, base_branch=base_branch
-            )
+        branch_protection_requires_current_base = protection_policy.get(
+            "requires_current_base"
         )
     # When protection cannot be read, fail closed by requiring current base
     # whenever the PR is behind (safer for protected repos). Callers may pass
@@ -15722,23 +15830,30 @@ def gitea_assess_pr_sync_status(
     except Exception:
         pass
 
+    # ── Live checks derivation (#751) ────────────────────────────────────
+    # Classify from the actual status-context collection plus the live
+    # protection policy. The combined ``state`` is never treated as proof that
+    # CI is executing, because Gitea reports ``pending`` for an empty
+    # collection. ``checks_required`` is always derived from live evidence and
+    # is deliberately not a caller-supplied input, so no session can declare
+    # checks optional without proof.
+    checks_snapshot = _commit_checks_snapshot(base, auth, sha=pr_head_sha or "")
+    checks_classification = pr_sync_status.classify_commit_checks(
+        combined_state=checks_snapshot.get("combined_state"),
+        statuses=checks_snapshot.get("statuses"),
+        checks_enabled=protection_policy.get("checks_enabled"),
+        required_contexts=protection_policy.get("required_contexts"),
+        policy_determinable=bool(protection_policy.get("determinable")),
+        status_determinable=bool(checks_snapshot.get("determinable")),
+    )
+    checks_required = bool(checks_classification.get("checks_required", True))
+    caller_supplied_checks_status = checks_status
     if checks_status is None:
-        checks_status = "unknown"
-        # Combined status on PR head when Actions/status API is available.
-        if pr_head_sha:
-            try:
-                st = api_request(
-                    "GET",
-                    f"{base}/commits/{pr_head_sha}/status",
-                    auth,
-                ) or {}
-                state = (st.get("state") or "").strip().lower()
-                if state:
-                    checks_status = state
-                elif not st:
-                    checks_status = "none"
-            except Exception:
-                checks_status = "unknown"
+        checks_status = checks_classification.get("checks_status")
+    elif checks_classification.get("checks_status") != pr_sync_status.CHECKS_UNKNOWN:
+        # Live evidence outranks a caller-supplied value; the override only
+        # applies when live status could not be classified at all.
+        checks_status = checks_classification.get("checks_status")
 
     assessment = pr_sync_status.assess_pr_sync_status(
         host=h,
@@ -15759,9 +15874,26 @@ def gitea_assess_pr_sync_status(
         active_reviewer_lease=active_reviewer_lease,
         active_merger_lease=active_merger_lease,
         prepared_verdict_head_sha=prepared_verdict_head_sha,
+        checks_required=checks_required,
     )
     assessment["remote"] = remote if remote in REMOTES else None
     assessment["base_branch"] = base_branch
+    # Evidence for the checks decision (#751) — no secrets, read-only.
+    assessment["checks_evidence"] = {
+        "combined_state": checks_classification.get("combined_state"),
+        "context_count": checks_classification.get("context_count"),
+        "observed_contexts": checks_classification.get("observed_contexts"),
+        "required_contexts": checks_classification.get("required_contexts"),
+        "missing_required_contexts": checks_classification.get(
+            "missing_required_contexts"
+        ),
+        "policy_determinable": checks_classification.get("policy_determinable"),
+        "status_determinable": checks_classification.get("status_determinable"),
+        "protection_found": protection_policy.get("protection_found"),
+        "checks_enabled": protection_policy.get("checks_enabled"),
+        "caller_supplied_checks_status": caller_supplied_checks_status,
+        "reasons": checks_classification.get("reasons"),
+    }
     assessment["success"] = True
     assessment["performed"] = False
     return assessment

pr_sync_status.py

@@ -41,6 +41,186 @@ _DENIED_UPDATE_ROLES = frozenset({"reviewer", "merger", "reconciler", "mixed", "
 UPDATE_STYLE_MERGE = "merge"
 _FORBIDDEN_UPDATE_STYLES = frozenset({"rebase", "rebase-merge", "squash", "force"})
 
+# ── Commit check classifications (#751) ──────────────────────────────────
+# Gitea's *combined* commit status reports ``state: pending`` both when a real
+# check is executing and when the status-context collection is empty. Reading
+# ``state`` alone therefore cannot distinguish "CI is running" from "no CI
+# exists", which permanently blocks a merge-ready PR that no check will ever
+# report on. These classifications are derived from the actual context
+# collection plus the live branch-protection policy.
+CHECKS_SUCCESS = "success"
+CHECKS_FAILURE = "failure"
+CHECKS_PENDING = "pending"
+CHECKS_NONE = "none"                          # configured/produced nothing
+CHECKS_NOT_REQUIRED = "not_required"          # protection does not require checks
+CHECKS_MISSING_REQUIRED = "missing_required"  # required contexts have no result
+CHECKS_UNKNOWN = "unknown"                    # indeterminable — fail closed
+
+# Values that permit merge_now when checks are required.
+_CHECKS_OK = frozenset({"success", "passed", "ok", "skipped", "not_required"})
+
+# Raw per-context state vocabularies reported by Gitea.
+_CTX_SUCCESS = frozenset({"success", "passed", "ok"})
+_CTX_FAILURE = frozenset({"failure", "failed", "error", "cancelled", "canceled"})
+_CTX_PENDING = frozenset({"pending", "running", "queued", "expected"})
+_CTX_SKIPPED = frozenset({"skipped", "neutral"})
+
+
+def _normalize_context_rows(statuses: Any) -> list[dict[str, str]]:
+    """Reduce a raw status collection to newest-wins ``{context, state}`` rows.
+
+    Gitea returns the status collection newest-first, so the first row seen for
+    a context wins. Rows without a usable state are discarded rather than being
+    silently treated as passing.
+    """
+    rows: list[dict[str, str]] = []
+    seen: set[str] = set()
+    if not isinstance(statuses, list):
+        return rows
+    for raw in statuses:
+        if not isinstance(raw, dict):
+            continue
+        context = (raw.get("context") or raw.get("name") or "").strip()
+        state = (raw.get("status") or raw.get("state") or "").strip().lower()
+        if not state:
+            continue
+        key = context or f"__unnamed__{len(rows)}"
+        if key in seen:
+            continue
+        seen.add(key)
+        rows.append({"context": context, "state": state})
+    return rows
+
+
+def _aggregate_context_states(rows: list[dict[str, str]]) -> str:
+    """Fail-closed aggregate: failure > pending > unknown-state > success."""
+    states = {row["state"] for row in rows}
+    if states & _CTX_FAILURE:
+        return CHECKS_FAILURE
+    if states & _CTX_PENDING:
+        return CHECKS_PENDING
+    unresolved = states - _CTX_SUCCESS - _CTX_SKIPPED
+    if unresolved:
+        # An unrecognized context state must never read as success.
+        return CHECKS_UNKNOWN
+    return CHECKS_SUCCESS
+
+
+def classify_commit_checks(
+    *,
+    combined_state: str | None = None,
+    statuses: Any = None,
+    checks_enabled: bool | None = None,
+    required_contexts: Any = None,
+    policy_determinable: bool = True,
+    status_determinable: bool = True,
+) -> dict[str, Any]:
+    """Classify head checks from live evidence (#751).
+
+    ``combined_state`` is deliberately **not** authoritative: it is recorded for
+    observability but never used to infer that CI is executing. The context
+    collection and the live protection policy decide.
+
+    Returns ``checks_status`` (one of the ``CHECKS_*`` values), the derived
+    ``checks_required`` flag, and structured ``reasons``.
+    """
+    reasons: list[str] = []
+    rows = _normalize_context_rows(statuses)
+    required = [
+        str(ctx).strip()
+        for ctx in (required_contexts or [])
+        if str(ctx or "").strip()
+    ]
+    observed_combined = (combined_state or "").strip().lower() or None
+
+    result: dict[str, Any] = {
+        "checks_status": CHECKS_UNKNOWN,
+        "checks_required": True,
+        "combined_state": observed_combined,
+        "context_count": len(rows),
+        "observed_contexts": [row["context"] for row in rows],
+        "required_contexts": required,
+        "missing_required_contexts": [],
+        "policy_determinable": bool(policy_determinable),
+        "status_determinable": bool(status_determinable),
+        "reasons": reasons,
+    }
+
+    # Policy unreadable → never assume checks are optional.
+    if not policy_determinable:
+        reasons.append(
+            "branch-protection check policy could not be read; cannot prove "
+            "whether status checks are required (fail closed)"
+        )
+        return result
+
+    if checks_enabled is False:
+        result["checks_required"] = False
+        result["checks_status"] = CHECKS_NOT_REQUIRED
+        reasons.append(
+            "live branch protection does not require status checks for the base "
+            "branch; head status contexts do not gate merge"
+        )
+        return result
+
+    if checks_enabled is None:
+        reasons.append(
+            "branch-protection status-check requirement is indeterminate "
+            "(fail closed)"
+        )
+        return result
+
+    # Checks are required from here on.
+    if not status_determinable:
+        reasons.append(
+            "head commit status collection could not be read while branch "
+            "protection requires status checks (fail closed)"
+        )
+        return result
+
+    if required:
+        by_context = {row["context"]: row["state"] for row in rows if row["context"]}
+        missing = [ctx for ctx in required if ctx not in by_context]
+        if missing:
+            result["missing_required_contexts"] = missing
+            result["checks_status"] = CHECKS_MISSING_REQUIRED
+            reasons.append(
+                "branch protection requires status context(s) "
+                f"{', '.join(missing)} but no matching status result exists at "
+                "the head commit (fail closed)"
+            )
+            return result
+        matched = [
+            {"context": ctx, "state": by_context[ctx]} for ctx in required
+        ]
+        result["checks_status"] = _aggregate_context_states(matched)
+        reasons.append(
+            f"evaluated {len(matched)} required status context(s) from live "
+            "branch protection; unrelated contexts were ignored"
+        )
+        return result
+
+    # Status checks enabled with no specific required contexts configured.
+    if not rows:
+        result["checks_status"] = CHECKS_NONE
+        reasons.append(
+            "branch protection enables status checks but no status context was "
+            "produced for the head commit"
+        )
+        if observed_combined in _CTX_PENDING:
+            reasons.append(
+                f"combined commit state '{observed_combined}' does not indicate "
+                "executing CI because the status-context collection is empty"
+            )
+        return result
+
+    result["checks_status"] = _aggregate_context_states(rows)
+    reasons.append(
+        f"aggregated {len(rows)} reported status context(s); branch protection "
+        "configures no explicit required-context list"
+    )
+    return result
+
 
 def _normalize_sha(value: str | None) -> str | None:
     text = (value or "").strip().lower()
@@ -120,6 +300,7 @@ def assess_pr_sync_status(
         "branch_protection_requires_current_base": requires_current,
         "approval_at_current_head": approval_ok if approval_at_current_head is not None else None,
         "checks_status": checks,
+        "checks_required": bool(checks_required),
         "active_locks_and_leases": {
             "author_lock": bool(active_author_lock) if active_author_lock is not None else None,
             "reviewer_lease": bool(active_reviewer_lease) if active_reviewer_lease is not None else None,
@@ -258,21 +439,41 @@ def assess_pr_sync_status(
         result["recommended_next_action"] = ACTION_BLOCKED
         return result
 
-    # ── Checks gate for merge_now ────────────────────────────────────────
-    if checks_required and checks not in ("success", "passed", "ok", "none", "skipped", "not_required"):
-        if checks in ("pending", "running", "queued"):
+    # ── Checks gate for merge_now (#751) ─────────────────────────────────
+    # ``checks_required`` is derived from the live branch-protection policy by
+    # the production caller. When protection does not require status checks,
+    # head contexts cannot gate the merge and this whole gate is skipped.
+    if not checks_required:
+        reasons.append(
+            "live branch protection does not require status checks; head check "
+            f"state ({checks}) does not gate merge"
+        )
+    elif checks not in _CHECKS_OK:
+        if checks in _CTX_PENDING:
             reasons.append(f"required checks are not finished (status={checks})")
-            result["recommended_next_action"] = ACTION_BLOCKED
-            return result
-        if checks in ("failure", "failed", "error", "cancelled"):
+        elif checks in _CTX_FAILURE:
             reasons.append(f"required checks failed (status={checks})")
-            result["recommended_next_action"] = ACTION_BLOCKED
-            return result
-        # unknown — fail closed when checks_required
-        if checks == "unknown":
+        elif checks == CHECKS_MISSING_REQUIRED:
+            reasons.append(
+                "branch protection configures required status context(s) but no "
+                "matching status result exists at the current head (fail closed)"
+            )
+        elif checks == CHECKS_NONE:
+            reasons.append(
+                "branch protection requires status checks but no status context "
+                "was produced for the current head (fail closed); an empty "
+                "status collection is not executing CI"
+            )
+        elif checks == CHECKS_UNKNOWN:
             reasons.append("checks status unknown (fail closed)")
-            result["recommended_next_action"] = ACTION_BLOCKED
-            return result
+        else:
+            # Unrecognized vocabulary must never fall through to merge_now.
+            reasons.append(
+                f"unrecognized checks status '{checks}' cannot prove required "
+                "checks passed (fail closed)"
+            )
+        result["recommended_next_action"] = ACTION_BLOCKED
+        return result
 
     # ── Ready to merge without update ────────────────────────────────────
     # Includes: current with approval; outdated when update is NOT required.

tests/test_issue_751_checks_assessor.py

@@ -0,0 +1,602 @@
+"""Regression coverage for the PR checks assessor defect (#751).
+
+Gitea's *combined* commit status reports ``state: pending`` both when a check is
+executing and when the status-context collection is empty. The assessor read
+``state`` alone and defaulted ``checks_required`` to ``True``, so a PR whose head
+had no status contexts — and never would — was routed to ``blocked`` forever.
+
+These tests pin the corrected semantics end to end: live branch protection
+decides whether checks are required, and the actual context collection decides
+what the checks say.
+"""
+
+from __future__ import annotations
+
+import sys
+import unittest
+from unittest.mock import patch
+
+sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
+
+import pr_sync_status  # noqa: E402
+from pr_sync_status import (  # noqa: E402
+    ACTION_BLOCKED,
+    ACTION_MERGE_NOW,
+    ACTION_UPDATE_BRANCH_BY_MERGE,
+    CHECKS_FAILURE,
+    CHECKS_MISSING_REQUIRED,
+    CHECKS_NONE,
+    CHECKS_NOT_REQUIRED,
+    CHECKS_PENDING,
+    CHECKS_SUCCESS,
+    CHECKS_UNKNOWN,
+    assess_pr_sync_status,
+    classify_commit_checks,
+)
+
+
+def _sha(prefix: str) -> str:
+    return (prefix + "0" * 40)[:40]
+
+
+PR_HEAD = _sha("aaaaaaaa")
+BASE_HEAD = _sha("bbbbbbbb")
+
+
+def _base_kwargs(**overrides):
+    data = {
+        "host": "gitea.prgs.cc",
+        "org": "Scaled-Tech-Consulting",
+        "repo": "Gitea-Tools",
+        "pr_number": 751,
+        "pr_state": "open",
+        "source_branch": "fix/issue-751-checks-assessor",
+        "pr_head_sha": PR_HEAD,
+        "base_head_sha": BASE_HEAD,
+        "commits_behind": 0,
+        "mergeable": True,
+        "has_conflicts": False,
+        "branch_protection_requires_current_base": False,
+        "approval_at_current_head": True,
+        "checks_status": "success",
+        "active_author_lock": True,
+        "active_reviewer_lease": False,
+        "active_merger_lease": False,
+    }
+    data.update(overrides)
+    return data
+
+
+def _reasons(result) -> str:
+    return " | ".join(result["reasons"]).lower()
+
+
+class TestClassifyCommitChecks(unittest.TestCase):
+    """Pure classification from live evidence."""
+
+    def test_empty_collection_with_combined_pending_is_not_executing_ci(self):
+        # The exact PR #750 failure mode at the classification layer.
+        result = classify_commit_checks(
+            combined_state="pending",
+            statuses=[],
+            checks_enabled=True,
+            required_contexts=[],
+        )
+        self.assertNotEqual(result["checks_status"], CHECKS_PENDING)
+        self.assertEqual(result["checks_status"], CHECKS_NONE)
+        self.assertEqual(result["context_count"], 0)
+        joined = " ".join(result["reasons"]).lower()
+        self.assertIn("empty", joined)
+        self.assertIn("does not indicate", joined)
+
+    def test_protection_disables_status_checks(self):
+        result = classify_commit_checks(
+            combined_state="pending",
+            statuses=[],
+            checks_enabled=False,
+            required_contexts=[],
+        )
+        self.assertFalse(result["checks_required"])
+        self.assertEqual(result["checks_status"], CHECKS_NOT_REQUIRED)
+
+    def test_no_required_contexts_aggregates_reported_contexts(self):
+        result = classify_commit_checks(
+            combined_state="success",
+            statuses=[{"context": "build", "status": "success"}],
+            checks_enabled=True,
+            required_contexts=[],
+        )
+        self.assertTrue(result["checks_required"])
+        self.assertEqual(result["checks_status"], CHECKS_SUCCESS)
+
+    def test_required_checks_pending(self):
+        result = classify_commit_checks(
+            combined_state="pending",
+            statuses=[{"context": "build", "status": "pending"}],
+            checks_enabled=True,
+            required_contexts=["build"],
+        )
+        self.assertEqual(result["checks_status"], CHECKS_PENDING)
+
+    def test_required_checks_failed(self):
+        result = classify_commit_checks(
+            combined_state="failure",
+            statuses=[{"context": "build", "status": "failure"}],
+            checks_enabled=True,
+            required_contexts=["build"],
+        )
+        self.assertEqual(result["checks_status"], CHECKS_FAILURE)
+
+    def test_required_checks_successful(self):
+        result = classify_commit_checks(
+            combined_state="success",
+            statuses=[{"context": "build", "status": "success"}],
+            checks_enabled=True,
+            required_contexts=["build"],
+        )
+        self.assertEqual(result["checks_status"], CHECKS_SUCCESS)
+        self.assertTrue(result["checks_required"])
+
+    def test_required_context_configured_with_no_matching_result(self):
+        result = classify_commit_checks(
+            combined_state="success",
+            statuses=[{"context": "lint", "status": "success"}],
+            checks_enabled=True,
+            required_contexts=["build"],
+        )
+        self.assertEqual(result["checks_status"], CHECKS_MISSING_REQUIRED)
+        self.assertEqual(result["missing_required_contexts"], ["build"])
+
+    def test_mixed_required_and_unrelated_contexts_ignores_unrelated(self):
+        # An unrelated failing context must not fail a satisfied required set.
+        result = classify_commit_checks(
+            combined_state="failure",
+            statuses=[
+                {"context": "build", "status": "success"},
+                {"context": "optional-scan", "status": "failure"},
+            ],
+            checks_enabled=True,
+            required_contexts=["build"],
+        )
+        self.assertEqual(result["checks_status"], CHECKS_SUCCESS)
+        # ...and a failing *required* context still fails despite passing extras.
+        failing = classify_commit_checks(
+            combined_state="success",
+            statuses=[
+                {"context": "build", "status": "failure"},
+                {"context": "optional-scan", "status": "success"},
+            ],
+            checks_enabled=True,
+            required_contexts=["build"],
+        )
+        self.assertEqual(failing["checks_status"], CHECKS_FAILURE)
+
+    def test_policy_unreadable_fails_closed(self):
+        result = classify_commit_checks(
+            combined_state=None,
+            statuses=[],
+            checks_enabled=None,
+            required_contexts=[],
+            policy_determinable=False,
+        )
+        self.assertTrue(result["checks_required"])
+        self.assertEqual(result["checks_status"], CHECKS_UNKNOWN)
+        self.assertIn("fail closed", " ".join(result["reasons"]).lower())
+
+    def test_malformed_policy_indeterminate_flag_fails_closed(self):
+        result = classify_commit_checks(
+            combined_state="success",
+            statuses=[{"context": "build", "status": "success"}],
+            checks_enabled=None,
+            required_contexts=[],
+            policy_determinable=True,
+        )
+        self.assertTrue(result["checks_required"])
+        self.assertEqual(result["checks_status"], CHECKS_UNKNOWN)
+
+    def test_status_collection_unreadable_fails_closed_when_required(self):
+        result = classify_commit_checks(
+            checks_enabled=True,
+            required_contexts=["build"],
+            status_determinable=False,
+        )
+        self.assertTrue(result["checks_required"])
+        self.assertEqual(result["checks_status"], CHECKS_UNKNOWN)
+
+    def test_unrecognized_context_state_never_reads_as_success(self):
+        result = classify_commit_checks(
+            combined_state="success",
+            statuses=[{"context": "build", "status": "banana"}],
+            checks_enabled=True,
+            required_contexts=["build"],
+        )
+        self.assertEqual(result["checks_status"], CHECKS_UNKNOWN)
+
+    def test_newest_wins_per_context(self):
+        # Gitea returns newest-first; the stale failure must not win.
+        result = classify_commit_checks(
+            combined_state="success",
+            statuses=[
+                {"context": "build", "status": "success"},
+                {"context": "build", "status": "failure"},
+            ],
+            checks_enabled=True,
+            required_contexts=["build"],
+        )
+        self.assertEqual(result["checks_status"], CHECKS_SUCCESS)
+
+    def test_malformed_status_rows_are_discarded_not_treated_as_passing(self):
+        result = classify_commit_checks(
+            combined_state="pending",
+            statuses=[{"context": "build"}, "not-a-dict", None],
+            checks_enabled=True,
+            required_contexts=["build"],
+        )
+        self.assertEqual(result["checks_status"], CHECKS_MISSING_REQUIRED)
+
+
+class TestChecksGateSemantics(unittest.TestCase):
+    """``assess_pr_sync_status`` routing and blocker reasons."""
+
+    def test_not_required_allows_merge_now_with_empty_checks(self):
+        result = assess_pr_sync_status(
+            **_base_kwargs(checks_status=CHECKS_NOT_REQUIRED),
+            checks_required=False,
+        )
+        self.assertEqual(result["recommended_next_action"], ACTION_MERGE_NOW)
+        self.assertTrue(result["approval_valid_for_merge"])
+        self.assertFalse(result["checks_required"])
+        self.assertIn("does not require status checks", _reasons(result))
+
+    def test_none_blocks_when_checks_required(self):
+        result = assess_pr_sync_status(
+            **_base_kwargs(checks_status=CHECKS_NONE),
+            checks_required=True,
+        )
+        self.assertEqual(result["recommended_next_action"], ACTION_BLOCKED)
+        self.assertIn("no status context", _reasons(result))
+        self.assertIn("not executing ci", _reasons(result))
+
+    def test_missing_required_blocks(self):
+        result = assess_pr_sync_status(
+            **_base_kwargs(checks_status=CHECKS_MISSING_REQUIRED),
+            checks_required=True,
+        )
+        self.assertEqual(result["recommended_next_action"], ACTION_BLOCKED)
+        self.assertIn("no matching status result", _reasons(result))
+
+    def test_required_pending_blocks(self):
+        result = assess_pr_sync_status(
+            **_base_kwargs(checks_status=CHECKS_PENDING), checks_required=True
+        )
+        self.assertEqual(result["recommended_next_action"], ACTION_BLOCKED)
+        self.assertIn("not finished", _reasons(result))
+
+    def test_required_failure_blocks(self):
+        result = assess_pr_sync_status(
+            **_base_kwargs(checks_status=CHECKS_FAILURE), checks_required=True
+        )
+        self.assertEqual(result["recommended_next_action"], ACTION_BLOCKED)
+        self.assertIn("failed", _reasons(result))
+
+    def test_required_success_merges(self):
+        result = assess_pr_sync_status(
+            **_base_kwargs(checks_status=CHECKS_SUCCESS), checks_required=True
+        )
+        self.assertEqual(result["recommended_next_action"], ACTION_MERGE_NOW)
+
+    def test_unknown_blocks_when_required(self):
+        result = assess_pr_sync_status(
+            **_base_kwargs(checks_status=CHECKS_UNKNOWN), checks_required=True
+        )
+        self.assertEqual(result["recommended_next_action"], ACTION_BLOCKED)
+        self.assertIn("unknown", _reasons(result))
+
+    def test_unrecognized_status_does_not_fall_through_to_merge(self):
+        # Regression: the previous gate only handled a fixed vocabulary and let
+        # anything else reach merge_now.
+        result = assess_pr_sync_status(
+            **_base_kwargs(checks_status="totally-unexpected"),
+            checks_required=True,
+        )
+        self.assertEqual(result["recommended_next_action"], ACTION_BLOCKED)
+        self.assertIn("unrecognized checks status", _reasons(result))
+
+    def test_checks_gate_does_not_bypass_approval_requirement(self):
+        result = assess_pr_sync_status(
+            **_base_kwargs(
+                checks_status=CHECKS_NOT_REQUIRED, approval_at_current_head=False
+            ),
+            checks_required=False,
+        )
+        self.assertNotEqual(result["recommended_next_action"], ACTION_MERGE_NOW)
+        self.assertFalse(result["approval_valid_for_merge"])
+
+    def test_checks_gate_does_not_bypass_current_base_protection(self):
+        # Existing current-base behavior stays intact (#727).
+        result = assess_pr_sync_status(
+            **_base_kwargs(
+                checks_status=CHECKS_NOT_REQUIRED,
+                commits_behind=3,
+                branch_protection_requires_current_base=True,
+            ),
+            checks_required=False,
+        )
+        self.assertEqual(
+            result["recommended_next_action"], ACTION_UPDATE_BRANCH_BY_MERGE
+        )
+
+    def test_checks_gate_does_not_bypass_conflicts(self):
+        result = assess_pr_sync_status(
+            **_base_kwargs(checks_status=CHECKS_NOT_REQUIRED, mergeable=False),
+            checks_required=False,
+        )
+        self.assertNotEqual(result["recommended_next_action"], ACTION_MERGE_NOW)
+
+
+class TestBranchProtectionPolicy(unittest.TestCase):
+    """Live protection reader derives the status-check requirement."""
+
+    def setUp(self):
+        import gitea_mcp_server as gms
+
+        self.gms = gms
+
+    def _policy(self, responses):
+        def fake_api_request(method, url, auth, *args, **kwargs):
+            for fragment, payload in responses.items():
+                if fragment in url:
+                    if isinstance(payload, Exception):
+                        raise payload
+                    return payload
+            return None
+
+        with patch.object(self.gms, "api_request", side_effect=fake_api_request):
+            return self.gms._branch_protection_policy(
+                "https://example/api/v1/repos/o/r", {"h": "1"}, base_branch="master"
+            )
+
+    def test_status_checks_enabled_with_contexts(self):
+        policy = self._policy({
+            "branch_protections": [{
+                "branch_name": "master",
+                "block_on_outdated_branch": True,
+                "enable_status_check": True,
+                "status_check_contexts": ["ci/build", " "],
+            }],
+        })
+        self.assertTrue(policy["determinable"])
+        self.assertTrue(policy["checks_enabled"])
+        self.assertTrue(policy["requires_current_base"])
+        self.assertEqual(policy["required_contexts"], ["ci/build"])
+
+    def test_status_checks_disabled(self):
+        policy = self._policy({
+            "branch_protections": [{
+                "branch_name": "master",
+                "enable_status_check": False,
+            }],
+        })
+        self.assertTrue(policy["determinable"])
+        self.assertFalse(policy["checks_enabled"])
+
+    def test_no_protection_rule_means_checks_not_required(self):
+        # PR #750's repository shape: protection list readable but empty.
+        policy = self._policy({"branch_protections": [], "branches/master": {}})
+        self.assertTrue(policy["determinable"])
+        self.assertFalse(policy["protection_found"])
+        self.assertFalse(policy["checks_enabled"])
+        self.assertIsNone(policy["requires_current_base"])
+
+    def test_protection_without_status_field_means_not_enabled(self):
+        policy = self._policy({
+            "branch_protections": [{
+                "branch_name": "master",
+                "block_on_outdated_branch": True,
+            }],
+        })
+        self.assertTrue(policy["determinable"])
+        self.assertFalse(policy["checks_enabled"])
+        self.assertTrue(policy["requires_current_base"])
+
+    def test_api_failure_is_not_determinable(self):
+        policy = self._policy({
+            "branch_protections": RuntimeError("boom"),
+        })
+        self.assertFalse(policy["determinable"])
+        self.assertIsNone(policy["checks_enabled"])
+
+    def test_missing_branch_is_not_determinable(self):
+        with patch.object(self.gms, "api_request", return_value=[]):
+            policy = self.gms._branch_protection_policy(
+                "https://example/api/v1/repos/o/r", {}, base_branch="  "
+            )
+        self.assertFalse(policy["determinable"])
+
+    def test_requires_current_base_accessor_preserved(self):
+        def fake(method, url, auth, *a, **k):
+            if "branch_protections" in url:
+                return [{"branch_name": "master",
+                         "block_on_outdated_branch": True}]
+            return None
+
+        with patch.object(self.gms, "api_request", side_effect=fake):
+            value = self.gms._branch_protection_requires_current_base(
+                "https://example/api/v1/repos/o/r", {}, base_branch="master"
+            )
+        self.assertTrue(value)
+
+
+class TestCommitChecksSnapshot(unittest.TestCase):
+    def setUp(self):
+        import gitea_mcp_server as gms
+
+        self.gms = gms
+
+    def test_reads_state_and_context_collection(self):
+        payload = {"state": "pending", "statuses": [{"context": "b",
+                                                     "status": "pending"}]}
+        with patch.object(self.gms, "api_request", return_value=payload):
+            snap = self.gms._commit_checks_snapshot(
+                "https://example/api/v1/repos/o/r", {}, sha=PR_HEAD
+            )
+        self.assertTrue(snap["determinable"])
+        self.assertEqual(snap["combined_state"], "pending")
+        self.assertEqual(len(snap["statuses"]), 1)
+
+    def test_empty_collection_recorded_as_determinable(self):
+        with patch.object(
+            self.gms, "api_request", return_value={"state": "pending", "statuses": []}
+        ):
+            snap = self.gms._commit_checks_snapshot(
+                "https://example/api/v1/repos/o/r", {}, sha=PR_HEAD
+            )
+        self.assertTrue(snap["determinable"])
+        self.assertEqual(snap["statuses"], [])
+
+    def test_api_failure_is_not_determinable(self):
+        with patch.object(self.gms, "api_request", side_effect=RuntimeError("x")):
+            snap = self.gms._commit_checks_snapshot(
+                "https://example/api/v1/repos/o/r", {}, sha=PR_HEAD
+            )
+        self.assertFalse(snap["determinable"])
+
+
+class TestMcpWrapperForwarding(unittest.TestCase):
+    """The production MCP path must reach the derived ``checks_required``."""
+
+    def setUp(self):
+        import gitea_mcp_server as gms
+
+        self.gms = gms
+        self.base = (
+            "https://gitea.prgs.cc/api/v1/repos/Scaled-Tech-Consulting/Gitea-Tools"
+        )
+
+    def _patches(self, fake_api_request, spy):
+        return [
+            patch.object(self.gms, "_profile_operation_gate", return_value=None),
+            patch.object(self.gms, "_resolve", return_value=(
+                "gitea.prgs.cc", "Scaled-Tech-Consulting", "Gitea-Tools")),
+            patch.object(self.gms, "_auth", return_value={"Authorization": "x"}),
+            patch.object(self.gms, "repo_api_url", return_value=self.base),
+            patch.object(self.gms, "api_request", side_effect=fake_api_request),
+            patch.object(self.gms, "gitea_get_pr_review_feedback", return_value={
+                "success": True, "approval_at_current_head": True}),
+            patch.object(self.gms, "_prove_author_ownership_for_pr", return_value={
+                "has_author_lock": True}),
+            patch.object(pr_sync_status, "assess_pr_sync_status", side_effect=spy),
+        ]
+
+    def _run(self, *, protections, status_payload, pr_number=750,
+             caller_checks_status=None):
+        captured = {}
+        real_assess = pr_sync_status.assess_pr_sync_status
+
+        def spy(**kwargs):
+            captured.update(kwargs)
+            return real_assess(**kwargs)
+
+        def fake_api_request(method, url, auth, *args, **kwargs):
+            if f"/pulls/{pr_number}" in url:
+                return {
+                    "number": pr_number,
+                    "state": "open",
+                    "title": "t",
+                    "body": "b",
+                    "mergeable": True,
+                    "head": {"sha": PR_HEAD, "ref": "fix/x"},
+                    "base": {"sha": BASE_HEAD, "ref": "master"},
+                }
+            if "/branch_protections" in url:
+                if isinstance(protections, Exception):
+                    raise protections
+                return protections
+            if "/branches/master" in url:
+                return {"commit": {"id": BASE_HEAD}}
+            if "/status" in url:
+                if isinstance(status_payload, Exception):
+                    raise status_payload
+                return status_payload
+            if "/compare/" in url:
+                return {"total_commits": 0}
+            if "/comments" in url:
+                return []
+            return None
+
+        stack = self._patches(fake_api_request, spy)
+        for p in stack:
+            p.start()
+        try:
+            result = self.gms.gitea_assess_pr_sync_status(
+                pr_number=pr_number, remote="prgs",
+                checks_status=caller_checks_status,
+            )
+        finally:
+            for p in reversed(stack):
+                p.stop()
+        return result, captured
+
+    def test_derived_checks_required_is_forwarded(self):
+        result, captured = self._run(
+            protections=[{"branch_name": "master", "enable_status_check": True,
+                          "status_check_contexts": ["ci/build"]}],
+            status_payload={"state": "pending", "statuses": [
+                {"context": "ci/build", "status": "pending"}]},
+        )
+        self.assertIn("checks_required", captured)
+        self.assertTrue(captured["checks_required"])
+        self.assertEqual(captured["checks_status"], CHECKS_PENDING)
+        self.assertEqual(result["recommended_next_action"], ACTION_BLOCKED)
+
+    def test_pr750_empty_status_with_no_protection_is_merge_ready(self):
+        # The exact reported failure: combined pending, zero contexts, and no
+        # branch protection requiring checks.
+        result, captured = self._run(
+            protections=[],
+            status_payload={"state": "pending", "statuses": []},
+        )
+        self.assertFalse(captured["checks_required"])
+        self.assertEqual(captured["checks_status"], CHECKS_NOT_REQUIRED)
+        self.assertNotEqual(result["checks_status"], CHECKS_PENDING)
+        self.assertEqual(result["recommended_next_action"], ACTION_MERGE_NOW)
+
+    def test_protection_read_failure_blocks(self):
+        result, captured = self._run(
+            protections=RuntimeError("protection unavailable"),
+            status_payload={"state": "success", "statuses": []},
+        )
+        self.assertTrue(captured["checks_required"])
+        self.assertEqual(result["recommended_next_action"], ACTION_BLOCKED)
+
+    def test_caller_supplied_status_cannot_mask_live_required_failure(self):
+        # A caller-declared "success" must not override live evidence.
+        result, captured = self._run(
+            protections=[{"branch_name": "master", "enable_status_check": True,
+                          "status_check_contexts": ["ci/build"]}],
+            status_payload={"state": "failure", "statuses": [
+                {"context": "ci/build", "status": "failure"}]},
+            caller_checks_status="success",
+        )
+        self.assertEqual(captured["checks_status"], CHECKS_FAILURE)
+        self.assertEqual(result["recommended_next_action"], ACTION_BLOCKED)
+        self.assertEqual(
+            result["checks_evidence"]["caller_supplied_checks_status"], "success"
+        )
+
+    def test_checks_evidence_is_reported_without_secrets(self):
+        result, _ = self._run(
+            protections=[],
+            status_payload={"state": "pending", "statuses": []},
+        )
+        evidence = result["checks_evidence"]
+        self.assertEqual(evidence["context_count"], 0)
+        self.assertEqual(evidence["combined_state"], "pending")
+        self.assertFalse(evidence["protection_found"])
+        blob = repr(result).lower()
+        self.assertNotIn("authorization", blob)
+        self.assertNotIn("token", blob)
+
+
+if __name__ == "__main__":
+    unittest.main()