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

Merge branch 'master' into fix/issue-749-create-issue-bootstrap

Browse Source
This commit is contained in:
jcwalker3
2026-07-18 21:16:23 -05:00
parent e349839fd7 043df0f7df
commit 5e0e474350
9 changed files with 2702 additions and 74 deletions
Download Patch File Download Diff File Expand all files Collapse all files
gitea_mcp_server.py
+296 -54
View File
@@ -1700,6 +1700,7 @@ import issue_lock_worktree # noqa: E402
import issue_lock_provenance # noqa: E402
import issue_lock_store # noqa: E402
import issue_lock_adoption # noqa: E402
import issue_lock_recovery # noqa: E402
import stacked_pr_support # noqa: E402
import merge_approval_gate # noqa: E402
import review_quarantine # noqa: E402 # #695 contaminated formal-review quarantine
@@ -2130,6 +2131,7 @@ def _assess_issue_duplicate_gate(
auth: str,
locked_branch: str | None = None,
phase: str,
recovered_owning_pr: dict | None = None,
) -> dict:
open_prs, branch_names, claim_entry = _collect_issue_duplicate_context(
h, o, r, auth, issue_number
@@ -2141,6 +2143,7 @@ def _assess_issue_duplicate_gate(
claim_entry=claim_entry,
locked_branch=locked_branch,
phase=phase,
recovered_owning_pr=recovered_owning_pr,
)
@@ -3205,8 +3208,11 @@ def gitea_lock_issue(
worktree_path, _canonical_local_git_root()
)
h, o, r = _resolve(remote, host, org, repo)
existing_issue_lock = _load_existing_issue_lock(
remote=remote, org=o, repo=r, issue_number=issue_number
)
active_lease_block = issue_lock_store.assess_same_issue_lease_conflict(
_load_existing_issue_lock(remote=remote, org=o, repo=r, issue_number=issue_number),
existing_issue_lock,
issue_number=issue_number,
branch_name=branch_name,
worktree_path=resolved_worktree,
@@ -3248,6 +3254,83 @@ def gitea_lock_issue(
org=org,
repo=repo,
)
# ── Dead-session lock recovery assessment (#753) ──
# When the MCP session that took a lock exits, the lock goes non-live
# (stale by dead PID) even inside its lease TTL, and the branch it owns is
# ahead of its base by construction — so the base-equivalence gate below
# makes normal re-lock unreachable for every PR that already exists.
#
# This grants a waiver ONLY for that case, proven against the durable lock
# record plus live git/Gitea observation. A refused assessment never raises:
# it simply withholds the waiver, leaving the pre-existing guard to fail
# closed exactly as before. Recovery can only ever add permission.
recovery_assessment: dict | None = None
if (
existing_issue_lock
and existing_issue_lock.get("issue_number") == issue_number
and not issue_lock_store.is_lease_live(existing_issue_lock)
):
recovery_auth = _auth(h)
try:
recovery_branches = api_get_all(
f"{repo_api_url(h, o, r)}/branches", recovery_auth
)
except Exception as exc:
raise RuntimeError(
f"Could not list branches to verify issue-lock recovery: {exc}"
)
recovery_remote_head: str | None = None
recovery_candidates: list[str] = []
for entry in recovery_branches:
entry_name = _branch_entry_name(entry)
if entry_name == branch_name:
recovery_remote_head = _branch_entry_commit_sha(entry)
if issue_lock_adoption.branch_carries_issue_marker(entry_name, issue_number):
recovery_candidates.append(entry_name)
recovery_pr_head: str | None = None
recovery_pr_number: int | None = None
for pull in _list_open_pulls(h, o, r, recovery_auth):
pull_head = pull.get("head") or {}
if str(pull_head.get("ref") or "") == branch_name:
recovery_pr_head = pull_head.get("sha")
recovery_pr_number = pull.get("number")
break
recovery_claimant = _work_lease_claimant(h)
recovery_assessment = issue_lock_recovery.assess_dead_session_lock_recovery(
existing_issue_lock,
issue_number=issue_number,
branch_name=branch_name,
worktree_path=resolved_worktree,
remote=remote,
org=o,
repo=r,
identity=recovery_claimant.get("username"),
profile=recovery_claimant.get("profile"),
current_branch=git_state.get("current_branch"),
porcelain_status=git_state.get("porcelain_status") or "",
head_sha=git_state.get("head_sha"),
remote_head_sha=recovery_remote_head,
pr_head_sha=recovery_pr_head,
pr_number=recovery_pr_number,
competing_live_locks=issue_lock_store.list_live_locks(),
candidate_branches=recovery_candidates,
current_pid=os.getpid(),
)
recovery_sanctioned = bool(
recovery_assessment and recovery_assessment.get("recovery_sanctioned")
)
# #755: a sanctioned dead-session recovery always has an owning open PR —
# that is what makes it a recovery rather than a fresh claim. Carry the
# server-derived owning-PR evidence into the duplicate-work gate below so
# the PR this lock already owns is not mistaken for competing duplicate
# work. Withheld (None) unless recovery was granted, so the ordinary
# duplicate blocker is untouched on every other path.
recovered_owning_pr = (
issue_lock_recovery.owning_pr_recovery_evidence(recovery_assessment)
if recovery_sanctioned
else None
)
lock_assessment = issue_lock_worktree.assess_issue_lock_worktree(
worktree_path=resolved_worktree,
current_branch=git_state.get("current_branch"),
@@ -3255,10 +3338,20 @@ def gitea_lock_issue(
base_equivalent=git_state.get("base_equivalent"),
inspected_git_root=git_state.get("inspected_git_root"),
base_branch=git_state.get("base_branch"),
recovery_sanctioned=recovery_sanctioned,
)
if lock_assessment["block"]:
reasons = list(lock_assessment.get("reasons") or [])
# Surface why recovery was unavailable, so a blocked caller sees the
# exact missing evidence instead of only the base-equivalence text.
if recovery_assessment and recovery_assessment.get("is_candidate"):
reasons.append(
issue_lock_recovery.format_recovery_refusal(recovery_assessment)
)
raise RuntimeError(
issue_lock_worktree.format_issue_lock_worktree_error(lock_assessment)
issue_lock_worktree.format_issue_lock_worktree_error(
{**lock_assessment, "reasons": reasons}
)
)
auth = _auth(h)
@@ -3270,6 +3363,7 @@ def gitea_lock_issue(
auth=auth,
locked_branch=branch_name,
phase=issue_work_duplicate_gate.PHASE_LOCK,
recovered_owning_pr=recovered_owning_pr,
)
if duplicate_gate.get("block"):
raise ValueError("; ".join(duplicate_gate.get("reasons") or [
@@ -3321,6 +3415,14 @@ def gitea_lock_issue(
}
if stacked_approved:
data["approved_stacked_base"] = stacked_approved
if recovery_sanctioned and recovery_assessment:
# #753 AC2/AC6: record that this claim was recovered after session
# death, with the prior and replacement session identity, so the
# takeover is auditable and never looks like an original claim.
data["dead_session_recovery"] = issue_lock_recovery.build_recovery_record(
recovery_assessment,
recovered_at=_work_lease_timestamp(_work_lease_now()),
)
lock_file_path = _save_issue_lock(data)
lock_record = issue_lock_store.read_lock_file(lock_file_path) or data
@@ -3354,6 +3456,14 @@ def gitea_lock_issue(
"lock_freshness": freshness,
"lock_proof": lock_proof,
}
if recovery_sanctioned and recovery_assessment:
result["dead_session_recovery"] = data["dead_session_recovery"]
result["message"] = (
f"Recovered the durable lock for issue #{issue_number} on branch "
f"'{branch_name}' after the owning MCP session (pid "
f"{recovery_assessment['evidence'].get('prior_session_pid')}) exited; "
"ownership evidence matched exactly (fail-closed check complete)."
)
if stacked_approved:
result["approved_stacked_base"] = stacked_approved
result["message"] = (
@@ -15417,46 +15527,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(
@@ -15701,11 +15916,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
@@ -15772,23 +15990,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,
@@ -15809,9 +16034,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