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

Merge pull request 'fix(mcp): allow author-lock recovery after the owning session exits (Closes #753)' (#754) from fix/issue-753-dead-pid-author-lock-recovery into master

Browse Source
This commit was merged in pull request #754.
This commit is contained in:
sysadmin
2026-07-18 20:03:26 -05:00
parent 0425bf9a43 3edeba4d7f
commit 08f67007c5
5 changed files with 892 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files
gitea_mcp_server.py
+98 -2
View File
@@ -1650,6 +1650,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
@@ -3155,8 +3156,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,
@@ -3198,6 +3202,72 @@ 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")
)
lock_assessment = issue_lock_worktree.assess_issue_lock_worktree(
worktree_path=resolved_worktree,
current_branch=git_state.get("current_branch"),
@@ -3205,10 +3275,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)
@@ -3271,6 +3351,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
@@ -3304,6 +3392,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"] = (
issue_lock_adoption.py
+9
View File
@@ -85,6 +85,15 @@ def _branch_carries_issue_marker(branch_name: str, issue_number: int) -> bool:
return re.search(pattern, name) is not None
def branch_carries_issue_marker(branch_name: str, issue_number: int) -> bool:
"""Public accessor for the exact issue-marker match (#753).
Dead-session lock recovery needs the same word-boundary matcher to detect
ambiguous branch claims, so it is exposed rather than reached into.
"""
return _branch_carries_issue_marker(branch_name, issue_number)
def assess_own_branch_adoption(
*,
issue_number: int,
issue_lock_recovery.py
+400
View File
@@ -0,0 +1,400 @@
"""Dead-session author issue-lock recovery (#753).
A durable author issue lock records the PID of the MCP session that took it.
When that process exits, ``issue_lock_store.assess_lock_freshness`` classifies
the lock as ``stale`` (``live=False``) even while its lease is still within TTL,
so every ownership check that requires a *live* lock fails closed.
Re-taking the lock through ``gitea_lock_issue`` is unreachable for real work:
``issue_lock_worktree.assess_issue_lock_worktree`` demands the worktree be
base-equivalent to ``master``/``main``/``dev``, and a branch that already
carries commits is ahead of its base by construction. The existing
``assess_expired_lock_reclaim`` affordance does not apply either, because
``assess_same_issue_lease_conflict`` only consults it once the lease has
*expired* — a dead PID under an unexpired lease never reaches it.
This module is the pure evidence assessor for that one narrow case. It grants
recovery only when every element of durable ownership still matches exactly and
the recorded process is demonstrably dead. It never trusts caller assertions:
every field is compared against durable lock state or live observation supplied
by the caller. It performs no mutation and no network I/O.
Recovery deliberately does **not** relax base-equivalence for brand-new issue
claims — only for a lock whose own prior record already proves the branch,
worktree, head, and author.
"""
from __future__ import annotations
import os
from typing import Any, Iterable, Mapping, Sequence
from issue_lock_store import is_process_alive
from reviewer_worktree import parse_dirty_tracked_files
# Outcome values
RECOVERY_SANCTIONED = "RECOVERY_SANCTIONED"
NO_CANDIDATE = "NO_CANDIDATE"
REFUSED = "REFUSED"
# Durable fields a lock must carry before it can be considered at all.
REQUIRED_LOCK_FIELDS = ("issue_number", "branch_name", "worktree_path")
def _same_realpath(left: str | None, right: str | None) -> bool:
if not left or not right:
return False
try:
return os.path.realpath(left) == os.path.realpath(right)
except OSError:
return left == right
def _text(value: Any) -> str:
return str(value or "").strip()
def _lock_claimant(lock: Mapping[str, Any]) -> dict[str, Any]:
claimant = lock.get("claimant")
if not isinstance(claimant, Mapping):
lease = lock.get("work_lease")
claimant = lease.get("claimant") if isinstance(lease, Mapping) else None
return dict(claimant) if isinstance(claimant, Mapping) else {}
def _recorded_pid(lock: Mapping[str, Any]) -> Any:
pid = lock.get("session_pid")
if pid is None:
pid = lock.get("pid")
return pid
def _malformed_reasons(lock: Mapping[str, Any]) -> list[str]:
"""Names of durable fields that are missing or unusable."""
missing: list[str] = []
for field in REQUIRED_LOCK_FIELDS:
if not _text(lock.get(field)):
missing.append(field)
pid = _recorded_pid(lock)
if pid is None or _text(pid) == "":
missing.append("session_pid/pid")
else:
try:
if int(pid) <= 0:
missing.append("session_pid/pid")
except (TypeError, ValueError):
missing.append("session_pid/pid")
return missing
def assess_dead_session_lock_recovery(
existing_lock: Mapping[str, Any] | None,
*,
issue_number: int,
branch_name: str,
worktree_path: str,
remote: str,
org: str,
repo: str,
identity: str | None,
profile: str | None,
current_branch: str | None,
porcelain_status: str,
head_sha: str | None,
remote_head_sha: str | None,
pr_head_sha: str | None = None,
pr_number: int | None = None,
competing_live_locks: Sequence[Mapping[str, Any]] | None = None,
candidate_branches: Iterable[str] | None = None,
current_pid: int | None = None,
) -> dict[str, Any]:
"""Decide whether a dead-session author lock may be natively recovered.
Returns a dict with ``recovery_sanctioned`` (bool), ``outcome``, ``reasons``
(why it was refused, or the positive proof when sanctioned), and
``evidence`` (a redaction-safe record for auditing).
``NO_CANDIDATE`` means no recovery was attempted at all — there is no
existing lock, or the lock does not describe this issue. The caller must
treat that exactly as it treated the pre-#753 world. ``REFUSED`` means a
candidate existed but the evidence did not agree; the caller fails closed.
"""
reasons: list[str] = []
evidence: dict[str, Any] = {
"issue_number": issue_number,
"branch_name": branch_name,
"worktree_path": worktree_path,
"remote": remote,
"org": org,
"repo": repo,
}
if not existing_lock:
return _result(
NO_CANDIDATE, False, ["no existing durable lock for this issue"], evidence
)
lock = dict(existing_lock)
# ── Candidate identification ────────────────────────────────────────────
# Recovery only ever applies to a lock that already claims THIS issue.
# Anything else is not a recovery candidate and must not be reinterpreted.
if lock.get("issue_number") != issue_number:
return _result(
NO_CANDIDATE,
False,
[
f"existing lock targets issue #{lock.get('issue_number')}, "
f"not #{issue_number}; not a recovery candidate"
],
evidence,
)
# A malformed/incomplete durable record can never prove ownership.
missing = _malformed_reasons(lock)
if missing:
return _result(
REFUSED,
False,
[
"durable lock record is incomplete and cannot prove ownership "
f"(missing/unusable: {', '.join(missing)})"
],
evidence,
)
recorded_pid = _recorded_pid(lock)
evidence["prior_session_pid"] = recorded_pid
evidence["replacement_session_pid"] = (
current_pid if current_pid is not None else os.getpid()
)
# ── Repository scope ────────────────────────────────────────────────────
for field, expected in (("remote", remote), ("org", org), ("repo", repo)):
actual = _text(lock.get(field))
if actual != _text(expected):
reasons.append(
f"lock {field} '{actual}' does not match requested '{_text(expected)}'"
)
# ── Branch identity ─────────────────────────────────────────────────────
locked_branch = _text(lock.get("branch_name"))
if locked_branch != _text(branch_name):
reasons.append(
f"lock branch '{locked_branch}' does not match requested "
f"'{_text(branch_name)}'"
)
evidence["locked_branch"] = locked_branch
# The worktree must actually be sitting on the locked branch. Without this
# a clean worktree parked elsewhere could stand in for the real work.
checked_out = _text(current_branch)
if not checked_out:
reasons.append(
"worktree is not on a named branch (detached HEAD); locked-branch "
"occupancy could not be proven"
)
elif checked_out != locked_branch:
reasons.append(
f"worktree is on branch '{checked_out}', not the locked branch "
f"'{locked_branch}'"
)
# ── Worktree identity ───────────────────────────────────────────────────
locked_worktree = _text(lock.get("worktree_path"))
if not _same_realpath(locked_worktree, worktree_path):
reasons.append(
f"lock worktree '{locked_worktree}' does not match declared "
f"'{_text(worktree_path)}'"
)
evidence["locked_worktree_path"] = locked_worktree
# ── Cleanliness (never waived) ──────────────────────────────────────────
dirty_files = parse_dirty_tracked_files(porcelain_status)
if dirty_files:
reasons.append(
"worktree has tracked local edits; recovery requires a clean "
f"worktree (dirty files: {', '.join(dirty_files)})"
)
evidence["dirty_files"] = dirty_files
# ── Head agreement: local == remote == PR ───────────────────────────────
local_head = _text(head_sha)
remote_head = _text(remote_head_sha)
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 and local_head != remote_head:
reasons.append(
f"local head {local_head} does not match remote branch head {remote_head}"
)
evidence["local_head"] = local_head or None
evidence["remote_head"] = remote_head or None
pr_head = _text(pr_head_sha)
if pr_head:
evidence["pr_head"] = pr_head
evidence["pr_number"] = pr_number
if local_head and pr_head != local_head:
reasons.append(
f"open PR #{pr_number} head {pr_head} does not match local head "
f"{local_head}"
)
# ── Author identity ─────────────────────────────────────────────────────
claimant = _lock_claimant(lock)
locked_identity = _text(claimant.get("username"))
locked_profile = _text(claimant.get("profile"))
evidence["locked_identity"] = locked_identity or None
evidence["locked_profile"] = locked_profile or None
if not locked_identity or not locked_profile:
reasons.append(
"durable lock does not record a claimant identity/profile; "
"author ownership could not be proven"
)
if not _text(identity) or not _text(profile):
reasons.append(
"active session identity/profile is unknown; author ownership "
"could not be proven"
)
if locked_identity and _text(identity) and locked_identity != _text(identity):
reasons.append(
f"lock claimant '{locked_identity}' does not match active identity "
f"'{_text(identity)}'"
)
if locked_profile and _text(profile) and locked_profile != _text(profile):
reasons.append(
f"lock profile '{locked_profile}' does not match active profile "
f"'{_text(profile)}'"
)
# ── The defining condition: the recorded owner must be dead ─────────────
prior_alive = is_process_alive(recorded_pid)
evidence["prior_pid_alive"] = prior_alive
if prior_alive:
reasons.append(
f"prior owner pid {recorded_pid} is still alive; this is not a "
"dead-session recovery"
)
if current_pid is not None and recorded_pid is not None:
try:
if int(recorded_pid) == int(current_pid):
reasons.append(
"recorded pid is the current session; nothing to recover"
)
except (TypeError, ValueError):
pass
# ── Competing ownership ─────────────────────────────────────────────────
competing: list[dict[str, Any]] = []
for entry in competing_live_locks or ():
if not isinstance(entry, Mapping):
continue
same_issue = entry.get("issue_number") == issue_number
same_branch = _text(entry.get("branch_name")) == locked_branch
if not (same_issue or same_branch):
continue
# The lock we are recovering is not competition with itself.
if (
same_issue
and same_branch
and _same_realpath(_text(entry.get("worktree_path")), worktree_path)
):
continue
competing.append(
{
"issue_number": entry.get("issue_number"),
"branch_name": entry.get("branch_name"),
"worktree_path": entry.get("worktree_path"),
"pid": entry.get("pid"),
}
)
if competing:
described = ", ".join(
f"issue #{c['issue_number']} branch '{c['branch_name']}'" for c in competing
)
reasons.append(f"competing live lock or lease exists ({described})")
evidence["competing_live_locks"] = competing
# ── Ambiguous branch claims ─────────────────────────────────────────────
others = [
name
for name in (candidate_branches or ())
if _text(name) and _text(name) != locked_branch
]
if others:
reasons.append(
"multiple branches claim this issue "
f"({', '.join(sorted(set(others)))}); ownership is ambiguous"
)
evidence["other_candidate_branches"] = sorted(set(others))
if reasons:
return _result(REFUSED, False, reasons, evidence)
return _result(
RECOVERY_SANCTIONED,
True,
[
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"
],
evidence,
)
def _result(
outcome: str,
sanctioned: bool,
reasons: list[str],
evidence: dict[str, Any],
) -> dict[str, Any]:
return {
"outcome": outcome,
"recovery_sanctioned": sanctioned,
"is_candidate": outcome != NO_CANDIDATE,
"reasons": reasons,
"evidence": evidence,
}
def build_recovery_record(
assessment: Mapping[str, Any],
*,
recovered_at: str,
) -> dict[str, Any]:
"""Durable, secret-free provenance for a completed recovery (#753 AC2/AC6)."""
evidence = dict(assessment.get("evidence") or {})
return {
"recovered": True,
"reason": "owning MCP session exited; durable ownership evidence matched",
"recovered_at": recovered_at,
"prior_session_pid": evidence.get("prior_session_pid"),
"replacement_session_pid": evidence.get("replacement_session_pid"),
"prior_pid_alive": evidence.get("prior_pid_alive"),
"branch_name": evidence.get("locked_branch"),
"worktree_path": evidence.get("locked_worktree_path"),
"local_head": evidence.get("local_head"),
"remote_head": evidence.get("remote_head"),
"pr_head": evidence.get("pr_head"),
"pr_number": evidence.get("pr_number"),
"identity": evidence.get("locked_identity"),
"profile": evidence.get("locked_profile"),
"proof": list(assessment.get("reasons") or []),
}
def format_recovery_refusal(assessment: Mapping[str, Any]) -> str:
"""Single fail-closed message for a refused recovery attempt."""
reasons = list(assessment.get("reasons") or []) or [
"dead-session lock recovery evidence did not agree"
]
return (
"Dead-session issue-lock recovery refused: "
+ "; ".join(reasons)
+ " (fail closed)"
)
issue_lock_worktree.py
+21 -2
View File
@@ -92,8 +92,19 @@ def assess_issue_lock_worktree(
inspected_git_root: str | None = None,
base_branch: str | None = None,
base_branches: frozenset[str] | None = None,
recovery_sanctioned: bool = False,
) -> dict:
"""Fail closed when lock preconditions are not met on the declared worktree."""
"""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.
"""
bases = base_branches or BASE_BRANCHES
reasons: list[str] = []
path = (worktree_path or "").strip()
@@ -111,7 +122,11 @@ def assess_issue_lock_worktree(
f"(dirty files: {', '.join(dirty_files)})"
)
if base_equivalent is False:
if recovery_sanctioned:
# Base-equivalence intentionally not evaluated: ownership was proven
# against the durable lock record instead (#753).
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 "
@@ -139,6 +154,7 @@ def assess_issue_lock_worktree(
inspected_git_root=inspected_git_root,
base_branch=base_branch,
base_equivalent=base_equivalent,
recovery_sanctioned=recovery_sanctioned,
)
@@ -197,6 +213,7 @@ def _assessment(
inspected_git_root: str | None = None,
base_branch: str | None = None,
base_equivalent: bool | None = None,
recovery_sanctioned: bool = False,
) -> dict:
return {
"proven": proven,
@@ -208,6 +225,8 @@ def _assessment(
"dirty_files": dirty_files,
"base_branch": base_branch,
"base_equivalent": base_equivalent,
"recovery_sanctioned": recovery_sanctioned,
"base_equivalence_waived": bool(recovery_sanctioned),
}
tests/test_issue_753_dead_pid_lock_recovery.py
+364
View File
@@ -0,0 +1,364 @@
"""Dead-session author issue-lock recovery (#753).
Covers the narrow recovery path that lets an author re-acquire a durable lock
after the MCP session that recorded it exits, plus every rejection condition
that must keep failing closed.
"""
import os
import subprocess
import sys
import unittest
from datetime import datetime, timedelta, timezone
sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
import issue_lock_recovery # noqa: E402
import issue_lock_store # noqa: E402
import issue_lock_worktree # noqa: E402
ISSUE = 4242
BRANCH = f"fix/issue-{ISSUE}-demo"
WORKTREE = "/scratch/wt"
HEAD = "a" * 40
OTHER_SHA = "b" * 40
IDENTITY = "example-user"
PROFILE = "example-author"
def dead_pid() -> int:
"""A PID that has certainly exited (spawned, then reaped)."""
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 assess(lock=None, **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": HEAD,
"remote_head_sha": HEAD,
"pr_head_sha": HEAD,
"pr_number": 99,
"competing_live_locks": [],
"candidate_branches": [BRANCH],
"current_pid": os.getpid(),
}
kwargs.update(overrides)
return issue_lock_recovery.assess_dead_session_lock_recovery(
make_lock() if lock is None else lock, **kwargs
)
class TestDeadSessionRecoveryGranted(unittest.TestCase):
def test_dead_pid_with_exact_evidence_recovers(self):
result = assess()
self.assertTrue(result["recovery_sanctioned"], result["reasons"])
self.assertEqual(result["outcome"], issue_lock_recovery.RECOVERY_SANCTIONED)
def test_recovery_still_granted_when_no_open_pr_exists(self):
# A locked branch need not have a PR yet; absence must not block.
result = assess(pr_head_sha=None, pr_number=None)
self.assertTrue(result["recovery_sanctioned"], result["reasons"])
def test_lease_expiry_is_not_required_for_recovery(self):
# The defining condition is PID death, not TTL expiry (the #601 gap).
lock = make_lock()
self.assertFalse(issue_lock_store.is_lease_expired(lock))
self.assertFalse(issue_lock_store.assess_lock_freshness(lock)["live"])
self.assertTrue(assess(lock)["recovery_sanctioned"])
class TestDeadSessionRecoveryRefused(unittest.TestCase):
def assert_refused(self, result, needle):
self.assertFalse(result["recovery_sanctioned"])
self.assertEqual(result["outcome"], issue_lock_recovery.REFUSED)
self.assertTrue(
any(needle in reason for reason in result["reasons"]),
f"expected {needle!r} in {result['reasons']}",
)
def test_live_prior_pid_refused(self):
lock = make_lock(session_pid=os.getpid(), pid=os.getpid())
# Distinct current pid so the refusal is attributable to liveness.
self.assert_refused(assess(lock, current_pid=os.getpid() + 1), "still alive")
def test_different_author_identity_refused(self):
self.assert_refused(
assess(identity="someone-else"), "does not match active identity"
)
def test_different_profile_refused(self):
self.assert_refused(
assess(profile="other-profile"), "does not match active profile"
)
def test_different_branch_refused(self):
lock = make_lock(branch_name=f"fix/issue-{ISSUE}-other")
self.assert_refused(assess(lock), "does not match requested")
def test_worktree_parked_on_another_branch_refused(self):
self.assert_refused(assess(current_branch="master"), "not the locked branch")
def test_detached_head_worktree_refused(self):
self.assert_refused(assess(current_branch=None), "detached HEAD")
def test_different_worktree_refused(self):
self.assert_refused(
assess(worktree_path="/scratch/elsewhere"), "does not match declared"
)
def test_dirty_worktree_refused(self):
self.assert_refused(
assess(porcelain_status=" M gitea_mcp_server.py\n"), "requires a clean"
)
def test_local_head_differing_from_remote_refused(self):
self.assert_refused(
assess(remote_head_sha=OTHER_SHA), "does not match remote branch head"
)
def test_pr_head_differing_refused(self):
self.assert_refused(assess(pr_head_sha=OTHER_SHA), "does not match local head")
def test_missing_remote_head_refused(self):
self.assert_refused(assess(remote_head_sha=None), "remote head")
def test_competing_live_lock_refused(self):
competing = [
{
"issue_number": ISSUE,
"branch_name": BRANCH,
"worktree_path": "/scratch/other-wt",
"pid": os.getpid(),
}
]
self.assert_refused(
assess(competing_live_locks=competing), "competing live lock"
)
def test_unrelated_live_lock_does_not_block(self):
unrelated = [
{
"issue_number": 999,
"branch_name": "fix/issue-999-unrelated",
"worktree_path": "/scratch/unrelated",
"pid": os.getpid(),
}
]
self.assertTrue(assess(competing_live_locks=unrelated)["recovery_sanctioned"])
def test_multiple_candidate_branches_refused(self):
self.assert_refused(
assess(candidate_branches=[BRANCH, f"feat/issue-{ISSUE}-rival"]),
"multiple branches claim this issue",
)
def test_repository_scope_mismatch_refused(self):
self.assert_refused(assess(repo="OtherRepo"), "does not match requested")
def test_malformed_lock_missing_worktree_refused(self):
lock = make_lock()
lock.pop("worktree_path")
self.assert_refused(assess(lock), "incomplete")
def test_malformed_lock_missing_pid_refused(self):
lock = make_lock()
lock.pop("session_pid", None)
lock.pop("pid", None)
self.assert_refused(assess(lock), "incomplete")
def test_lock_without_claimant_refused(self):
lock = make_lock()
lock["work_lease"] = dict(lock["work_lease"])
lock["work_lease"].pop("claimant")
self.assert_refused(assess(lock), "claimant identity/profile")
class TestNotACandidate(unittest.TestCase):
def test_absent_lock_is_not_a_candidate(self):
result = issue_lock_recovery.assess_dead_session_lock_recovery(
None,
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=HEAD,
remote_head_sha=HEAD,
)
self.assertEqual(result["outcome"], issue_lock_recovery.NO_CANDIDATE)
self.assertFalse(result["recovery_sanctioned"])
self.assertFalse(result["is_candidate"])
def test_lock_for_a_different_issue_is_not_a_candidate(self):
result = assess(make_lock(issue_number=7777))
self.assertEqual(result["outcome"], issue_lock_recovery.NO_CANDIDATE)
self.assertFalse(result["recovery_sanctioned"])
class TestWorktreeGateWaiver(unittest.TestCase):
def test_new_issue_claim_still_requires_base_equivalence(self):
result = issue_lock_worktree.assess_issue_lock_worktree(
worktree_path=WORKTREE,
current_branch=BRANCH,
porcelain_status="",
base_equivalent=False,
)
self.assertTrue(result["block"])
self.assertFalse(result["base_equivalence_waived"])
def test_sanctioned_recovery_waives_base_equivalence(self):
result = issue_lock_worktree.assess_issue_lock_worktree(
worktree_path=WORKTREE,
current_branch=BRANCH,
porcelain_status="",
base_equivalent=False,
recovery_sanctioned=True,
)
self.assertTrue(result["proven"], result["reasons"])
self.assertTrue(result["base_equivalence_waived"])
def test_recovery_never_waives_cleanliness(self):
result = issue_lock_worktree.assess_issue_lock_worktree(
worktree_path=WORKTREE,
current_branch=BRANCH,
porcelain_status=" M gitea_mcp_server.py\n",
base_equivalent=False,
recovery_sanctioned=True,
)
self.assertTrue(result["block"])
self.assertTrue(
any("tracked file edits" in reason for reason in result["reasons"])
)
def test_unproven_base_equivalence_still_blocks_without_recovery(self):
result = issue_lock_worktree.assess_issue_lock_worktree(
worktree_path=WORKTREE,
current_branch=BRANCH,
porcelain_status="",
base_equivalent=None,
)
self.assertTrue(result["block"])
class TestRecoveryRecordAndDownstream(unittest.TestCase):
def test_recovery_record_preserves_truthful_provenance(self):
assessment = assess()
prior = assessment["evidence"]["prior_session_pid"]
record = issue_lock_recovery.build_recovery_record(
assessment, recovered_at="2026-07-18T23:21:40Z"
)
self.assertTrue(record["recovered"])
self.assertEqual(record["prior_session_pid"], prior)
self.assertEqual(record["replacement_session_pid"], os.getpid())
self.assertNotEqual(
record["prior_session_pid"], record["replacement_session_pid"]
)
self.assertFalse(record["prior_pid_alive"])
self.assertEqual(record["recovered_at"], "2026-07-18T23:21:40Z")
self.assertEqual(record["branch_name"], BRANCH)
self.assertEqual(record["local_head"], HEAD)
self.assertEqual(record["identity"], IDENTITY)
self.assertTrue(record["proof"])
def test_recovery_record_carries_no_secret_material(self):
record = issue_lock_recovery.build_recovery_record(
assess(), recovered_at="2026-07-18T23:21:40Z"
)
blob = repr(record).lower()
for banned in ("token", "password", "authorization", "secret", "api_key"):
self.assertNotIn(banned, blob)
def test_recovered_lock_satisfies_update_by_merge_ownership(self):
# After recovery the lock is rebound to the live session, so the
# ownership re-check used by gitea_update_pr_branch_by_merge passes.
assessment = assess()
recovered_lock = make_lock(session_pid=os.getpid(), pid=os.getpid())
recovered_lock["dead_session_recovery"] = (
issue_lock_recovery.build_recovery_record(
assessment, recovered_at="2026-07-18T23:21:40Z"
)
)
freshness = issue_lock_store.assess_lock_freshness(recovered_lock)
self.assertTrue(freshness["live"], freshness)
verdict = issue_lock_store.verify_lock_for_mutation(
recovered_lock,
issue_number=ISSUE,
branch_name=BRANCH,
worktree_path=WORKTREE,
)
self.assertTrue(verdict["proven"], verdict["reasons"])
self.assertFalse(verdict["block"])
def test_pre_recovery_lock_fails_ownership_check(self):
# Guards against a false positive above: the dead-PID lock must fail.
verdict = issue_lock_store.verify_lock_for_mutation(
make_lock(),
issue_number=ISSUE,
branch_name=BRANCH,
worktree_path=WORKTREE,
)
self.assertTrue(verdict["block"])
self.assertTrue(any("not live" in reason for reason in verdict["reasons"]))
def test_no_manual_file_seeding_required(self):
# The whole decision is reachable from the durable record plus live
# observation; nothing is written to disk to reach a verdict.
self.assertTrue(assess()["recovery_sanctioned"])
def test_refusal_message_is_fail_closed(self):
message = issue_lock_recovery.format_recovery_refusal(
assess(porcelain_status=" M x.py\n")
)
self.assertIn("fail closed", message)
self.assertIn("recovery refused", message.lower())
if __name__ == "__main__":
unittest.main()