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

Merge pull request 'fix(mcp): recover clean unpublished author work after the owning session exits (Closes #772)' (#774) from fix/issue-772-unpublished-claim-recovery into master

Browse Source
This commit was merged in pull request #774.
This commit is contained in:
sysadmin
2026-07-20 15:52:07 -05:00
parent ccfaa0ec0c 6c15aa88b3
commit 702ceb2480
5 changed files with 1501 additions and 83 deletions
Download Patch File Download Diff File Expand all files Collapse all files
gitea_mcp_server.py
+187 -63
View File
@@ -2240,7 +2240,7 @@ def _resolve_issue_lock_for_pr(
return lock_data
def _save_issue_lock(data: dict) -> str:
def _save_issue_lock(data: dict, *, expected_generation: int | None = None) -> str:
existing = issue_lock_store.load_issue_lock(
remote=str(data.get("remote") or ""),
org=str(data.get("org") or ""),
@@ -2251,11 +2251,127 @@ def _save_issue_lock(data: dict) -> str:
if overwrite_block:
raise RuntimeError(overwrite_block)
try:
return issue_lock_store.bind_session_lock(data)
return issue_lock_store.bind_session_lock(
data, expected_generation=expected_generation
)
except Exception as e:
raise RuntimeError(f"Could not write issue lock file: {e}") from e
def _evaluate_issue_lock_recovery(
existing_lock: dict,
*,
issue_number: int,
branch_name: str,
worktree_path: str,
remote: str,
h: str | None,
o: str,
r: str,
git_state: dict,
) -> dict:
"""Gather recovery evidence and decide the disposition (#753/#768/#772).
The single decision point for dead-session issue-lock recovery. The mutating
``gitea_lock_issue`` path and the read-only diagnostic assessor both call
this, so the two cannot report different verdicts or different supporting
evidence for the same durable state (#772 AC9) — a divergence between them
would itself be the defect.
Every input is either durable lock state or a live server-side observation
(Gitea branch/PR inventory, git in the declared worktree). Nothing is
reachable from an MCP caller's parameters (#772 AC1).
"""
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}"
)
remote_head: str | None = None
remote_branch_exists = False
candidates: list[str] = []
for entry in recovery_branches:
entry_name = _branch_entry_name(entry)
if entry_name == branch_name:
remote_branch_exists = True
remote_head = _branch_entry_commit_sha(entry)
if issue_lock_adoption.branch_carries_issue_marker(entry_name, issue_number):
candidates.append(entry_name)
pr_head: str | None = None
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:
pr_head = pull_head.get("sha")
pr_number = pull.get("number")
break
local_head = git_state.get("head_sha")
# #768: the recovering author's clean worktree is, by construction, one
# commit ahead of the head recorded at lock time — committing is the only
# sanctioned way to clean it without discarding the work. Observe the
# ancestry here, server-side, so the assessor can tell an honest
# fast-forward from a rewritten head. Probed only when the heads actually
# differ, and never from any caller-supplied value.
head_ancestry: dict | None = None
if remote_head and local_head and remote_head != local_head:
head_ancestry = issue_lock_worktree.read_head_ancestry(
worktree_path,
ancestor_sha=remote_head,
descendant_sha=local_head,
)
# #772: with no remote branch there is no head to measure against, so the
# base the branch was cut from is observed instead. Probed only in that
# case, so the published path's evidence is untouched (#772 AC8).
recorded_base: str | None = None
base_ancestry: dict | None = None
if not remote_branch_exists and local_head:
base_observation = issue_lock_worktree.read_recorded_base(
worktree_path,
head_sha=local_head,
)
recorded_base = base_observation.get("base_sha")
if recorded_base:
base_ancestry = issue_lock_worktree.read_head_ancestry(
worktree_path,
ancestor_sha=recorded_base,
descendant_sha=local_head,
)
claimant = _work_lease_claimant(h)
return issue_lock_recovery.assess_dead_session_lock_recovery(
existing_lock,
issue_number=issue_number,
branch_name=branch_name,
worktree_path=worktree_path,
remote=remote,
org=o,
repo=r,
identity=claimant.get("username"),
profile=claimant.get("profile"),
current_branch=git_state.get("current_branch"),
porcelain_status=git_state.get("porcelain_status") or "",
head_sha=local_head,
remote_head_sha=remote_head,
pr_head_sha=pr_head,
pr_number=pr_number,
competing_live_locks=issue_lock_store.list_live_locks(),
candidate_branches=candidates,
current_pid=os.getpid(),
head_ancestry=head_ancestry,
remote_branch_exists=remote_branch_exists,
recorded_base_sha=recorded_base,
base_ancestry=base_ancestry,
)
def _work_lease_claimant(host: str | None) -> dict:
profile = get_profile()
username = _IDENTITY_CACHE.get(host) if host else None
@@ -3542,70 +3658,16 @@ def gitea_lock_issue(
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)
# #768: the recovering author's clean worktree is, by construction, one
# commit ahead of the head recorded at lock time — committing is the
# only sanctioned way to clean it without discarding the work. Observe
# the ancestry here, server-side, so the assessor can tell an honest
# fast-forward from a rewritten head. Probed only when the heads
# actually differ, and never from any caller-supplied value.
recovery_local_head = git_state.get("head_sha")
recovery_ancestry: dict | None = None
if (
recovery_remote_head
and recovery_local_head
and recovery_remote_head != recovery_local_head
):
recovery_ancestry = issue_lock_worktree.read_head_ancestry(
resolved_worktree,
ancestor_sha=recovery_remote_head,
descendant_sha=recovery_local_head,
)
recovery_assessment = issue_lock_recovery.assess_dead_session_lock_recovery(
recovery_assessment = _evaluate_issue_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(),
head_ancestry=recovery_ancestry,
h=h,
o=o,
r=r,
git_state=git_state,
)
recovery_sanctioned = bool(
@@ -3715,7 +3777,17 @@ def gitea_lock_issue(
recovered_at=_work_lease_timestamp(_work_lease_now()),
)
lock_file_path = _save_issue_lock(data)
# #772 AC5: a recovery replaces a claim another session already owned, so
# its write is a compare-and-swap against the generation the assessment was
# made on. Two replacement sessions that both observed the same dead owner
# cannot both succeed — the second finds a moved generation and fails
# closed. Ordinary first-time claims keep the unconditional write.
expected_generation = (
issue_lock_store.lock_generation(existing_issue_lock)
if recovery_sanctioned
else None
)
lock_file_path = _save_issue_lock(data, expected_generation=expected_generation)
lock_record = issue_lock_store.read_lock_file(lock_file_path) or data
freshness = issue_lock_store.assess_lock_freshness(lock_record)
competing = [
@@ -3832,7 +3904,59 @@ def gitea_assess_work_issue_duplicate(
phase=phase,
recovered_owning_pr=recovered_owning_pr,
)
return {"success": not gate.get("block"), **gate}
# #772 AC9: report the recovery disposition this issue's durable lock would
# receive, decided by the same evaluator the mutating lock path uses, so the
# diagnostic and the mutator can never disagree. Read-only: assessment only,
# never a write, and a probe failure degrades to a reported reason rather
# than turning a diagnostic into a hard error.
lock_recovery: dict | None = None
if (
lock_data
and int(lock_data.get("issue_number") or 0) == int(issue_number)
and not issue_lock_store.is_lease_live(lock_data)
):
recovery_worktree = str(lock_data.get("worktree_path") or "")
recovery_branch = str(
branch_name or lock_data.get("branch_name") or ""
)
if recovery_worktree and recovery_branch:
try:
recovery_state = issue_lock_worktree.read_worktree_git_state(
recovery_worktree
)
assessment = _evaluate_issue_lock_recovery(
lock_data,
issue_number=int(issue_number),
branch_name=recovery_branch,
worktree_path=recovery_worktree,
remote=remote,
h=h,
o=o,
r=r,
git_state=recovery_state,
)
lock_recovery = {
"outcome": assessment.get("outcome"),
"recovery_sanctioned": assessment.get("recovery_sanctioned"),
"is_candidate": assessment.get("is_candidate"),
"reasons": assessment.get("reasons"),
"evidence": assessment.get("evidence"),
}
except Exception as exc:
lock_recovery = {
"outcome": issue_lock_recovery.REFUSED,
"recovery_sanctioned": False,
"is_candidate": True,
"reasons": [
f"recovery evidence could not be gathered: {exc}"
],
"evidence": {},
}
return {
"success": not gate.get("block"),
**gate,
"lock_recovery": lock_recovery,
}
@mcp.tool()
issue_lock_recovery.py
+188 -18
View File
@@ -34,6 +34,33 @@ ancestor, still unmodified — so it is accepted, and nothing else is. The
descendant fact is observed server-side by
``issue_lock_worktree.read_head_ancestry`` and handed in as ``head_ancestry``;
no caller can assert it.
#772 adds the remaining uncovered quadrant: a claim that was never published at
all. Two recovery modes now exist, and they require different evidence because
they are answering the same question against different available facts:
``published_owning_pr``
The branch exists on the remote. Ownership is proven by comparing the local
head against the remote/PR head — equal (#753) or a strict descendant
(#768). This is the pre-existing behavior and is unchanged.
``unpublished_claim``
The branch is absent from the remote and no PR claims it, so there is no
head to compare against; that absence is the defining fact, not a degraded
published case. Ownership is instead proven by the durable lock record
(issue, branch, worktree, claimant, profile, dead PID) plus the local HEAD
strictly descending from the base the branch was cut from, observed
server-side by ``issue_lock_worktree.read_recorded_base`` and re-checked
through ``base_ancestry``.
They cannot share one head-comparison implementation: the published path's
comparison target does not exist in the unpublished case, and inventing one
(defaulting to the base, say) would silently weaken the published path from
"matches what was actually pushed" to "descends from some base". The modes are
therefore selected by observed publication state and never by a caller — and
critically, the absence of a remote head is never itself treated as permission:
every identity, profile, branch, worktree, cleanliness, liveness, and competing
-claim check still applies in full.
"""
from __future__ import annotations
@@ -55,6 +82,18 @@ REQUIRED_LOCK_FIELDS = ("issue_number", "branch_name", "worktree_path")
# How the clean local head relates to the head recorded at lock time (#768).
HEAD_RELATION_EQUAL = "equal"
HEAD_RELATION_STRICT_DESCENDANT = "strict_descendant"
# #772: an unpublished claim has no recorded head to compare against at all, so
# its head is measured against the base the branch was cut from instead.
HEAD_RELATION_DESCENDS_FROM_BASE = "descends_from_recorded_base"
# Which body of evidence a recovery was decided on (#772 AC10). These are not
# interchangeable: a published claim proves ownership against a remote/PR head,
# an unpublished one against the recorded base plus durable lock state. They
# cannot share a single head-comparison implementation because the unpublished
# case has no head to compare — that absence is the defining fact, not a
# degraded version of the published case.
RECOVERY_MODE_PUBLISHED_OWNING_PR = "published_owning_pr"
RECOVERY_MODE_UNPUBLISHED_CLAIM = "unpublished_claim"
def _same_realpath(left: str | None, right: str | None) -> bool:
@@ -165,6 +204,68 @@ def _assess_strict_descendant(
]
def _assess_base_descendancy(
base_ancestry: Mapping[str, Any] | None,
*,
recorded_base: str,
local_head: str,
) -> tuple[bool, list[str]]:
"""Is ``local_head`` a proven strict descendant of ``recorded_base`` (#772)?
The unpublished-claim analogue of ``_assess_strict_descendant``. The
comparison target is the base the branch was cut from — observed server-side
by ``issue_lock_worktree.read_recorded_base`` — rather than a remote or PR
head, because an unpublished claim has neither.
The probe's own endpoints are re-checked against the values under
assessment, so an observation taken for some other pair of commits cannot
authorize recovery. Equality is refused: a HEAD that merely equals its base
carries no committed work, and that is the ordinary base-equivalent case the
normal lock path already handles.
"""
if not isinstance(base_ancestry, Mapping):
return False, [
"no server-derived ancestry observation was available; an "
"unpublished claim cannot be recovered without proving its HEAD "
"descends from the recorded base"
]
probe_ancestor = _text(base_ancestry.get("ancestor_sha"))
probe_descendant = _text(base_ancestry.get("descendant_sha"))
if probe_ancestor != recorded_base or probe_descendant != local_head:
return False, [
f"ancestry observation covers {probe_ancestor or 'unknown'} -> "
f"{probe_descendant or 'unknown'}, not the commits under assessment "
f"({recorded_base} -> {local_head})"
]
if not base_ancestry.get("probe_ok"):
return False, (
list(base_ancestry.get("reasons") or [])
or ["ancestry probe did not complete; ancestry unproven"]
)
if not base_ancestry.get("ancestor_present"):
return False, [
f"recorded base {recorded_base} is no longer reachable; a rewritten "
"or force-moved base cannot be recovered"
]
if not base_ancestry.get("is_strict_descendant"):
return False, (
list(base_ancestry.get("reasons") or [])
or [
f"local head {local_head} is not a strict descendant of the "
f"recorded base {recorded_base}"
]
)
proof = _text(base_ancestry.get("proof")) or (
f"{recorded_base} is an ancestor of {local_head}"
)
return True, [
f"local head {local_head} strictly descends from recorded base "
f"{recorded_base} ({proof})"
]
def assess_dead_session_lock_recovery(
existing_lock: Mapping[str, Any] | None,
*,
@@ -186,6 +287,9 @@ def assess_dead_session_lock_recovery(
candidate_branches: Iterable[str] | None = None,
current_pid: int | None = None,
head_ancestry: Mapping[str, Any] | None = None,
remote_branch_exists: bool | None = None,
recorded_base_sha: str | None = None,
base_ancestry: Mapping[str, Any] | None = None,
) -> dict[str, Any]:
"""Decide whether a dead-session author lock may be natively recovered.
@@ -304,32 +408,73 @@ def assess_dead_session_lock_recovery(
# to reach the clean worktree recovery itself demands.
local_head = _text(head_sha)
remote_head = _text(remote_head_sha)
recorded_base = _text(recorded_base_sha)
head_relation: str | None = None
ancestry_proof: str | None = None
if not local_head:
reasons.append("local head SHA could not be determined")
if not remote_head:
reasons.append(
f"remote head for branch '{locked_branch}' could not be determined"
)
if local_head and remote_head:
if local_head == remote_head:
head_relation = HEAD_RELATION_EQUAL
else:
descends, notes = _assess_strict_descendant(
head_ancestry,
recorded_head=remote_head,
# #772: which body of evidence applies is decided by observed publication
# state, never by a caller. ``remote_branch_exists is False`` is a positive
# server-side observation that the branch is absent from the remote — it is
# not the same as "the head lookup failed", which must still fail closed.
unpublished = remote_branch_exists is False and not remote_head
recovery_mode = (
RECOVERY_MODE_UNPUBLISHED_CLAIM if unpublished
else RECOVERY_MODE_PUBLISHED_OWNING_PR
)
evidence["recovery_mode"] = recovery_mode
evidence["remote_branch_exists"] = remote_branch_exists
if unpublished:
# No remote branch: ownership is measured against the recorded base.
# An open PR here is contradictory — a PR cannot exist without a remote
# branch — so it is a mismatch, never a thing to reconcile.
if _text(pr_head_sha) or pr_number is not None:
reasons.append(
f"branch '{locked_branch}' is absent from the remote yet PR "
f"#{pr_number} claims it; publication state is contradictory"
)
if not recorded_base:
reasons.append(
f"recorded base for branch '{locked_branch}' could not be "
"determined; an unpublished claim cannot be recovered without it"
)
if local_head and recorded_base:
descends, notes = _assess_base_descendancy(
base_ancestry,
recorded_base=recorded_base,
local_head=local_head,
)
if descends:
head_relation = HEAD_RELATION_STRICT_DESCENDANT
head_relation = HEAD_RELATION_DESCENDS_FROM_BASE
ancestry_proof = notes[0] if notes else None
else:
reasons.append(
f"local head {local_head} does not match remote branch head "
f"{remote_head}"
)
reasons.extend(notes)
else:
if not remote_head:
reasons.append(
f"remote head for branch '{locked_branch}' could not be determined"
)
if local_head and remote_head:
if local_head == remote_head:
head_relation = HEAD_RELATION_EQUAL
else:
descends, notes = _assess_strict_descendant(
head_ancestry,
recorded_head=remote_head,
local_head=local_head,
)
if descends:
head_relation = HEAD_RELATION_STRICT_DESCENDANT
ancestry_proof = notes[0] if notes else None
else:
reasons.append(
f"local head {local_head} does not match remote branch head "
f"{remote_head}"
)
reasons.extend(notes)
evidence["recorded_base"] = recorded_base or None
evidence["local_head"] = local_head or None
evidence["remote_head"] = remote_head or None
# ``recorded_head`` is the head recovery is being measured against;
@@ -344,7 +489,9 @@ def assess_dead_session_lock_recovery(
if pr_head:
evidence["pr_head"] = pr_head
evidence["pr_number"] = pr_number
if local_head and pr_head != local_head:
# In unpublished mode the presence of any PR was already refused above as
# contradictory; re-stating it as a head mismatch would only obscure why.
if not unpublished and local_head and pr_head != local_head:
# A descendant recovery has not been published yet, so the open PR
# legitimately still points at the recorded head. Any other
# disagreement is a real mismatch.
@@ -449,12 +596,32 @@ def assess_dead_session_lock_recovery(
if reasons:
return _result(REFUSED, False, reasons, evidence)
# No disposition may be granted without a proven head relation. Every path
# above that leaves it unset also records a reason, so this is a belt-and-
# braces guard against a future path forgetting one (#772 AC4).
if head_relation is None:
return _result(
REFUSED,
False,
["head relation to the recorded head or base was never proven"],
evidence,
)
proof = [
f"durable lock for issue #{issue_number} matches branch "
f"'{locked_branch}', worktree '{locked_worktree}', head {local_head}, "
f"and claimant '{locked_identity}'; recorded pid {recorded_pid} is dead"
]
if head_relation == HEAD_RELATION_STRICT_DESCENDANT and ancestry_proof:
if recovery_mode == RECOVERY_MODE_UNPUBLISHED_CLAIM:
proof.append(
f"branch '{locked_branch}' has no remote head and no open PR; "
f"ownership proven against recorded base {recorded_base}"
)
if (
head_relation
in (HEAD_RELATION_STRICT_DESCENDANT, HEAD_RELATION_DESCENDS_FROM_BASE)
and ancestry_proof
):
proof.append(ancestry_proof)
return _result(RECOVERY_SANCTIONED, True, proof, evidence)
@@ -639,6 +806,9 @@ def build_recovery_record(
"prior_pid_alive": evidence.get("prior_pid_alive"),
"branch_name": evidence.get("locked_branch"),
"worktree_path": evidence.get("locked_worktree_path"),
"recovery_mode": evidence.get("recovery_mode"),
"remote_branch_exists": evidence.get("remote_branch_exists"),
"recorded_base": evidence.get("recorded_base"),
"local_head": evidence.get("local_head"),
"remote_head": evidence.get("remote_head"),
"recorded_head": evidence.get("recorded_head"),
issue_lock_store.py
+45 -2
View File
@@ -148,8 +148,37 @@ def save_lock_file(path: str, data: dict[str, Any]) -> None:
pass
def bind_session_lock(lock_data: dict[str, Any], lock_dir: str | None = None) -> str:
"""Persist a keyed lock and bind it to the current process session."""
def lock_generation(lock: dict[str, Any] | None) -> int:
"""Monotonic write counter for a durable lock record (#772 AC5).
Absent or unusable values read as ``0`` so a lock written before generations
existed still participates in compare-and-swap: its first recovery expects
``0`` and writes ``1``.
"""
if not isinstance(lock, dict):
return 0
try:
return int(lock.get("lock_generation") or 0)
except (TypeError, ValueError):
return 0
def bind_session_lock(
lock_data: dict[str, Any],
lock_dir: str | None = None,
*,
expected_generation: int | None = None,
) -> str:
"""Persist a keyed lock and bind it to the current process session.
``expected_generation`` turns the write into a compare-and-swap (#772 AC5).
Recovery decides it may take over a claim by reading the durable lock, but
that read and this write are separate steps; without a CAS two replacement
sessions can both observe the same dead owner, both pass assessment, and
both write — the second silently clobbering the first. Passing the
generation observed at assessment time makes exactly one of them win: the
loser's expectation no longer matches and it fails closed.
"""
remote = str(lock_data.get("remote") or "")
org = str(lock_data.get("org") or "")
repo = str(lock_data.get("repo") or "")
@@ -194,6 +223,20 @@ def bind_session_lock(lock_data: dict[str, Any], lock_dir: str | None = None) ->
)
if lease_block:
raise RuntimeError(lease_block)
# #772 AC5: compare-and-swap inside the same critical section that
# already serializes writers, so the check and the write cannot be
# separated by another session's successful recovery.
current_generation = lock_generation(existing)
if (
expected_generation is not None
and current_generation != expected_generation
):
raise RuntimeError(
f"Issue #{issue_number} lock generation changed: expected "
f"{expected_generation}, found {current_generation}; another "
"session already recovered or replaced this claim (fail closed)"
)
record["lock_generation"] = current_generation + 1
save_lock_file(path, record)
save_lock_file(session_pointer_path(root), pointer)
except LockContentionError as exc:
issue_lock_worktree.py
+77
View File
@@ -145,6 +145,83 @@ def read_head_ancestry(
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] = (),
tests/test_issue_772_unpublished_claim_recovery.py
+1004
View File
File diff suppressed because it is too large Load Diff