656 lines
24 KiB
Python
656 lines
24 KiB
Python
"""Issue-lock worktree validation (#249).
|
|
|
|
Author issue locks must validate the caller's own scratch clone (or declared
|
|
worktree path), not the shared MCP server working directory. A clean scratch at
|
|
``master``/``main`` must remain lockable while an unrelated session leaves the
|
|
shared dev worktree dirty or on a feature branch.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import subprocess
|
|
|
|
from reviewer_worktree import parse_dirty_tracked_files
|
|
|
|
AUTHOR_WORKTREE_ENV = "GITEA_AUTHOR_WORKTREE"
|
|
BASE_BRANCHES = frozenset({"master", "main", "dev"})
|
|
|
|
|
|
def resolve_author_worktree_path(
|
|
explicit: str | None,
|
|
project_root: str,
|
|
*,
|
|
session_lock_worktree: str | None = None,
|
|
) -> str:
|
|
"""Resolve the author worktree path for lock/PR gates.
|
|
|
|
#618: prefer explicit path, then env, then the active issue lock worktree.
|
|
Does not invent a branches/ worktree. Falling back to *project_root* is
|
|
retained only for lock-time bootstrap when the process itself is already
|
|
under branches/ or no binding exists yet (callers still fail closed via
|
|
preflight / durable resolution before mutation).
|
|
"""
|
|
path = (explicit or "").strip()
|
|
if not path:
|
|
path = (os.environ.get(AUTHOR_WORKTREE_ENV) or "").strip()
|
|
if not path:
|
|
path = (os.environ.get("GITEA_ACTIVE_WORKTREE") or "").strip()
|
|
if not path:
|
|
path = (session_lock_worktree or "").strip()
|
|
if not path:
|
|
path = project_root
|
|
return os.path.realpath(os.path.abspath(path))
|
|
|
|
|
|
def read_head_ancestry(
|
|
worktree_path: str,
|
|
*,
|
|
ancestor_sha: str | None,
|
|
descendant_sha: str | None,
|
|
) -> dict:
|
|
"""Observe whether ``descendant_sha`` strictly descends from ``ancestor_sha`` (#768).
|
|
|
|
Server-side git observation for dead-session lock recovery. The recovering
|
|
author's only reachable clean-worktree state is one commit *ahead* of the
|
|
head recorded at lock time, so recovery needs to know whether that commit
|
|
extends the recorded head or replaces it.
|
|
|
|
Reports facts only; the disposition lives in ``issue_lock_recovery``. Every
|
|
field is read from git in the declared worktree — nothing here is supplied
|
|
by, or reachable from, an MCP caller (#768 AC6).
|
|
|
|
``ancestor_present`` proves the recorded head is still reachable, which is
|
|
what separates an honest fast-forward from a rewritten or force-moved
|
|
history: a rewritten recorded head leaves the object graph and the probe
|
|
fails closed.
|
|
"""
|
|
path = (worktree_path or "").strip()
|
|
ancestor = (ancestor_sha or "").strip()
|
|
descendant = (descendant_sha or "").strip()
|
|
result: dict = {
|
|
"ancestor_sha": ancestor or None,
|
|
"descendant_sha": descendant or None,
|
|
"probe_ok": False,
|
|
"ancestor_present": False,
|
|
"descendant_present": False,
|
|
"is_ancestor": False,
|
|
"is_strict_descendant": False,
|
|
"proof": None,
|
|
"reasons": [],
|
|
}
|
|
if not path or not ancestor or not descendant:
|
|
result["reasons"].append(
|
|
"ancestry probe requires a worktree path and both commit SHAs"
|
|
)
|
|
return result
|
|
|
|
def _present(sha: str) -> bool:
|
|
res = subprocess.run(
|
|
["git", "-C", path, "rev-parse", "--verify", "--quiet", f"{sha}^{{commit}}"],
|
|
capture_output=True,
|
|
text=True,
|
|
check=False,
|
|
)
|
|
return res.returncode == 0
|
|
|
|
try:
|
|
result["ancestor_present"] = _present(ancestor)
|
|
result["descendant_present"] = _present(descendant)
|
|
except OSError as exc: # git unavailable — fail closed, never assume
|
|
result["reasons"].append(f"ancestry probe could not run: {exc}")
|
|
return result
|
|
|
|
if not result["ancestor_present"]:
|
|
result["reasons"].append(
|
|
f"recorded head {ancestor} is not reachable in '{path}'; history may "
|
|
"have been rewritten or force-moved"
|
|
)
|
|
if not result["descendant_present"]:
|
|
result["reasons"].append(
|
|
f"local head {descendant} is not reachable in '{path}'"
|
|
)
|
|
if not (result["ancestor_present"] and result["descendant_present"]):
|
|
return result
|
|
|
|
probe = subprocess.run(
|
|
["git", "-C", path, "merge-base", "--is-ancestor", ancestor, descendant],
|
|
capture_output=True,
|
|
text=True,
|
|
check=False,
|
|
)
|
|
# 0 = is an ancestor, 1 = is not. Anything else is a failed probe, not a "no".
|
|
if probe.returncode not in (0, 1):
|
|
result["reasons"].append(
|
|
f"ancestry probe failed with exit {probe.returncode}; ancestry unproven"
|
|
)
|
|
return result
|
|
|
|
result["probe_ok"] = True
|
|
result["is_ancestor"] = probe.returncode == 0
|
|
result["is_strict_descendant"] = result["is_ancestor"] and ancestor != descendant
|
|
result["proof"] = (
|
|
f"git -C <worktree> merge-base --is-ancestor {ancestor} {descendant} "
|
|
f"-> exit {probe.returncode}"
|
|
)
|
|
if not result["is_ancestor"]:
|
|
result["reasons"].append(
|
|
f"local head {descendant} does not descend from recorded head {ancestor}"
|
|
)
|
|
elif not result["is_strict_descendant"]:
|
|
result["reasons"].append(
|
|
f"local head {descendant} equals the recorded head; no descendant "
|
|
"recovery is involved"
|
|
)
|
|
return result
|
|
|
|
|
|
def read_merge_sync_provenance(
|
|
worktree_path: str,
|
|
*,
|
|
prior_head_sha: str | None,
|
|
synced_head_sha: str | None,
|
|
remote: str | None = None,
|
|
) -> dict:
|
|
"""Observe whether ``synced_head_sha`` is a sanctioned merge-sync of a base
|
|
into the branch above ``prior_head_sha`` (#871/#872).
|
|
|
|
Reports server-derived git facts only; the recovery disposition lives in
|
|
``issue_lock_recovery``. All comparisons are executed locally in the
|
|
declared worktree -- nothing is taken from caller parameters.
|
|
|
|
Provenance is proven only when ALL hold:
|
|
|
|
* both commits are present (a rewritten/force-moved prior head leaves the
|
|
object graph and fails closed);
|
|
* ``prior_head_sha`` is a strict ancestor of ``synced_head_sha`` (the branch
|
|
history is preserved, never replaced);
|
|
* ``synced_head_sha`` is a merge commit (two or more parents), i.e. a base
|
|
merged in — a plain fast-forward of new direct commits is not a sync;
|
|
* ``prior_head_sha`` is an ancestor of the merge's **first** parent, so the
|
|
branch mainline (first-parent lineage) still reaches the prior head — a
|
|
rebase/force-push that re-authored the branch side fails this.
|
|
"""
|
|
path = (worktree_path or "").strip()
|
|
prior = (prior_head_sha or "").strip()
|
|
synced = (synced_head_sha or "").strip()
|
|
target_remote = (remote or "").strip() or None
|
|
result: dict = {
|
|
"prior_head_sha": prior or None,
|
|
"synced_head_sha": synced or None,
|
|
"probe_ok": False,
|
|
"prior_present": False,
|
|
"synced_present": False,
|
|
"prior_is_ancestor": False,
|
|
"synced_is_merge": False,
|
|
"first_parent_reaches_prior": False,
|
|
"is_merge_sync": False,
|
|
"first_parent_sha": None,
|
|
"parent_count": None,
|
|
"proof": None,
|
|
"reasons": [],
|
|
}
|
|
if not path or not prior or not synced:
|
|
result["reasons"].append(
|
|
"merge-sync provenance probe requires a worktree path and both "
|
|
"commit SHAs"
|
|
)
|
|
return result
|
|
if prior == synced:
|
|
result["reasons"].append(
|
|
"prior and synced heads are identical; no branch sync occurred"
|
|
)
|
|
return result
|
|
|
|
def _present(sha: str) -> bool:
|
|
res = subprocess.run(
|
|
["git", "-C", path, "rev-parse", "--verify", "--quiet", f"{sha}^{{commit}}"],
|
|
capture_output=True,
|
|
text=True,
|
|
check=False,
|
|
)
|
|
return res.returncode == 0
|
|
|
|
def _is_ancestor(ancestor: str, descendant: str) -> bool | None:
|
|
res = subprocess.run(
|
|
["git", "-C", path, "merge-base", "--is-ancestor", ancestor, descendant],
|
|
capture_output=True,
|
|
text=True,
|
|
check=False,
|
|
)
|
|
if res.returncode == 0:
|
|
return True
|
|
if res.returncode == 1:
|
|
return False
|
|
return None # failed probe — never a silent "no"
|
|
|
|
try:
|
|
result["prior_present"] = _present(prior)
|
|
result["synced_present"] = _present(synced)
|
|
if not result["synced_present"] and path and os.path.isdir(path):
|
|
if target_remote:
|
|
subprocess.run(
|
|
["git", "-C", path, "fetch", target_remote, "--quiet"],
|
|
capture_output=True,
|
|
text=True,
|
|
check=False,
|
|
)
|
|
result["synced_present"] = _present(synced)
|
|
if not result["synced_present"]:
|
|
subprocess.run(
|
|
["git", "-C", path, "fetch", "--quiet"],
|
|
capture_output=True,
|
|
text=True,
|
|
check=False,
|
|
)
|
|
result["synced_present"] = _present(synced)
|
|
except OSError as exc: # git unavailable — fail closed, never assume
|
|
result["reasons"].append(f"merge-sync provenance probe could not run: {exc}")
|
|
return result
|
|
|
|
if not result["prior_present"]:
|
|
result["reasons"].append(
|
|
f"prior head {prior} is not reachable in '{path}'; history may have "
|
|
"been rewritten or force-moved"
|
|
)
|
|
if not result["synced_present"]:
|
|
result["reasons"].append(
|
|
f"synced head {synced} is not reachable in '{path}'"
|
|
)
|
|
if not (result["prior_present"] and result["synced_present"]):
|
|
return result
|
|
|
|
ancestor = _is_ancestor(prior, synced)
|
|
if ancestor is None:
|
|
result["reasons"].append(
|
|
"ancestry probe failed; merge-sync provenance unproven"
|
|
)
|
|
return result
|
|
result["prior_is_ancestor"] = bool(ancestor)
|
|
if not ancestor:
|
|
result["reasons"].append(
|
|
f"prior head {prior} is not an ancestor of synced head {synced}; "
|
|
"the branch history was not preserved (not a merge-based sync)"
|
|
)
|
|
return result
|
|
|
|
parents_res = subprocess.run(
|
|
["git", "-C", path, "rev-list", "--parents", "-n", "1", synced],
|
|
capture_output=True,
|
|
text=True,
|
|
check=False,
|
|
)
|
|
if parents_res.returncode != 0:
|
|
result["reasons"].append(
|
|
f"could not read parents of {synced}; merge-sync provenance unproven"
|
|
)
|
|
return result
|
|
tokens = (parents_res.stdout or "").split()
|
|
# tokens[0] is the commit itself; the rest are its parents.
|
|
parents = tokens[1:]
|
|
result["parent_count"] = len(parents)
|
|
result["synced_is_merge"] = len(parents) >= 2
|
|
if not result["synced_is_merge"]:
|
|
result["probe_ok"] = True
|
|
result["reasons"].append(
|
|
f"synced head {synced} has {len(parents)} parent(s); a merge-based "
|
|
"branch sync produces a merge commit (two or more parents)"
|
|
)
|
|
return result
|
|
first_parent = parents[0]
|
|
result["first_parent_sha"] = first_parent
|
|
|
|
fp_reaches = _is_ancestor(prior, first_parent) if prior != first_parent else True
|
|
if fp_reaches is None:
|
|
result["reasons"].append(
|
|
"first-parent ancestry probe failed; merge-sync provenance unproven"
|
|
)
|
|
return result
|
|
result["first_parent_reaches_prior"] = bool(fp_reaches)
|
|
result["probe_ok"] = True
|
|
if not fp_reaches:
|
|
result["reasons"].append(
|
|
f"merge first parent {first_parent} does not reach prior head "
|
|
f"{prior}; the branch mainline was re-authored (not a sanctioned sync)"
|
|
)
|
|
return result
|
|
|
|
result["is_merge_sync"] = True
|
|
result["proof"] = (
|
|
f"synced head {synced} is a merge commit (parents={len(parents)}) whose "
|
|
f"first-parent lineage reaches prior head {prior}; base merged into branch"
|
|
)
|
|
return result
|
|
|
|
|
|
def read_recorded_base(
|
|
worktree_path: str,
|
|
*,
|
|
head_sha: str | None,
|
|
extra_bases: tuple[str, ...] | list[str] = (),
|
|
base_branches: frozenset[str] | None = None,
|
|
) -> dict:
|
|
"""Observe the base commit an unpublished claim was branched from (#772).
|
|
|
|
A published claim records its base implicitly: the remote branch head is the
|
|
thing recovery measures against. An unpublished claim has no remote ref, so
|
|
the base must be observed here, server-side, as the merge-base between the
|
|
worktree HEAD and the base branch it was cut from.
|
|
|
|
Reports facts only; the disposition lives in ``issue_lock_recovery``. Every
|
|
field is read from git in the declared worktree — nothing is supplied by, or
|
|
reachable from, an MCP caller, so a caller cannot nominate a base that would
|
|
make unrelated history look like a descendant (#772 AC1/AC4).
|
|
|
|
A HEAD with no common ancestor in any base branch yields ``probe_ok`` with no
|
|
``base_sha``: unrelated history is reported as exactly that, never as a base.
|
|
"""
|
|
path = (worktree_path or "").strip()
|
|
head = (head_sha or "").strip()
|
|
bases = base_branches or BASE_BRANCHES
|
|
candidates = [*extra_bases, *sorted(bases)]
|
|
result: dict = {
|
|
"base_branch": None,
|
|
"base_sha": None,
|
|
"head_sha": head or None,
|
|
"probe_ok": False,
|
|
"candidates": candidates,
|
|
"reasons": [],
|
|
}
|
|
if not path or not head:
|
|
result["reasons"].append(
|
|
"recorded-base probe requires a worktree path and a HEAD sha"
|
|
)
|
|
return result
|
|
|
|
probed_any = False
|
|
for candidate in candidates:
|
|
name = (candidate or "").strip()
|
|
if not name:
|
|
continue
|
|
probe = subprocess.run(
|
|
["git", "-C", path, "merge-base", name, head],
|
|
capture_output=True,
|
|
text=True,
|
|
check=False,
|
|
)
|
|
if probe.returncode not in (0, 1):
|
|
# 0 = merge base found, 1 = no common ancestor. Anything else is a
|
|
# failed probe (missing ref, broken repo) — try the next candidate.
|
|
continue
|
|
probed_any = True
|
|
merge_base = (probe.stdout or "").strip()
|
|
if probe.returncode == 0 and merge_base:
|
|
result["base_branch"] = name
|
|
result["base_sha"] = merge_base
|
|
result["probe_ok"] = True
|
|
return result
|
|
|
|
result["probe_ok"] = probed_any
|
|
if probed_any:
|
|
result["reasons"].append(
|
|
f"HEAD {head} shares no common ancestor with any of "
|
|
f"{_base_list(bases)}; history is unrelated to this repository's base"
|
|
)
|
|
else:
|
|
result["reasons"].append(
|
|
f"recorded-base probe could not run against any of {_base_list(bases)} "
|
|
f"in '{path}'"
|
|
)
|
|
return result
|
|
|
|
|
|
def read_worktree_git_state(
|
|
worktree_path: str,
|
|
extra_bases: tuple[str, ...] | list[str] = (),
|
|
) -> dict:
|
|
"""Read branch name and porcelain status from a git worktree.
|
|
|
|
``extra_bases`` names additional branches (e.g. an approved stacked base)
|
|
that may anchor base-equivalence in addition to master/main/dev. When empty
|
|
(the default), only the normal base branches are considered.
|
|
"""
|
|
path = (worktree_path or "").strip()
|
|
if not path:
|
|
return {"current_branch": None, "porcelain_status": ""}
|
|
|
|
branch_res = subprocess.run(
|
|
["git", "-C", path, "branch", "--show-current"],
|
|
capture_output=True,
|
|
text=True,
|
|
check=False,
|
|
)
|
|
current_branch = (branch_res.stdout or "").strip() or None
|
|
|
|
status_res = subprocess.run(
|
|
["git", "-C", path, "status", "--porcelain"],
|
|
capture_output=True,
|
|
text=True,
|
|
check=False,
|
|
)
|
|
root_res = subprocess.run(
|
|
["git", "-C", path, "rev-parse", "--show-toplevel"],
|
|
capture_output=True,
|
|
text=True,
|
|
check=False,
|
|
)
|
|
head_res = subprocess.run(
|
|
["git", "-C", path, "rev-parse", "HEAD"],
|
|
capture_output=True,
|
|
text=True,
|
|
check=False,
|
|
)
|
|
head_sha = (head_res.stdout or "").strip() if head_res.returncode == 0 else None
|
|
base_branch, base_sha = _find_matching_base_ref(path, head_sha, extra_bases)
|
|
return {
|
|
"current_branch": current_branch,
|
|
"porcelain_status": status_res.stdout or "",
|
|
"inspected_git_root": (root_res.stdout or "").strip() if root_res.returncode == 0 else None,
|
|
"head_sha": head_sha,
|
|
"base_branch": base_branch,
|
|
"base_sha": base_sha,
|
|
"base_equivalent": bool(head_sha and base_sha and head_sha == base_sha),
|
|
}
|
|
|
|
|
|
def assess_issue_lock_worktree(
|
|
*,
|
|
worktree_path: str,
|
|
current_branch: str | None,
|
|
porcelain_status: str,
|
|
base_equivalent: bool | None = None,
|
|
inspected_git_root: str | None = None,
|
|
base_branch: str | None = None,
|
|
base_branches: frozenset[str] | None = None,
|
|
recovery_sanctioned: bool = False,
|
|
renewal_sanctioned: bool = False,
|
|
) -> dict:
|
|
"""Fail closed when lock preconditions are not met on the declared worktree.
|
|
|
|
``recovery_sanctioned`` is set only when ``issue_lock_recovery`` has already
|
|
proven, from the durable lock itself, that this is a dead-session recovery of
|
|
an existing claim (#753): same issue, branch, worktree, author, and head, with
|
|
the recording process dead. In that one case the base-equivalence requirement
|
|
is waived, because a branch that already carries the work is ahead of its base
|
|
by construction and could never satisfy it. Every other precondition —
|
|
notably worktree cleanliness — still applies unchanged, and brand-new issue
|
|
claims keep the full base-equivalence requirement.
|
|
|
|
``renewal_sanctioned`` waives base-equivalence on exactly the same grounds
|
|
for the other proven-ownership case (#760): ``issue_lock_renewal`` has shown
|
|
that an *expired* lease is being renewed by its exact recorded owner — same
|
|
remote, org, repo, issue, operation, branch, realpath-normalized worktree,
|
|
claimant username and profile — with the local head matching the remote head
|
|
and any owning PR head. Such a branch carries committed work for the same
|
|
reason a recovered one does, so it can never be base-equivalent either.
|
|
|
|
Both waivers relax this one requirement and nothing else. Neither is
|
|
caller-supplied: each is computed server-side from durable lock state plus
|
|
live observation. With both False every precondition applies exactly as
|
|
before.
|
|
"""
|
|
bases = base_branches or BASE_BRANCHES
|
|
reasons: list[str] = []
|
|
path = (worktree_path or "").strip()
|
|
if not path:
|
|
reasons.append("worktree path not declared for issue lock; fail closed")
|
|
return _assessment(False, reasons, path, None, [])
|
|
|
|
branch = (current_branch or "").strip()
|
|
dirty_files = parse_dirty_tracked_files(porcelain_status)
|
|
|
|
if dirty_files:
|
|
reasons.append(
|
|
"tracked file edits exist before issue lock; "
|
|
f"lock must precede implementation work in '{path}' "
|
|
f"(dirty files: {', '.join(dirty_files)})"
|
|
)
|
|
|
|
if recovery_sanctioned or renewal_sanctioned:
|
|
# Base-equivalence intentionally not evaluated: ownership was proven
|
|
# against the durable lock record instead — by dead-session recovery
|
|
# (#753) or by exact-owner renewal of an expired lease (#760). Every
|
|
# other precondition above and below still applies; cleanliness in
|
|
# particular is checked before this branch and is never waived.
|
|
pass
|
|
elif base_equivalent is False:
|
|
reasons.append(
|
|
"issue lock worktree must be base-equivalent to one of "
|
|
f"{_base_list(bases)} before implementation work; inspected "
|
|
f"branch '{branch or '(detached)'}' at '{path}'"
|
|
)
|
|
elif base_equivalent is None:
|
|
if not branch:
|
|
reasons.append(
|
|
"current branch unknown (detached HEAD?); issue lock base-equivalence "
|
|
f"to {_base_list(bases)} could not be proven"
|
|
)
|
|
elif branch not in bases:
|
|
reasons.append(
|
|
"issue lock worktree base-equivalence could not be proven; "
|
|
f"branch '{branch}' is not {_base_list(bases)}"
|
|
)
|
|
|
|
proven = not reasons
|
|
return _assessment(
|
|
proven,
|
|
reasons,
|
|
path,
|
|
branch or None,
|
|
dirty_files,
|
|
inspected_git_root=inspected_git_root,
|
|
base_branch=base_branch,
|
|
base_equivalent=base_equivalent,
|
|
recovery_sanctioned=recovery_sanctioned,
|
|
renewal_sanctioned=renewal_sanctioned,
|
|
)
|
|
|
|
|
|
def format_issue_lock_worktree_error(assessment: dict) -> str:
|
|
"""Format a single fail-closed error for ``gitea_lock_issue``."""
|
|
reasons = list(assessment.get("reasons") or [])
|
|
if not reasons:
|
|
reasons = ["issue lock worktree validation failed"]
|
|
return "; ".join(reasons) + " (fail closed)"
|
|
|
|
|
|
def verify_pr_worktree_matches_lock(
|
|
locked_worktree_path: str | None,
|
|
declared_worktree_path: str | None,
|
|
project_root: str,
|
|
) -> dict:
|
|
"""PR creation must use the same worktree the lock was validated against."""
|
|
locked = (locked_worktree_path or "").strip()
|
|
if not locked:
|
|
return {"proven": True, "block": False, "reasons": []}
|
|
|
|
declared = resolve_author_worktree_path(declared_worktree_path, project_root)
|
|
locked_real = os.path.realpath(locked)
|
|
declared_real = os.path.realpath(declared)
|
|
if locked_real != declared_real:
|
|
return {
|
|
"proven": False,
|
|
"block": True,
|
|
"reasons": [
|
|
f"PR worktree '{declared_real}' does not match locked worktree "
|
|
f"'{locked_real}' (fail closed)"
|
|
],
|
|
"locked_worktree_path": locked_real,
|
|
"declared_worktree_path": declared_real,
|
|
}
|
|
return {
|
|
"proven": True,
|
|
"block": False,
|
|
"reasons": [],
|
|
"locked_worktree_path": locked_real,
|
|
"declared_worktree_path": declared_real,
|
|
}
|
|
|
|
|
|
def _base_list(bases: frozenset[str]) -> str:
|
|
return "/".join(sorted(bases))
|
|
|
|
|
|
def _assessment(
|
|
proven: bool,
|
|
reasons: list[str],
|
|
worktree_path: str,
|
|
current_branch: str | None,
|
|
dirty_files: list[str],
|
|
*,
|
|
inspected_git_root: str | None = None,
|
|
base_branch: str | None = None,
|
|
base_equivalent: bool | None = None,
|
|
recovery_sanctioned: bool = False,
|
|
renewal_sanctioned: bool = False,
|
|
) -> dict:
|
|
return {
|
|
"proven": proven,
|
|
"block": not proven,
|
|
"reasons": reasons,
|
|
"worktree_path": worktree_path or None,
|
|
"inspected_git_root": inspected_git_root,
|
|
"current_branch": current_branch,
|
|
"dirty_files": dirty_files,
|
|
"base_branch": base_branch,
|
|
"base_equivalent": base_equivalent,
|
|
"recovery_sanctioned": recovery_sanctioned,
|
|
"renewal_sanctioned": renewal_sanctioned,
|
|
# Either proven-ownership waiver relaxes base-equivalence; the two are
|
|
# reported separately so an audit can tell which one applied.
|
|
"base_equivalence_waived": bool(recovery_sanctioned or renewal_sanctioned),
|
|
}
|
|
|
|
|
|
def _find_matching_base_ref(
|
|
path: str,
|
|
head_sha: str | None,
|
|
extra_bases: tuple[str, ...] | list[str] = (),
|
|
) -> tuple[str | None, str | None]:
|
|
"""Return the stable branch ref whose commit matches HEAD, if any.
|
|
|
|
Normal base branches (master/main/dev) are always considered. ``extra_bases``
|
|
adds explicitly-approved stacked bases; each is checked as a local ref and via
|
|
the ``prgs``/``origin`` remotes.
|
|
"""
|
|
if not head_sha:
|
|
return None, None
|
|
candidates: list[str] = []
|
|
for branch in sorted(BASE_BRANCHES):
|
|
candidates.extend((f"origin/{branch}", branch))
|
|
for branch in extra_bases:
|
|
name = (branch or "").strip()
|
|
if name:
|
|
candidates.extend((f"prgs/{name}", f"origin/{name}", name))
|
|
for ref in candidates:
|
|
res = subprocess.run(
|
|
["git", "-C", path, "rev-parse", "--verify", ref],
|
|
capture_output=True,
|
|
text=True,
|
|
check=False,
|
|
)
|
|
sha = (res.stdout or "").strip()
|
|
if res.returncode == 0 and sha == head_sha:
|
|
return ref, sha
|
|
return None, None
|