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

fix(mcp): accept strict-descendant dead-session recovery (Closes #768)

Browse Source 
Permit fail-closed recovery when a clean local head is a strict
descendant of the recorded PR/remote head, with server-side ancestry
proof. Propagate recovery evidence through commit, push, and PR
duplicate gates so an owning PR is not re-blocked as competing work.

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
This commit is contained in:
jcwalker3
2026-07-20 02:52:29 -05:00
co-authored by Claude Opus 4.8
parent d12adabeb1
commit 5547399037
6 changed files with 971 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_issue_755_owning_pr_recovery.py
+10 -1
View File
@@ -73,12 +73,21 @@ def owning_pr(number=OWNING_PR, ref=BRANCH, sha=HEAD, issue=ISSUE):
def sanctioned_token(
issue_number=ISSUE, pr_number=OWNING_PR, branch=BRANCH, head=HEAD
):
"""The evidence shape the server derives from a granted recovery."""
"""The evidence shape the server derives from a granted recovery.
#768 extends the token with recorded/accepted heads and the head relation
so a strict-descendant recovery can still exempt the owning PR after the
remediation commit lands. Exact-head recovery (#753/#755) reports equal
heads under the same shape.
"""
return {
"issue_number": issue_number,
"pr_number": pr_number,
"branch_name": branch,
"head_sha": head,
"recorded_head": head,
"accepted_head": head,
"head_relation": issue_lock_recovery.HEAD_RELATION_EQUAL,
}
tests/test_issue_768_strict_descendant_recovery.py
+551
View File
@@ -0,0 +1,551 @@
"""Strict-descendant dead-session recovery (#768).
After a dead author session, a preserved clean remediation commit that strictly
descends from the head recorded at lock time must be recoverable so the author
can publish. Equality alone is still accepted (#753); every other divergence
must keep failing closed.
"""
from __future__ import annotations
import os
import subprocess
import sys
import tempfile
import unittest
from datetime import datetime, timedelta, timezone
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
import issue_lock_recovery # noqa: E402
import issue_lock_store # noqa: E402
import issue_lock_worktree # noqa: E402
import issue_work_duplicate_gate # noqa: E402
ISSUE = 7680
PR_NUMBER = 7681
BRANCH = f"fix/issue-{ISSUE}-descendant-recovery"
WORKTREE = "/scratch/wt-768"
RECORDED = "a" * 40
DESCENDANT = "c" * 40
DIVERGED = "d" * 40
BEHIND = "b" * 40
IDENTITY = "example-user"
PROFILE = "example-author"
def dead_pid() -> int:
proc = subprocess.Popen([sys.executable, "-c", "pass"])
proc.wait()
return proc.pid
def future_ts(hours: int = 4) -> str:
return (
(datetime.now(timezone.utc) + timedelta(hours=hours))
.isoformat()
.replace("+00:00", "Z")
)
def make_lock(**overrides):
lock = {
"issue_number": ISSUE,
"branch_name": BRANCH,
"worktree_path": WORKTREE,
"remote": "prgs",
"org": "ExampleOrg",
"repo": "ExampleRepo",
"session_pid": dead_pid(),
"work_lease": {
"operation_type": issue_lock_store.AUTHOR_ISSUE_WORK_LEASE,
"issue_number": ISSUE,
"branch": BRANCH,
"worktree_path": WORKTREE,
"claimant": {"username": IDENTITY, "profile": PROFILE},
"expires_at": future_ts(),
},
}
lock.update(overrides)
return lock
def ancestry_ok(
*,
ancestor: str = RECORDED,
descendant: str = DESCENDANT,
is_strict: bool = True,
probe_ok: bool = True,
ancestor_present: bool = True,
reasons: list[str] | None = None,
) -> dict:
return {
"ancestor_sha": ancestor,
"descendant_sha": descendant,
"probe_ok": probe_ok,
"ancestor_present": ancestor_present,
"descendant_present": True,
"is_ancestor": is_strict or ancestor == descendant,
"is_strict_descendant": is_strict,
"proof": f"git merge-base --is-ancestor {ancestor} {descendant} -> exit 0",
"reasons": list(reasons or []),
}
def assess(**overrides):
kwargs = {
"issue_number": ISSUE,
"branch_name": BRANCH,
"worktree_path": WORKTREE,
"remote": "prgs",
"org": "ExampleOrg",
"repo": "ExampleRepo",
"identity": IDENTITY,
"profile": PROFILE,
"current_branch": BRANCH,
"porcelain_status": "",
"head_sha": RECORDED,
"remote_head_sha": RECORDED,
"pr_head_sha": RECORDED,
"pr_number": PR_NUMBER,
"competing_live_locks": [],
"candidate_branches": [BRANCH],
"current_pid": os.getpid(),
"head_ancestry": None,
}
kwargs.update(overrides)
lock = kwargs.pop("lock", None)
return issue_lock_recovery.assess_dead_session_lock_recovery(
make_lock() if lock is None else lock, **kwargs
)
class TestExactHeadRecoveryStillSucceeds(unittest.TestCase):
def test_equal_heads_still_sanctioned(self):
result = assess()
self.assertTrue(result["recovery_sanctioned"], result["reasons"])
self.assertEqual(
result["evidence"]["head_relation"],
issue_lock_recovery.HEAD_RELATION_EQUAL,
)
self.assertEqual(result["evidence"]["recorded_head"], RECORDED)
self.assertEqual(result["evidence"]["accepted_head"], RECORDED)
def test_exact_match_record_carries_relation(self):
record = issue_lock_recovery.build_recovery_record(
assess(), recovered_at="2026-07-20T00:00:00Z"
)
self.assertEqual(record["head_relation"], issue_lock_recovery.HEAD_RELATION_EQUAL)
self.assertEqual(record["recorded_head"], RECORDED)
self.assertEqual(record["accepted_head"], RECORDED)
class TestStrictDescendantRecoverySucceeds(unittest.TestCase):
def test_clean_strict_descendant_recovers(self):
result = assess(
head_sha=DESCENDANT,
remote_head_sha=RECORDED,
pr_head_sha=RECORDED,
head_ancestry=ancestry_ok(),
)
self.assertTrue(result["recovery_sanctioned"], result["reasons"])
self.assertEqual(
result["evidence"]["head_relation"],
issue_lock_recovery.HEAD_RELATION_STRICT_DESCENDANT,
)
self.assertEqual(result["evidence"]["recorded_head"], RECORDED)
self.assertEqual(result["evidence"]["accepted_head"], DESCENDANT)
self.assertIsNotNone(result["evidence"]["ancestry_proof"])
self.assertTrue(
any("strictly descends" in r for r in result["reasons"]),
result["reasons"],
)
def test_pr_still_at_recorded_head_is_ok_for_descendant(self):
# Remediation is local only; open PR still points at the recorded head.
result = assess(
head_sha=DESCENDANT,
remote_head_sha=RECORDED,
pr_head_sha=RECORDED,
head_ancestry=ancestry_ok(),
)
self.assertTrue(result["recovery_sanctioned"], result["reasons"])
def test_recovery_record_names_both_heads_and_proof(self):
result = assess(
head_sha=DESCENDANT,
remote_head_sha=RECORDED,
pr_head_sha=RECORDED,
head_ancestry=ancestry_ok(),
)
record = issue_lock_recovery.build_recovery_record(
result, recovered_at="2026-07-20T00:00:00Z"
)
self.assertEqual(record["recorded_head"], RECORDED)
self.assertEqual(record["accepted_head"], DESCENDANT)
self.assertEqual(
record["head_relation"],
issue_lock_recovery.HEAD_RELATION_STRICT_DESCENDANT,
)
self.assertIn("strictly descends", record["ancestry_proof"] or "")
self.assertEqual(record["prior_session_pid"], result["evidence"]["prior_session_pid"])
self.assertEqual(record["replacement_session_pid"], os.getpid())
class TestDescendantEvidenceReachesPublicationGates(unittest.TestCase):
def _descendant_assessment(self):
return assess(
head_sha=DESCENDANT,
remote_head_sha=RECORDED,
pr_head_sha=RECORDED,
head_ancestry=ancestry_ok(),
)
def test_owning_pr_evidence_carries_accepted_head(self):
token = issue_lock_recovery.owning_pr_recovery_evidence(
self._descendant_assessment()
)
self.assertIsNotNone(token)
assert token is not None
self.assertEqual(token["head_sha"], RECORDED)
self.assertEqual(token["accepted_head"], DESCENDANT)
self.assertEqual(token["recorded_head"], RECORDED)
self.assertEqual(
token["head_relation"],
issue_lock_recovery.HEAD_RELATION_STRICT_DESCENDANT,
)
def test_persisted_lock_rebuilds_owning_pr_evidence(self):
assessment = self._descendant_assessment()
record = issue_lock_recovery.build_recovery_record(
assessment, recovered_at="2026-07-20T00:00:00Z"
)
lock = make_lock(dead_session_recovery=record)
token = issue_lock_recovery.recovered_owning_pr_from_lock(lock)
self.assertIsNotNone(token)
assert token is not None
self.assertEqual(token["pr_number"], PR_NUMBER)
self.assertEqual(token["head_sha"], RECORDED)
self.assertEqual(token["accepted_head"], DESCENDANT)
def test_duplicate_gate_accepts_pr_at_recorded_or_accepted_head(self):
token = issue_lock_recovery.owning_pr_recovery_evidence(
self._descendant_assessment()
)
for live_sha in (RECORDED, DESCENDANT):
with self.subTest(live_sha=live_sha):
gate = issue_work_duplicate_gate.assess_work_issue_duplicate_gate(
ISSUE,
open_prs=[
{
"number": PR_NUMBER,
"title": f"Closes #{ISSUE}",
"body": f"Closes #{ISSUE}",
"head": {"ref": BRANCH, "sha": live_sha},
}
],
branch_names=[BRANCH],
claim_entry={"status": "unclaimed"},
locked_branch=BRANCH,
phase=issue_work_duplicate_gate.PHASE_COMMIT,
recovered_owning_pr=token,
)
self.assertFalse(gate["block"], gate)
self.assertTrue(gate["owning_pr_recovery_exempted"])
def test_duplicate_gate_still_rejects_foreign_head(self):
token = issue_lock_recovery.owning_pr_recovery_evidence(
self._descendant_assessment()
)
gate = issue_work_duplicate_gate.assess_work_issue_duplicate_gate(
ISSUE,
open_prs=[
{
"number": PR_NUMBER,
"title": f"Closes #{ISSUE}",
"body": f"Closes #{ISSUE}",
"head": {"ref": BRANCH, "sha": DIVERGED},
}
],
branch_names=[BRANCH],
claim_entry={"status": "unclaimed"},
locked_branch=BRANCH,
phase=issue_work_duplicate_gate.PHASE_COMMIT,
recovered_owning_pr=token,
)
self.assertTrue(gate["block"])
self.assertFalse(gate["owning_pr_recovery_exempted"])
class TestDirtyDescendantRejected(unittest.TestCase):
def test_dirty_descendant_refused(self):
result = assess(
head_sha=DESCENDANT,
remote_head_sha=RECORDED,
pr_head_sha=RECORDED,
head_ancestry=ancestry_ok(),
porcelain_status=" M issue_lock_recovery.py\n",
)
self.assertFalse(result["recovery_sanctioned"])
self.assertTrue(any("dirty" in r.lower() for r in result["reasons"]))
class TestDivergedAndBehindRejected(unittest.TestCase):
def test_diverged_head_refused(self):
result = assess(
head_sha=DIVERGED,
remote_head_sha=RECORDED,
pr_head_sha=RECORDED,
head_ancestry=ancestry_ok(
ancestor=RECORDED,
descendant=DIVERGED,
is_strict=False,
reasons=[
f"local head {DIVERGED} does not descend from recorded head "
f"{RECORDED}"
],
),
)
self.assertFalse(result["recovery_sanctioned"])
self.assertIsNone(result["evidence"].get("head_relation"))
self.assertTrue(
any("does not match remote" in r for r in result["reasons"]),
result["reasons"],
)
def test_local_behind_recorded_refused(self):
# merge-base --is-ancestor RECORDED BEHIND is false when BEHIND is ancestor.
result = assess(
head_sha=BEHIND,
remote_head_sha=RECORDED,
pr_head_sha=RECORDED,
head_ancestry=ancestry_ok(
ancestor=RECORDED,
descendant=BEHIND,
is_strict=False,
reasons=[
f"local head {BEHIND} does not descend from recorded head "
f"{RECORDED}"
],
),
)
self.assertFalse(result["recovery_sanctioned"])
self.assertTrue(
any("does not match remote" in r or "not a strict descendant" in r
for r in result["reasons"]),
result["reasons"],
)
class TestUnrelatedAndMalformedAncestryRejected(unittest.TestCase):
def test_missing_ancestry_observation_fails_closed(self):
result = assess(
head_sha=DESCENDANT,
remote_head_sha=RECORDED,
pr_head_sha=RECORDED,
head_ancestry=None,
)
self.assertFalse(result["recovery_sanctioned"])
self.assertTrue(
any("ancestry" in r.lower() for r in result["reasons"]),
result["reasons"],
)
def test_mismatched_probe_pair_fails_closed(self):
# Observation for a different commit pair must not authorize this pair.
result = assess(
head_sha=DESCENDANT,
remote_head_sha=RECORDED,
pr_head_sha=RECORDED,
head_ancestry=ancestry_ok(ancestor=DIVERGED, descendant=DESCENDANT),
)
self.assertFalse(result["recovery_sanctioned"])
self.assertTrue(
any("not the heads under assessment" in r for r in result["reasons"]),
result["reasons"],
)
def test_rewritten_recorded_head_fails_closed(self):
result = assess(
head_sha=DESCENDANT,
remote_head_sha=RECORDED,
pr_head_sha=RECORDED,
head_ancestry=ancestry_ok(ancestor_present=False, is_strict=False),
)
self.assertFalse(result["recovery_sanctioned"])
self.assertTrue(
any("no longer reachable" in r or "rewritten" in r
for r in result["reasons"]),
result["reasons"],
)
def test_failed_probe_fails_closed(self):
result = assess(
head_sha=DESCENDANT,
remote_head_sha=RECORDED,
pr_head_sha=RECORDED,
head_ancestry=ancestry_ok(
probe_ok=False,
is_strict=False,
reasons=["ancestry probe failed with exit 128; ancestry unproven"],
),
)
self.assertFalse(result["recovery_sanctioned"])
def test_pr_head_not_equal_to_recorded_blocks_descendant(self):
result = assess(
head_sha=DESCENDANT,
remote_head_sha=RECORDED,
pr_head_sha=DIVERGED,
head_ancestry=ancestry_ok(),
)
self.assertFalse(result["recovery_sanctioned"])
self.assertTrue(
any("open PR" in r and "does not match" in r for r in result["reasons"]),
result["reasons"],
)
class TestLiveOwnerStillRejected(unittest.TestCase):
def test_live_prior_pid_refused_even_with_descendant_proof(self):
result = assess(
lock=make_lock(session_pid=os.getpid()),
head_sha=DESCENDANT,
remote_head_sha=RECORDED,
pr_head_sha=RECORDED,
head_ancestry=ancestry_ok(),
)
self.assertFalse(result["recovery_sanctioned"])
self.assertTrue(
any("still alive" in r or "live" in r.lower() for r in result["reasons"]),
result["reasons"],
)
class TestDiagnosticsIdentifyDisposition(unittest.TestCase):
def test_equal_disposition_named(self):
result = assess()
self.assertEqual(
result["evidence"]["head_relation"],
issue_lock_recovery.HEAD_RELATION_EQUAL,
)
def test_descendant_disposition_named(self):
result = assess(
head_sha=DESCENDANT,
remote_head_sha=RECORDED,
pr_head_sha=RECORDED,
head_ancestry=ancestry_ok(),
)
self.assertEqual(
result["evidence"]["head_relation"],
issue_lock_recovery.HEAD_RELATION_STRICT_DESCENDANT,
)
def test_rejected_divergence_has_no_accepted_relation(self):
result = assess(
head_sha=DIVERGED,
remote_head_sha=RECORDED,
head_ancestry=None,
)
self.assertIsNone(result["evidence"].get("head_relation"))
message = issue_lock_recovery.format_recovery_refusal(result)
self.assertIn("fail closed", message)
self.assertIn("does not match remote", message)
class TestReadHeadAncestryRealGit(unittest.TestCase):
def _git(self, repo: str, *args: str) -> subprocess.CompletedProcess[str]:
return subprocess.run(
["git", "-C", repo, *args],
capture_output=True,
text=True,
check=True,
)
def _init_repo_with_chain(self) -> tuple[str, str, str, str]:
"""Return (repo, parent_sha, child_sha, sibling_sha)."""
repo = tempfile.mkdtemp(prefix="issue-768-ancestry-")
self._git(repo, "init")
self._git(repo, "config", "user.email", "[email protected]")
self._git(repo, "config", "user.name", "Test")
path = Path(repo) / "f.txt"
path.write_text("one\n")
self._git(repo, "add", "f.txt")
self._git(repo, "commit", "-m", "parent")
parent = self._git(repo, "rev-parse", "HEAD").stdout.strip()
path.write_text("two\n")
self._git(repo, "add", "f.txt")
self._git(repo, "commit", "-m", "child")
child = self._git(repo, "rev-parse", "HEAD").stdout.strip()
# Divergent sibling: branch from parent, then unique commit.
self._git(repo, "checkout", "-B", "side", parent)
path.write_text("side\n")
self._git(repo, "add", "f.txt")
self._git(repo, "commit", "-m", "sibling")
sibling = self._git(repo, "rev-parse", "HEAD").stdout.strip()
self._git(repo, "checkout", "-B", "main", child)
return repo, parent, child, sibling
def test_strict_descendant_observation(self):
repo, parent, child, _sibling = self._init_repo_with_chain()
obs = issue_lock_worktree.read_head_ancestry(
repo, ancestor_sha=parent, descendant_sha=child
)
self.assertTrue(obs["probe_ok"])
self.assertTrue(obs["ancestor_present"])
self.assertTrue(obs["is_ancestor"])
self.assertTrue(obs["is_strict_descendant"])
self.assertEqual(obs["ancestor_sha"], parent)
self.assertEqual(obs["descendant_sha"], child)
def test_equal_heads_not_strict_descendant(self):
repo, parent, _child, _sibling = self._init_repo_with_chain()
obs = issue_lock_worktree.read_head_ancestry(
repo, ancestor_sha=parent, descendant_sha=parent
)
self.assertTrue(obs["probe_ok"])
self.assertTrue(obs["is_ancestor"])
self.assertFalse(obs["is_strict_descendant"])
def test_diverged_not_ancestor(self):
repo, _parent, child, sibling = self._init_repo_with_chain()
# child and sibling share a parent but neither descends from the other.
obs = issue_lock_worktree.read_head_ancestry(
repo, ancestor_sha=child, descendant_sha=sibling
)
self.assertTrue(obs["probe_ok"])
self.assertFalse(obs["is_ancestor"])
self.assertFalse(obs["is_strict_descendant"])
def test_missing_sha_fails_closed(self):
repo, _parent, child, _ = self._init_repo_with_chain()
obs = issue_lock_worktree.read_head_ancestry(
repo, ancestor_sha="0" * 40, descendant_sha=child
)
self.assertFalse(obs["probe_ok"])
self.assertFalse(obs["ancestor_present"])
def test_end_to_end_real_git_descendant_recovery(self):
repo, parent, child, _sibling = self._init_repo_with_chain()
obs = issue_lock_worktree.read_head_ancestry(
repo, ancestor_sha=parent, descendant_sha=child
)
result = assess(
worktree_path=repo,
head_sha=child,
remote_head_sha=parent,
pr_head_sha=parent,
head_ancestry=obs,
lock=make_lock(worktree_path=repo),
)
self.assertTrue(result["recovery_sanctioned"], result["reasons"])
self.assertEqual(
result["evidence"]["head_relation"],
issue_lock_recovery.HEAD_RELATION_STRICT_DESCENDANT,
)
if __name__ == "__main__":
unittest.main()