Scaled-Tech-Consulting/Gitea-Tools
2
0
Fork 0
Code Issues 72 Pull Requests Actions Packages Projects Releases Wiki Activity

fix(mcp): derive checks requirement from live branch protection (Closes #751)

Browse Source 
`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]>
This commit is contained in:
sysadmin
2026-07-18 17:38:54 -04:00
co-authored by Claude Opus 4.8
parent fdab6b6c69
commit eac7afe5cb
3 changed files with 999 additions and 64 deletions
Download Patch File Download Diff File Expand all files Collapse all files
gitea_mcp_server.py
+184 -52
View File
@@ -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