jcwalker3andClaude Opus 4.8
5547399037
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]>
552 lines
20 KiB
Python
552 lines
20 KiB
Python
"""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()
|