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

Merge pull request 'feat(lease): guarded cleanup for obsolete reviewer comment leases (Closes #691)' (#692) from fix/issue-691-obsolete-reviewer-lease-cleanup into master

Browse Source
This commit was merged in pull request #692.
This commit is contained in:
sysadmin
2026-07-12 22:28:59 -05:00
parent 56f1230a10 6d86db793b
commit 237656702f
5 changed files with 1542 additions and 48 deletions
Download Patch File Download Diff File Expand all files Collapse all files
gitea_mcp_server.py
+275 -5
View File
@@ -7964,11 +7964,12 @@ def gitea_diagnose_reviewer_pr_lease_handoff(
org: str | None = None,
repo: str | None = None,
) -> dict:
"""Read-only: diagnose open-PR reviewer lease handoff and emit next action (#599).
"""Read-only: diagnose open-PR reviewer lease handoff and emit next action (#599, #691).
Classifies no-lease / own / foreign / instructed-lease-missing-with-replacement
and worktree binding mismatch. Returns a canonical ``next_action`` without
mutating Gitea state. Never steals foreign leases.
Classifies no-lease / own / foreign / superseded-head / expired / orphan /
instructed-lease-missing-with-replacement and worktree binding mismatch.
Returns a canonical ``next_action`` (including guarded obsolete-lease cleanup)
without mutating Gitea state. Never steals foreign leases.
"""
read_block = _profile_operation_gate("gitea.read")
if read_block:
@@ -7979,7 +7980,7 @@ def gitea_diagnose_reviewer_pr_lease_handoff(
}
comments = _fetch_pr_comments(
pr_number, remote=remote, host=host, org=org, repo=repo)
h, _o, _r = _resolve(remote, host, org, repo)
h, o, r = _resolve(remote, host, org, repo)
try:
username = _authenticated_username(h)
except Exception:
@@ -7993,6 +7994,64 @@ def gitea_diagnose_reviewer_pr_lease_handoff(
# Prefer in-session lease id when present; otherwise diagnose as a fresh
# caller without inventing ownership of an on-thread lease.
session_id = (session_lease.get("session_id") or "").strip() or None
# Live PR head + formal reviews for #691 superseded/completed classification.
current_head_sha = None
formal_reviews: list[dict] = []
try:
auth = _auth(h)
pr_live = api_request(
"GET", f"{repo_api_url(h, o, r)}/pulls/{pr_number}", auth
) or {}
head = pr_live.get("head") or {}
current_head_sha = head.get("sha") or pr_live.get("head_commit_sha")
try:
reviews = api_request(
"GET",
f"{repo_api_url(h, o, r)}/pulls/{pr_number}/reviews",
auth,
) or []
for rev in reviews if isinstance(reviews, list) else []:
formal_reviews.append({
"verdict": rev.get("state") or rev.get("verdict"),
"reviewed_head_sha": rev.get("commit_id") or rev.get(
"reviewed_head_sha"
),
"dismissed": bool(rev.get("dismissed")),
"reviewer": ((rev.get("user") or {}).get("login")),
"submitted_at": rev.get("submitted_at"),
})
except Exception:
formal_reviews = []
except Exception:
current_head_sha = None
lease_wt = None
active = reviewer_pr_lease.find_newest_nonterminal_lease(
comments, pr_number=pr_number, include_expired=True
)
if active:
lease_wt = (active.get("worktree") or "").strip() or None
worktree_exists = None
worktree_clean = None
if lease_wt:
worktree_exists = os.path.isdir(lease_wt)
if worktree_exists:
try:
import subprocess
st = subprocess.run(
["git", "-C", lease_wt, "status", "--porcelain"],
capture_output=True,
text=True,
timeout=30,
check=False,
)
if st.returncode == 0:
worktree_clean = not bool((st.stdout or "").strip())
except Exception:
worktree_clean = None
diagnosis = reviewer_pr_lease.diagnose_reviewer_pr_lease_handoff(
comments,
pr_number=pr_number,
@@ -8002,6 +8061,11 @@ def gitea_diagnose_reviewer_pr_lease_handoff(
env_bound_worktree=env_wt,
instructed_session_id=instructed_session_id,
instructed_comment_id=instructed_comment_id,
current_head_sha=current_head_sha,
formal_reviews=formal_reviews,
worktree_exists=worktree_exists,
worktree_clean=worktree_clean,
owner_process_alive=None, # PID never proves ownership; leave unset
)
diagnosis["success"] = True
diagnosis["remote"] = remote
@@ -8129,6 +8193,212 @@ def gitea_cleanup_post_merge_moot_lease(
return report
@mcp.tool()
def gitea_cleanup_obsolete_reviewer_comment_lease(
pr_number: int,
confirmation: str = "",
apply: bool = False,
controller_recovery_authorized: bool = False,
expected_lease_comment_id: int | None = None,
expected_session_id: str | None = None,
expected_leased_head: str | None = None,
owner_process_alive: bool | None = None,
current_head_review_in_progress: bool = False,
remote: str = "dadeschools",
host: str | None = None,
org: str | None = None,
repo: str | None = None,
) -> dict:
"""Guarded cleanup for obsolete comment-backed reviewer leases on open PRs (#691).
Neutralises an expired and/or superseded-head foreign lease by posting a
terminal ``phase: released`` lease marker (append-only audit). Never deletes
comments, never repoints the lease to the new head, never transfers
validation/decision/workflow state, never uses PID equality as ownership,
and never steals a genuinely active foreign lease on the current head.
Read-first when ``apply`` is false. Apply requires:
- ``controller_recovery_authorized=True`` (explicit operator/controller
recovery capability — not implied by profile alone)
- ``confirmation`` exactly ``CLEANUP OBSOLETE REVIEWER LEASE <pr_number>``
- full eligibility evidence from
``reviewer_pr_lease.assess_obsolete_reviewer_comment_lease_cleanup``
Comment-backed leases need no control-plane DB ``lease_id``.
"""
read_block = _profile_operation_gate("gitea.read")
if read_block:
return {
"success": False,
"cleanup_performed": False,
"no_steal_or_adoption": True,
"reasons": read_block,
"permission_report": _permission_block_report("gitea.read"),
}
h, o, r = _resolve(remote, host, org, repo)
auth = _auth(h)
expected_repo = f"{o}/{r}"
comments = _fetch_pr_comments(
pr_number, remote=remote, host=host, org=org, repo=repo
)
pr_live = api_request(
"GET", f"{repo_api_url(h, o, r)}/pulls/{pr_number}", auth
) or {}
head = pr_live.get("head") or {}
current_head_sha = head.get("sha") or pr_live.get("head_commit_sha")
formal_reviews: list[dict] = []
try:
reviews = api_request(
"GET",
f"{repo_api_url(h, o, r)}/pulls/{pr_number}/reviews",
auth,
) or []
for rev in reviews if isinstance(reviews, list) else []:
formal_reviews.append({
"verdict": rev.get("state") or rev.get("verdict"),
"reviewed_head_sha": rev.get("commit_id")
or rev.get("reviewed_head_sha"),
"dismissed": bool(rev.get("dismissed")),
"reviewer": ((rev.get("user") or {}).get("login")),
"submitted_at": rev.get("submitted_at"),
})
except Exception:
formal_reviews = []
lease = reviewer_pr_lease.find_newest_nonterminal_lease(
comments, pr_number=pr_number, include_expired=True
)
lease_wt = ((lease or {}).get("worktree") or "").strip() or None
worktree_exists = None
worktree_clean = None
worktree_has_unpreserved_work = None
if lease_wt:
worktree_exists = os.path.isdir(lease_wt)
if worktree_exists:
try:
import subprocess
st = subprocess.run(
["git", "-C", lease_wt, "status", "--porcelain"],
capture_output=True,
text=True,
timeout=30,
check=False,
)
if st.returncode == 0:
dirty = bool((st.stdout or "").strip())
worktree_clean = not dirty
worktree_has_unpreserved_work = dirty
except Exception:
worktree_clean = None
session = reviewer_pr_lease.get_session_lease() or {}
assessment = reviewer_pr_lease.assess_obsolete_reviewer_comment_lease_cleanup(
comments,
pr_number=pr_number,
current_head_sha=current_head_sha,
formal_reviews=formal_reviews,
repo=expected_repo,
expected_repo=expected_repo,
requesting_session_id=(session.get("session_id") or "").strip() or None,
controller_recovery_authorized=controller_recovery_authorized,
worktree_exists=worktree_exists,
worktree_clean=worktree_clean,
worktree_has_unpreserved_work=worktree_has_unpreserved_work,
owner_process_alive=owner_process_alive,
owner_pid_observed=None,
requesting_pid=os.getpid(),
current_head_review_in_progress=current_head_review_in_progress,
expected_lease_comment_id=expected_lease_comment_id,
expected_session_id=expected_session_id,
expected_leased_head=expected_leased_head,
confirmation=confirmation,
apply=apply,
)
report = {
"success": True,
"pr_number": pr_number,
"pr_state": pr_live.get("state"),
"cleanup_allowed": assessment.get("cleanup_allowed"),
"cleanup_performed": False,
"classification": assessment.get("classification"),
"blocker_kind": assessment.get("blocker_kind"),
"exact_next_action": assessment.get("exact_next_action"),
"mutation_eligibility": assessment.get("mutation_eligibility"),
"leased_head": assessment.get("leased_head"),
"current_head": assessment.get("current_head"),
"expires_at": assessment.get("expires_at"),
"terminal_review": assessment.get("terminal_review"),
"worktree_state": assessment.get("worktree_state"),
"owner_session_evidence": assessment.get("owner_session_evidence"),
"cleanup_tool": assessment.get("cleanup_tool"),
"required_confirmation": assessment.get("required_confirmation"),
"active_lease": assessment.get("active_lease"),
"no_steal_or_adoption": True,
"no_repoint": True,
"no_validation_transfer": True,
"mode": "apply" if apply else "read_only",
"reasons": assessment.get("reasons") or [],
"forbidden": assessment.get("forbidden") or [],
}
if not apply:
return report
if not assessment.get("cleanup_allowed"):
report["success"] = False
report["cleanup_skipped_reason"] = (
assessment.get("fail_closed_reasons")
or assessment.get("reasons")
or ["cleanup not allowed"]
)
return report
comment_block = _profile_operation_gate("gitea.pr.comment")
if comment_block:
report["success"] = False
report["reasons"] = comment_block
report["permission_report"] = _permission_block_report("gitea.pr.comment")
return report
verify_preflight_purity(remote)
# Terminal lease marker (preserves history; does not delete comment 10749-style
# entries — newest released marker neutralises the ledger).
release_body = assessment.get("release_body") or ""
audit_body = assessment.get("audit_comment_body") or ""
comment_url = f"{repo_api_url(h, o, r)}/issues/{pr_number}/comments"
with _audited(
"comment_pr",
host=h,
remote=remote,
org=o,
repo=r,
pr_number=pr_number,
request_metadata={
"source": "cleanup_obsolete_reviewer_comment_lease",
"classification": assessment.get("classification"),
},
):
posted_release = api_request(
"POST", comment_url, auth, {"body": release_body}
)
posted_audit = None
if audit_body.strip():
posted_audit = api_request(
"POST", comment_url, auth, {"body": audit_body}
)
# Never seed session lease from cleaned foreign lease.
report["cleanup_performed"] = True
report["released_comment_id"] = (posted_release or {}).get("id")
report["audit_comment_id"] = (posted_audit or {}).get("id")
report["session_lease_seeded"] = False
return report
@mcp.tool()
def gitea_release_reviewer_pr_lease(
pr_number: int,
mcp_native_cleanup_proof.py
+1
View File
@@ -13,6 +13,7 @@ from typing import Any
AUTHORIZED_CLEANUP_TOOLS = frozenset({
"gitea_cleanup_post_merge_moot_lease",
"gitea_cleanup_obsolete_reviewer_comment_lease",
"gitea_reconcile_merged_cleanups",
"gitea_delete_branch",
"gitea_cleanup_merged_pr_branch",
reviewer_pr_lease.py
+657 -43
View File
@@ -524,13 +524,26 @@ def assess_lease_inventory(
"reclaimable_review_leases": reclaimable,
}
# Canonical next-action vocabulary for reviewer lease handoff (#599).
# Canonical next-action vocabulary for reviewer lease handoff (#599, #691).
NEXT_ACTION_ACQUIRE = "acquire"
NEXT_ACTION_WAIT = "wait"
NEXT_ACTION_RESUME_EXACT_OWNER_SESSION = "resume_exact_owner_session"
NEXT_ACTION_RELEASE_EXPIRED_LEASE = "release_expired_lease"
NEXT_ACTION_OPERATOR_AUTHORIZED_CLEANUP = "operator_authorized_cleanup"
NEXT_ACTION_REPAIR_WORKTREE_BINDING = "repair_worktree_binding"
NEXT_ACTION_CLEANUP_OBSOLETE_LEASE = "cleanup_obsolete_reviewer_comment_lease"
CLEANUP_OBSOLETE_LEASE_TOOL = "gitea_cleanup_obsolete_reviewer_comment_lease"
CLEANUP_OBSOLETE_LEASE_CONFIRMATION_PREFIX = "CLEANUP OBSOLETE REVIEWER LEASE "
# Formal terminal review verdicts that complete a leased-head workflow (#691).
_TERMINAL_REVIEW_VERDICTS = frozenset({
"APPROVED",
"REQUEST_CHANGES",
"approved",
"request_changes",
"REQUESTCHANGES",
})
_HANDOFF_CLASSIFICATIONS = frozenset({
"no_lease",
@@ -539,6 +552,12 @@ _HANDOFF_CLASSIFICATIONS = frozenset({
"foreign_active",
"foreign_reclaimable",
"foreign_expired",
"foreign_active_current_head",
"foreign_expired_current_head",
"foreign_completed_superseded_head",
"foreign_expired_superseded_head",
"orphaned_owner_missing",
"ambiguous_conflicting_evidence",
"instructed_lease_missing_with_replacement",
"worktree_binding_mismatch",
})
@@ -551,6 +570,476 @@ def _norm_path(value: str | None) -> str:
return os.path.normpath(text.rstrip("/"))
def _normalize_verdict(value: str | None) -> str:
text = (value or "").strip().upper().replace("-", "_").replace(" ", "_")
if text in {"REQUESTCHANGES", "REQUEST_CHANGES", "CHANGES_REQUESTED"}:
return "REQUEST_CHANGES"
if text in {"APPROVED", "APPROVE"}:
return "APPROVED"
return text
def formal_terminal_review_for_head(
formal_reviews: list[dict] | None,
*,
leased_head: str | None,
) -> dict[str, Any] | None:
"""Return the latest undismissed terminal formal review for *leased_head*."""
head = _normalize_sha(leased_head)
if not head:
return None
matches: list[dict[str, Any]] = []
for review in formal_reviews or []:
if review.get("dismissed"):
continue
verdict = _normalize_verdict(
review.get("verdict") or review.get("state") or review.get("body_state")
)
if verdict not in {"APPROVED", "REQUEST_CHANGES"}:
continue
rhead = _normalize_sha(
review.get("reviewed_head_sha")
or review.get("commit_id")
or review.get("head_sha")
)
if rhead != head:
continue
matches.append({**review, "verdict": verdict, "reviewed_head_sha": rhead})
if not matches:
return None
return matches[-1]
def find_newest_nonterminal_lease(
comments: list[dict],
*,
pr_number: int,
now: datetime | None = None,
include_expired: bool = True,
) -> dict[str, Any] | None:
"""Newest non-terminal lease marker, optionally including expired ones (#691).
Unlike ``find_active_reviewer_lease``, this retains expired non-terminal
markers so diagnosis/cleanup can still name the obsolete lease after
``expires_at`` (comment-backed ledger; no control-plane ``lease_id``).
"""
now = now or datetime.now(timezone.utc)
entries = list(reversed(_lease_entries(comments, pr_number=pr_number)))
for entry in entries:
phase = (entry.get("phase") or "").strip().lower()
if phase in _TERMINAL_PHASES:
# Newest terminal ends the ledger chain for active acquisition, but
# an older non-terminal is not "active". Stop at newest marker.
return None
expired = _lease_expired(entry, now=now)
if expired and not include_expired:
return None
lease = dict(entry)
lease["freshness"] = classify_lease_freshness(lease, now=now)
lease["expired"] = expired
return lease
return None
def cleanup_confirmation_for_pr(pr_number: int) -> str:
return f"{CLEANUP_OBSOLETE_LEASE_CONFIRMATION_PREFIX}{int(pr_number)}"
def assess_obsolete_reviewer_comment_lease_cleanup(
comments: list[dict],
*,
pr_number: int,
current_head_sha: str | None,
formal_reviews: list[dict] | None = None,
repo: str | None = None,
expected_repo: str | None = None,
requesting_session_id: str | None = None,
controller_recovery_authorized: bool = False,
worktree_exists: bool | None = None,
worktree_clean: bool | None = None,
worktree_has_unpreserved_work: bool | None = None,
owner_process_alive: bool | None = None,
owner_pid_observed: int | None = None,
requesting_pid: int | None = None,
current_head_review_in_progress: bool = False,
expected_lease_comment_id: int | None = None,
expected_session_id: str | None = None,
expected_leased_head: str | None = None,
confirmation: str | None = None,
apply: bool = False,
now: datetime | None = None,
) -> dict[str, Any]:
"""Guarded cleanup eligibility for obsolete comment-backed reviewer leases (#691).
Cleanup is never transfer/adoption/repoint and never uses PID equality as
ownership. Eligible only when evidence proves the lease is past expiry and/or
pinned to a superseded head with a completed formal terminal review, and
every safety boundary holds.
"""
now = now or datetime.now(timezone.utc)
reasons: list[str] = []
fail_closed_reasons: list[str] = []
current_head = _normalize_sha(current_head_sha)
req_session = (requesting_session_id or "").strip()
lease = find_newest_nonterminal_lease(
comments, pr_number=pr_number, now=now, include_expired=True
)
if not lease:
# Fall back to active finder (unexpired only) for report consistency.
lease = find_active_reviewer_lease(comments, pr_number=pr_number, now=now)
classification = "no_lease"
cleanup_allowed = False
release_body: str | None = None
audit_comment_body: str | None = None
terminal_review = None
leased_head = None
head_superseded = False
past_expiry = False
expires_parseable = True
if not lease:
fail_closed_reasons.append(
f"PR #{pr_number}: no non-terminal comment-backed reviewer lease to clean"
)
classification = "no_lease"
else:
leased_head = _normalize_sha(lease.get("candidate_head"))
expires_raw = lease.get("expires_at")
expires_at = _parse_timestamp(expires_raw)
if expires_raw and expires_at is None:
expires_parseable = False
fail_closed_reasons.append(
"lease expires_at cannot be parsed or trusted (fail closed)"
)
past_expiry = bool(expires_at and expires_at <= now)
freshness = lease.get("freshness") or classify_lease_freshness(lease, now=now)
owner_session = (lease.get("session_id") or "").strip()
is_requesting_owner = bool(
req_session and owner_session and req_session == owner_session
)
# Identity pins (repo / PR / session / head / comment) must match when
# the caller supplies expected evidence.
lease_repo = (lease.get("repo") or "").strip()
exp_repo = (expected_repo or repo or "").strip()
if exp_repo and lease_repo and exp_repo != lease_repo:
fail_closed_reasons.append(
f"repository identity mismatch: lease repo={lease_repo!r} "
f"expected={exp_repo!r}"
)
if lease.get("pr_number") not in (None, pr_number):
fail_closed_reasons.append(
f"PR identity mismatch: lease pr={lease.get('pr_number')} "
f"requested={pr_number}"
)
if expected_lease_comment_id is not None:
cid = lease.get("comment_id")
if cid is None or int(cid) != int(expected_lease_comment_id):
fail_closed_reasons.append(
"lease comment_id does not match expected evidence "
f"(lease={cid}, expected={expected_lease_comment_id})"
)
if expected_session_id:
if owner_session != expected_session_id.strip():
fail_closed_reasons.append(
"lease session_id does not match expected evidence "
f"(lease={owner_session}, expected={expected_session_id})"
)
if expected_leased_head:
exp_head = _normalize_sha(expected_leased_head)
if not exp_head or exp_head != leased_head:
fail_closed_reasons.append(
"leased head does not match expected evidence "
f"(lease={leased_head}, expected={exp_head})"
)
# PID equality is never ownership proof (#691).
if (
owner_pid_observed is not None
and requesting_pid is not None
and int(owner_pid_observed) == int(requesting_pid)
):
reasons.append(
"observed PID equals requesting PID but PID equality is NOT "
"ownership proof; ignored for authorization"
)
if is_requesting_owner:
fail_closed_reasons.append(
"requesting session owns the lease; use "
"gitea_release_reviewer_pr_lease (owner path), not non-owner cleanup"
)
if current_head_review_in_progress:
fail_closed_reasons.append(
"a current-head review submission is in progress; cleanup denied"
)
if not current_head:
fail_closed_reasons.append(
"current PR head SHA missing or unparseable (fail closed)"
)
if not leased_head:
fail_closed_reasons.append(
"lease candidate_head missing or unparseable (fail closed)"
)
head_superseded = bool(
current_head and leased_head and current_head != leased_head
)
head_matches_current = bool(
current_head and leased_head and current_head == leased_head
)
terminal_review = formal_terminal_review_for_head(
formal_reviews, leased_head=leased_head
)
has_terminal = terminal_review is not None
# Worktree safety.
if worktree_has_unpreserved_work is True or worktree_clean is False:
fail_closed_reasons.append(
"lease worktree is dirty or has unpreserved work; cleanup denied"
)
if (
worktree_exists is True
and worktree_clean is None
and worktree_has_unpreserved_work is None
):
# Unknown cleanliness when path exists — fail closed.
fail_closed_reasons.append(
"lease worktree exists but cleanliness is unknown (fail closed)"
)
# Classification matrix (#691).
if head_matches_current and freshness in {"active", "stale_warning"}:
classification = "foreign_active_current_head"
fail_closed_reasons.append(
"genuinely active foreign lease on the current PR head; "
"cleanup denied (fail closed)"
)
elif head_matches_current and past_expiry:
classification = "foreign_expired_current_head"
# Expired on current head: owner-missing / orphan path may apply.
if owner_process_alive is True:
fail_closed_reasons.append(
"lease expired on current head but owner process still alive "
"with plausible activity; cleanup denied"
)
elif worktree_clean is False:
fail_closed_reasons.append(
"owner process absent but worktree dirty; cleanup denied"
)
elif owner_process_alive is False and (
worktree_clean is True or worktree_exists is False
):
if controller_recovery_authorized:
classification = "orphaned_owner_missing"
else:
fail_closed_reasons.append(
"controller/recovery capability not authorized for orphan cleanup"
)
else:
fail_closed_reasons.append(
"insufficient orphan evidence for expired current-head lease "
"(need owner_process_alive=false and clean/absent worktree)"
)
elif head_superseded and has_terminal and past_expiry:
classification = "foreign_expired_superseded_head"
elif head_superseded and has_terminal and not past_expiry:
classification = "foreign_completed_superseded_head"
elif head_superseded and not has_terminal:
classification = "ambiguous_conflicting_evidence"
fail_closed_reasons.append(
"lease head is superseded but no formal terminal review exists "
"for the leased head (fail closed)"
)
elif not head_superseded and not past_expiry and freshness in {
"active", "stale_warning"
}:
classification = "foreign_active_current_head"
fail_closed_reasons.append(
"foreign lease still active on current head; wait or use owner release"
)
elif past_expiry and not head_superseded:
classification = "foreign_expired_current_head"
else:
classification = "ambiguous_conflicting_evidence"
fail_closed_reasons.append(
"lease evidence is ambiguous; cleanup denied (fail closed)"
)
# Recent owner progress on a non-superseded active lease blocks cleanup.
minutes = _minutes_since_activity(lease, now=now)
if (
head_matches_current
and freshness == "active"
and minutes is not None
and minutes < STALE_WARNING_MINUTES
):
fail_closed_reasons.append(
"owner session has recent authenticated progress on current head; "
"cleanup denied"
)
# Authority + confirmation for apply.
if not controller_recovery_authorized:
if classification in {
"foreign_completed_superseded_head",
"foreign_expired_superseded_head",
"foreign_expired_current_head",
"orphaned_owner_missing",
}:
fail_closed_reasons.append(
"controller/recovery capability required "
"(controller_recovery_authorized=true)"
)
expected_conf = cleanup_confirmation_for_pr(pr_number)
if apply:
if (confirmation or "").strip() != expected_conf:
fail_closed_reasons.append(
f"confirmation must equal exactly {expected_conf!r}"
)
# Superseded + terminal path does not require past_expiry (AC1).
eligible_class = classification in {
"foreign_completed_superseded_head",
"foreign_expired_superseded_head",
"orphaned_owner_missing",
"foreign_expired_current_head",
}
# foreign_expired_current_head needs orphan/clean worktree + authority.
if classification == "foreign_expired_current_head":
if worktree_clean is not True and worktree_exists is not False:
if "worktree" not in " ".join(fail_closed_reasons):
fail_closed_reasons.append(
"expired current-head cleanup requires proven-clean "
"worktree or absent worktree"
)
if owner_process_alive is True:
fail_closed_reasons.append(
"owner process still alive on expired current-head lease"
)
cleanup_allowed = (
eligible_class
and expires_parseable
and not fail_closed_reasons
and not is_requesting_owner
)
if cleanup_allowed:
release_body = format_lease_body(
repo=lease.get("repo") or exp_repo or "",
pr_number=pr_number,
issue_number=lease.get("issue_number"),
reviewer_identity=lease.get("reviewer_identity") or "",
profile=lease.get("profile") or "unknown",
session_id=owner_session or "unknown",
worktree=lease.get("worktree") or "",
phase="released",
candidate_head=leased_head,
target_branch=lease.get("target_branch") or "master",
target_branch_sha=lease.get("target_branch_sha"),
last_activity=now,
blocker="obsolete-superseded-or-expired-lease",
)
audit_comment_body = (
"## Canonical obsolete reviewer lease cleanup (#691)\n\n"
f"- pr: #{pr_number}\n"
f"- lease_comment_id: {lease.get('comment_id')}\n"
f"- session_id: {owner_session}\n"
f"- leased_head: {leased_head}\n"
f"- current_head: {current_head}\n"
f"- expires_at: {lease.get('expires_at')}\n"
f"- classification: {classification}\n"
f"- terminal_review_verdict: "
f"{(terminal_review or {}).get('verdict')}\n"
f"- tool: {CLEANUP_OBSOLETE_LEASE_TOOL}\n"
"- action: posted terminal phase=released lease marker; "
"did not transfer validation, decision, or workflow proof; "
"did not repoint lease to the new head; did not delete history\n"
)
reasons.append(
f"cleanup eligible ({classification}); post terminal released "
"marker via sanctioned tool"
)
else:
reasons.extend(fail_closed_reasons)
return {
"pr_number": pr_number,
"classification": classification,
"blocker_kind": classification if not cleanup_allowed else "none",
"cleanup_allowed": cleanup_allowed,
"mutation_allowed": False, # never grants review mutation
"mutation_eligibility": "prohibited" if not cleanup_allowed else "cleanup_only",
"exact_next_action": (
NEXT_ACTION_CLEANUP_OBSOLETE_LEASE
if cleanup_allowed
else (
NEXT_ACTION_WAIT
if classification
in {
"foreign_active_current_head",
"ambiguous_conflicting_evidence",
}
else NEXT_ACTION_OPERATOR_AUTHORIZED_CLEANUP
)
),
"cleanup_tool": CLEANUP_OBSOLETE_LEASE_TOOL,
"required_confirmation": cleanup_confirmation_for_pr(pr_number),
"leased_head": leased_head,
"current_head": current_head,
"head_superseded": head_superseded,
"past_expiry": past_expiry,
"expires_at": (lease or {}).get("expires_at") if lease else None,
"expires_parseable": expires_parseable,
"terminal_review": terminal_review,
"terminal_review_present": terminal_review is not None,
"worktree_state": {
"path": (lease or {}).get("worktree") if lease else None,
"exists": worktree_exists,
"clean": worktree_clean,
"has_unpreserved_work": worktree_has_unpreserved_work,
},
"owner_session_evidence": {
"session_id": (lease or {}).get("session_id") if lease else None,
"reviewer_identity": (
(lease or {}).get("reviewer_identity") if lease else None
),
"process_alive": owner_process_alive,
"pid_observed": owner_pid_observed,
"pid_is_not_ownership_proof": True,
"requesting_session_is_owner": bool(
lease
and req_session
and (lease.get("session_id") or "").strip() == req_session
),
},
"active_lease": lease,
"release_body": release_body,
"audit_comment_body": audit_comment_body,
"controller_recovery_authorized": controller_recovery_authorized,
"reasons": reasons if reasons else fail_closed_reasons,
"fail_closed_reasons": fail_closed_reasons,
"forbidden": [
"manual comment deletion",
"database edits",
"mtime manipulation",
"PID-based ownership",
"direct session-state seeding",
"lease stealing or adoption",
"repointing old lease to new head",
"transfer of validation or decision state",
"worktree reuse by replacement reviewer",
],
}
def diagnose_reviewer_pr_lease_handoff(
comments: list[dict],
*,
@@ -561,40 +1050,50 @@ def diagnose_reviewer_pr_lease_handoff(
env_bound_worktree: str | None = None,
instructed_session_id: str | None = None,
instructed_comment_id: int | None = None,
current_head_sha: str | None = None,
formal_reviews: list[dict] | None = None,
worktree_exists: bool | None = None,
worktree_clean: bool | None = None,
owner_process_alive: bool | None = None,
now: datetime | None = None,
) -> dict[str, Any]:
"""Classify open-PR reviewer lease handoff and emit a canonical next action (#599).
"""Classify open-PR reviewer lease handoff and emit a canonical next action (#599, #691).
Read-only diagnosis. Never steals, releases, or adopts a foreign lease.
Fail-closed acquisition rules for active foreign leases remain intact.
Returns a structured diagnosis with:
- classification
- next_action (one of wait / resume_exact_owner_session /
release_expired_lease / operator_authorized_cleanup /
repair_worktree_binding / acquire)
- classification (including #691 superseded/expired distinctions)
- next_action (including cleanup_obsolete_reviewer_comment_lease)
- active_lease identity fields when present
- worktree_binding match result
- instructed-lease mismatch flags
- cleanup tool/confirmation when cleanup-eligible
"""
now = now or datetime.now(timezone.utc)
session_id = (current_session_id or "").strip()
identity = (current_reviewer_identity or "").strip()
instructed_sid = (instructed_session_id or "").strip()
reasons: list[str] = []
current_head = _normalize_sha(current_head_sha)
active = find_active_reviewer_lease(comments, pr_number=pr_number, now=now)
# Also surface expired non-terminal newest marker for #691 diagnosis.
newest_nt = find_newest_nonterminal_lease(
comments, pr_number=pr_number, now=now, include_expired=True
)
lease_for_class = active or newest_nt
session_lease = get_session_lease()
# Worktree binding: env-bound vs proposed vs active lease worktree.
env_wt = _norm_path(env_bound_worktree)
prop_wt = _norm_path(proposed_worktree)
lease_wt = _norm_path((active or {}).get("worktree") if active else None)
lease_wt = _norm_path((lease_for_class or {}).get("worktree") if lease_for_class else None)
binding_mismatch = False
binding_details: dict[str, Any] = {
"env_bound_worktree": env_bound_worktree or None,
"proposed_worktree": proposed_worktree or None,
"lease_worktree": (active or {}).get("worktree") if active else None,
"lease_worktree": (lease_for_class or {}).get("worktree") if lease_for_class else None,
"match": True,
}
paths = [p for p in (env_wt, prop_wt, lease_wt) if p]
@@ -608,13 +1107,13 @@ def diagnose_reviewer_pr_lease_handoff(
# Instructed lease gone while a different lease is active (PR #592-style).
instructed_missing_with_replacement = False
if instructed_sid or instructed_comment_id is not None:
if not active:
if not lease_for_class:
reasons.append(
"instructed lease is gone and no active replacement lease remains"
)
else:
owner = (active.get("session_id") or "").strip()
cid = active.get("comment_id")
owner = (lease_for_class.get("session_id") or "").strip()
cid = lease_for_class.get("comment_id")
sid_mismatch = bool(instructed_sid and owner and owner != instructed_sid)
cid_mismatch = (
instructed_comment_id is not None
@@ -631,35 +1130,94 @@ def diagnose_reviewer_pr_lease_handoff(
# Classification + next_action.
classification = "no_lease"
next_action = NEXT_ACTION_ACQUIRE
cleanup_hint: dict[str, Any] | None = None
if active:
owner = (active.get("session_id") or "").strip()
freshness = active.get("freshness") or classify_lease_freshness(
active, now=now
if lease_for_class:
owner = (lease_for_class.get("session_id") or "").strip()
freshness = lease_for_class.get("freshness") or classify_lease_freshness(
lease_for_class, now=now
)
owner_identity = (active.get("reviewer_identity") or "").strip()
owner_identity = (lease_for_class.get("reviewer_identity") or "").strip()
is_own = bool(session_id and owner and owner == session_id)
# Same identity alone is NOT ownership for resume; session_id must match.
same_identity = bool(
identity and owner_identity and identity == owner_identity
)
leased_head = _normalize_sha(lease_for_class.get("candidate_head"))
head_superseded = bool(
current_head and leased_head and current_head != leased_head
)
head_current = bool(
current_head and leased_head and current_head == leased_head
)
past_expiry = bool(lease_for_class.get("expired")) or freshness == "expired"
if not past_expiry:
past_expiry = _lease_expired(lease_for_class, now=now)
terminal = formal_terminal_review_for_head(
formal_reviews, leased_head=leased_head
)
if is_own and freshness in {"active", "stale_warning"}:
if is_own and freshness in {"active", "stale_warning"} and not past_expiry:
classification = "own_active"
next_action = NEXT_ACTION_RESUME_EXACT_OWNER_SESSION
elif is_own and freshness in {"reclaimable", "expired"}:
elif is_own and (freshness in {"reclaimable", "expired"} or past_expiry):
classification = "own_expired"
next_action = NEXT_ACTION_RELEASE_EXPIRED_LEASE
reasons.append(
f"own lease freshness is '{freshness}'; release via "
"gitea_release_reviewer_pr_lease then re-acquire"
)
elif not is_own and freshness in {"active", "stale_warning"}:
classification = "foreign_active"
elif not is_own and head_superseded and terminal and past_expiry:
classification = "foreign_expired_superseded_head"
next_action = NEXT_ACTION_CLEANUP_OBSOLETE_LEASE
reasons.append(
"foreign lease expired and pinned to superseded head with "
"formal terminal review; use guarded obsolete-lease cleanup"
)
elif not is_own and head_superseded and terminal and not past_expiry:
classification = "foreign_completed_superseded_head"
next_action = NEXT_ACTION_CLEANUP_OBSOLETE_LEASE
reasons.append(
"foreign lease pinned to superseded head with completed formal "
"review; use guarded obsolete-lease cleanup (do not wait indefinitely)"
)
elif not is_own and head_superseded and not terminal:
classification = "ambiguous_conflicting_evidence"
next_action = NEXT_ACTION_WAIT
reasons.append(
"lease head superseded but no formal terminal review for leased "
"head; fail closed / wait (no indefinite steal)"
)
elif not is_own and head_current and past_expiry:
classification = "foreign_expired_current_head"
if owner_process_alive is False and worktree_clean is True:
classification = "orphaned_owner_missing"
next_action = NEXT_ACTION_CLEANUP_OBSOLETE_LEASE
reasons.append(
"foreign expired lease on current head; owner process absent "
"and worktree clean — guarded orphan cleanup eligible"
)
elif owner_process_alive is False and worktree_clean is False:
classification = "ambiguous_conflicting_evidence"
next_action = NEXT_ACTION_WAIT
reasons.append(
"owner process absent but worktree dirty; cleanup denied"
)
else:
next_action = NEXT_ACTION_CLEANUP_OBSOLETE_LEASE
reasons.append(
"foreign expired lease on current head; use guarded cleanup "
"when worktree/process evidence permits"
)
elif not is_own and freshness in {"active", "stale_warning"} and not past_expiry:
classification = (
"foreign_active_current_head" if head_current or not current_head
else "foreign_active"
)
next_action = NEXT_ACTION_WAIT
reasons.append(
f"foreign active reviewer lease (session_id={owner}, "
f"phase={active.get('phase')}, freshness={freshness}); "
f"phase={lease_for_class.get('phase')}, freshness={freshness}); "
"do not submit; do not steal"
)
if same_identity:
@@ -674,11 +1232,12 @@ def diagnose_reviewer_pr_lease_handoff(
f"foreign reclaimable lease (session_id={owner}); clear only via "
"sanctioned gitea_release_reviewer_pr_lease when reclaimable"
)
elif not is_own and freshness == "expired":
elif not is_own and (freshness == "expired" or past_expiry):
classification = "foreign_expired"
next_action = NEXT_ACTION_RELEASE_EXPIRED_LEASE
reasons.append(
f"foreign expired lease (session_id={owner}); use sanctioned release"
f"foreign expired lease (session_id={owner}); use sanctioned release "
f"or {CLEANUP_OBSOLETE_LEASE_TOOL}"
)
else:
classification = "foreign_active"
@@ -691,13 +1250,26 @@ def diagnose_reviewer_pr_lease_handoff(
if instructed_missing_with_replacement and classification.startswith(
"foreign"
):
classification = "instructed_lease_missing_with_replacement"
# Foreign active still means wait; reclaimable still release.
if next_action == NEXT_ACTION_WAIT:
reasons.append(
"replacement foreign lease is active — wait; "
"operator_authorized_cleanup only with explicit operator authority"
)
# Preserve #691 cleanup path when superseded/expired; otherwise
# keep the PR #592 instructed-missing label.
if next_action != NEXT_ACTION_CLEANUP_OBSOLETE_LEASE:
classification = "instructed_lease_missing_with_replacement"
if next_action == NEXT_ACTION_WAIT:
reasons.append(
"replacement foreign lease is active — wait; "
"operator_authorized_cleanup only with explicit operator authority"
)
if next_action == NEXT_ACTION_CLEANUP_OBSOLETE_LEASE:
cleanup_hint = {
"cleanup_tool": CLEANUP_OBSOLETE_LEASE_TOOL,
"required_confirmation": cleanup_confirmation_for_pr(pr_number),
"controller_recovery_authorized_required": True,
"leased_head": leased_head,
"current_head": current_head,
"expires_at": lease_for_class.get("expires_at"),
"terminal_review_verdict": (terminal or {}).get("verdict"),
}
else:
classification = "no_lease"
next_action = NEXT_ACTION_ACQUIRE
@@ -720,29 +1292,55 @@ def diagnose_reviewer_pr_lease_handoff(
)
lease_summary = None
if active:
if lease_for_class:
lease_summary = {
"comment_id": active.get("comment_id"),
"session_id": active.get("session_id"),
"phase": active.get("phase"),
"candidate_head": active.get("candidate_head"),
"expires_at": active.get("expires_at"),
"last_activity": active.get("last_activity"),
"freshness": active.get("freshness")
or classify_lease_freshness(active, now=now),
"reviewer_identity": active.get("reviewer_identity"),
"profile": active.get("profile"),
"worktree": active.get("worktree"),
"blocker": active.get("blocker"),
"comment_id": lease_for_class.get("comment_id"),
"session_id": lease_for_class.get("session_id"),
"phase": lease_for_class.get("phase"),
"candidate_head": lease_for_class.get("candidate_head"),
"expires_at": lease_for_class.get("expires_at"),
"last_activity": lease_for_class.get("last_activity"),
"freshness": lease_for_class.get("freshness")
or classify_lease_freshness(lease_for_class, now=now),
"reviewer_identity": lease_for_class.get("reviewer_identity"),
"profile": lease_for_class.get("profile"),
"worktree": lease_for_class.get("worktree"),
"blocker": lease_for_class.get("blocker"),
}
return {
"pr_number": pr_number,
"classification": classification,
"blocker_kind": classification,
"next_action": next_action,
"exact_next_action": next_action,
"active_lease": lease_summary,
"session_lease": session_lease,
"worktree_binding": binding_details,
"leased_head": (lease_summary or {}).get("candidate_head"),
"current_head": current_head,
"expires_at": (lease_summary or {}).get("expires_at"),
"terminal_review_state": (
formal_terminal_review_for_head(
formal_reviews,
leased_head=(lease_summary or {}).get("candidate_head"),
)
if lease_summary
else None
),
"worktree_state": {
"exists": worktree_exists,
"clean": worktree_clean,
"path": (lease_summary or {}).get("worktree"),
},
"owner_session_evidence": {
"session_id": (lease_summary or {}).get("session_id"),
"process_alive": owner_process_alive,
"pid_is_not_ownership_proof": True,
},
"cleanup": cleanup_hint,
"cleanup_tool": (cleanup_hint or {}).get("cleanup_tool"),
"required_confirmation": (cleanup_hint or {}).get("required_confirmation"),
"instructed_session_id": instructed_session_id,
"instructed_comment_id": instructed_comment_id,
"instructed_lease_missing_with_replacement": instructed_missing_with_replacement,
@@ -751,6 +1349,19 @@ def diagnose_reviewer_pr_lease_handoff(
and not binding_mismatch
and bool(session_lease)
),
"mutation_eligibility": (
"allowed"
if (
next_action == NEXT_ACTION_RESUME_EXACT_OWNER_SESSION
and not binding_mismatch
and bool(session_lease)
)
else (
"cleanup_only"
if next_action == NEXT_ACTION_CLEANUP_OBSOLETE_LEASE
else "prohibited"
)
),
"reasons": reasons,
"forbidden": [
"manual lock deletion",
@@ -758,5 +1369,8 @@ def diagnose_reviewer_pr_lease_handoff(
"mtime manipulation",
"direct _SESSION_LEASE seeding",
"silent foreign lease steal",
"PID-based ownership",
"repointing old lease to new head",
"transfer of validation or decision state",
],
}
task_capability_map.py
+10
View File
@@ -76,6 +76,16 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = {
"permission": "gitea.pr.comment",
"role": "reviewer",
},
# #691: guarded non-owner cleanup of obsolete comment-backed reviewer leases.
# Apply path posts lease release + audit comments (gitea.pr.comment).
"cleanup_obsolete_reviewer_comment_lease": {
"permission": "gitea.pr.comment",
"role": "reviewer",
},
"gitea_cleanup_obsolete_reviewer_comment_lease": {
"permission": "gitea.pr.comment",
"role": "reviewer",
},
"blind_pr_queue_review": {
"permission": "gitea.pr.review",
"role": "reviewer",
tests/test_issue_691_obsolete_reviewer_lease_cleanup.py
+599
View File
@@ -0,0 +1,599 @@
"""Guarded cleanup for obsolete comment-backed reviewer leases (#691).
Reproduces PR #688 lease comment 10749 class of defect: completed
REQUEST_CHANGES on head A, author pushes head B, foreign lease remains
pinned to A (and after expiry still has no non-owner cleanup path that
diagnosis can prescribe beyond indefinite wait).
"""
from __future__ import annotations
import sys
import unittest
from datetime import datetime, timedelta, timezone
from pathlib import Path
from unittest.mock import patch
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
import merger_lease_adoption as mla # noqa: E402
import reviewer_pr_lease as leases # noqa: E402
# PR #688 / comment 10749 reproduction fixtures
HEAD_A = "c7a444eb4b41cf916fdbd20a4999ffd78af496d0"
HEAD_B = "4a6357800364718a27a36ebc73578d4b929ff4aa"
SESSION_FOREIGN = "25883-7d4c6e6ebd53"
COMMENT_10749 = 10749
REPO = "Scaled-Tech-Consulting/Gitea-Tools"
PR = 688
ISSUE = 687
EXPIRES_10749 = "2026-07-13T04:23:00Z"
def _utc(dt: datetime) -> datetime:
return dt if dt.tzinfo else dt.replace(tzinfo=timezone.utc)
def _lease_comment(
*,
pr_number: int = PR,
session_id: str = SESSION_FOREIGN,
phase: str = "claimed",
candidate_head: str = HEAD_A,
comment_id: int = COMMENT_10749,
last_activity: datetime | None = None,
expires_at: datetime | None = None,
worktree: str = "branches/review-pr688-feat-issue-687-reconciler-branch-delete",
repo: str = REPO,
) -> dict:
now = last_activity or datetime.now(timezone.utc)
body = leases.format_lease_body(
repo=repo,
pr_number=pr_number,
issue_number=ISSUE,
reviewer_identity="sysadmin",
profile="prgs-reviewer",
session_id=session_id,
worktree=worktree,
phase=phase,
candidate_head=candidate_head,
target_branch="master",
target_branch_sha="b" * 40,
last_activity=now,
expires_at=expires_at,
)
return {
"id": comment_id,
"body": body,
"user": {"login": "sysadmin"},
"author": "sysadmin",
"created_at": now.isoformat(),
"updated_at": now.isoformat(),
}
def _reviews_for_head(head: str, verdict: str = "REQUEST_CHANGES") -> list[dict]:
return [
{
"verdict": verdict,
"reviewed_head_sha": head,
"dismissed": False,
"reviewer": "sysadmin",
}
]
def _assess(
comments,
*,
current_head=HEAD_B,
formal_reviews=None,
controller=True,
worktree_exists=True,
worktree_clean=True,
owner_process_alive=False,
requesting_session="fresh-controller",
apply=False,
confirmation=None,
**kwargs,
):
return leases.assess_obsolete_reviewer_comment_lease_cleanup(
comments,
pr_number=PR,
current_head_sha=current_head,
formal_reviews=formal_reviews if formal_reviews is not None else _reviews_for_head(HEAD_A),
repo=REPO,
expected_repo=REPO,
requesting_session_id=requesting_session,
controller_recovery_authorized=controller,
worktree_exists=worktree_exists,
worktree_clean=worktree_clean,
worktree_has_unpreserved_work=not worktree_clean if worktree_exists else False,
owner_process_alive=owner_process_alive,
owner_pid_observed=kwargs.get("owner_pid_observed"),
requesting_pid=kwargs.get("requesting_pid", 99999),
current_head_review_in_progress=kwargs.get(
"current_head_review_in_progress", False
),
expected_lease_comment_id=kwargs.get("expected_lease_comment_id"),
expected_session_id=kwargs.get("expected_session_id"),
expected_leased_head=kwargs.get("expected_leased_head"),
confirmation=confirmation
if confirmation is not None
else (leases.cleanup_confirmation_for_pr(PR) if apply else ""),
apply=apply,
now=kwargs.get("now"),
)
class TestIssue691SupersededHeadCleanup(unittest.TestCase):
def setUp(self):
leases.clear_session_lease()
def test_1_request_changes_then_push_expired_cleanup_succeeds(self):
"""Completed REQUEST_CHANGES on A, push B, lease A expires → cleanup OK."""
now = datetime(2026, 7, 13, 5, 0, 0, tzinfo=timezone.utc)
expires = datetime(2026, 7, 13, 4, 23, 0, tzinfo=timezone.utc)
claim = _lease_comment(
last_activity=expires - timedelta(hours=2),
expires_at=expires,
)
result = _assess(
[claim],
formal_reviews=_reviews_for_head(HEAD_A, "REQUEST_CHANGES"),
now=now,
apply=True,
)
self.assertTrue(result["cleanup_allowed"], result["reasons"])
self.assertEqual(result["classification"], "foreign_expired_superseded_head")
self.assertEqual(
result["exact_next_action"], leases.NEXT_ACTION_CLEANUP_OBSOLETE_LEASE
)
self.assertIn("phase: released", result["release_body"])
self.assertIn("obsolete-superseded-or-expired-lease", result["release_body"])
self.assertIsNotNone(result["audit_comment_body"])
self.assertEqual(result["cleanup_tool"], leases.CLEANUP_OBSOLETE_LEASE_TOOL)
def test_2_approve_then_push_cleanup_policy(self):
"""Completed APPROVE on A, push B → cleanup eligible (completed superseded)."""
now = datetime(2026, 7, 13, 1, 0, 0, tzinfo=timezone.utc)
expires = datetime(2026, 7, 13, 4, 23, 0, tzinfo=timezone.utc) # not yet expired
claim = _lease_comment(
last_activity=now - timedelta(minutes=10),
expires_at=expires,
)
result = _assess(
[claim],
formal_reviews=_reviews_for_head(HEAD_A, "APPROVED"),
now=now,
)
self.assertTrue(result["cleanup_allowed"], result["reasons"])
self.assertEqual(result["classification"], "foreign_completed_superseded_head")
def test_3_active_current_head_cleanup_denied(self):
now = datetime.now(timezone.utc)
claim = _lease_comment(
candidate_head=HEAD_B,
last_activity=now - timedelta(minutes=5),
expires_at=now + timedelta(hours=2),
)
result = _assess(
[claim],
current_head=HEAD_B,
formal_reviews=_reviews_for_head(HEAD_B, "REQUEST_CHANGES"),
now=now,
)
self.assertFalse(result["cleanup_allowed"])
self.assertEqual(result["classification"], "foreign_active_current_head")
def test_4_foreign_owner_recent_progress_denied(self):
now = datetime.now(timezone.utc)
claim = _lease_comment(
candidate_head=HEAD_B,
last_activity=now - timedelta(minutes=2),
expires_at=now + timedelta(hours=2),
)
result = _assess(
[claim],
current_head=HEAD_B,
formal_reviews=[],
owner_process_alive=True,
now=now,
)
self.assertFalse(result["cleanup_allowed"])
self.assertTrue(
any("recent authenticated progress" in r for r in result["reasons"])
or any("current PR head" in r for r in result["reasons"])
)
def test_5_owner_absent_worktree_dirty_denied(self):
now = datetime(2026, 7, 13, 5, 0, 0, tzinfo=timezone.utc)
expires = datetime(2026, 7, 13, 4, 23, 0, tzinfo=timezone.utc)
claim = _lease_comment(
candidate_head=HEAD_B,
last_activity=expires - timedelta(hours=1),
expires_at=expires,
)
result = _assess(
[claim],
current_head=HEAD_B,
formal_reviews=[],
worktree_clean=False,
owner_process_alive=False,
now=now,
)
self.assertFalse(result["cleanup_allowed"])
self.assertTrue(any("dirty" in r for r in result["reasons"]))
def test_6_owner_absent_worktree_clean_orphan_ok(self):
now = datetime(2026, 7, 13, 5, 0, 0, tzinfo=timezone.utc)
expires = datetime(2026, 7, 13, 4, 23, 0, tzinfo=timezone.utc)
claim = _lease_comment(
candidate_head=HEAD_B,
last_activity=expires - timedelta(hours=1),
expires_at=expires,
)
result = _assess(
[claim],
current_head=HEAD_B,
formal_reviews=[],
worktree_clean=True,
owner_process_alive=False,
now=now,
)
self.assertTrue(result["cleanup_allowed"], result["reasons"])
self.assertEqual(result["classification"], "orphaned_owner_missing")
def test_7_pid_reuse_not_ownership(self):
now = datetime(2026, 7, 13, 5, 0, 0, tzinfo=timezone.utc)
expires = datetime(2026, 7, 13, 4, 23, 0, tzinfo=timezone.utc)
claim = _lease_comment(expires_at=expires, last_activity=expires - timedelta(hours=1))
result = _assess(
[claim],
now=now,
owner_pid_observed=4242,
requesting_pid=4242,
)
self.assertTrue(
any("PID equality" in r or "NOT ownership" in r for r in result["reasons"])
or result["owner_session_evidence"]["pid_is_not_ownership_proof"]
)
# Still eligible via superseded+terminal+expired despite matching PID.
self.assertTrue(result["cleanup_allowed"], result["reasons"])
def test_8_comment_backed_no_db_lease_id(self):
"""Cleanup uses comment ledger only — no control-plane lease_id required."""
now = datetime(2026, 7, 13, 5, 0, 0, tzinfo=timezone.utc)
expires = datetime(2026, 7, 13, 4, 23, 0, tzinfo=timezone.utc)
claim = _lease_comment(expires_at=expires, last_activity=expires - timedelta(hours=1))
# Explicitly no lease_id field anywhere.
self.assertNotIn("lease_id", claim)
result = _assess([claim], now=now)
self.assertTrue(result["cleanup_allowed"], result["reasons"])
self.assertEqual(result["active_lease"]["comment_id"], COMMENT_10749)
self.assertIsNone(result["active_lease"].get("lease_id"))
def test_9_cleanup_posts_durable_audit_bodies(self):
now = datetime(2026, 7, 13, 5, 0, 0, tzinfo=timezone.utc)
expires = datetime(2026, 7, 13, 4, 23, 0, tzinfo=timezone.utc)
claim = _lease_comment(expires_at=expires, last_activity=expires - timedelta(hours=1))
result = _assess([claim], now=now, apply=True)
self.assertTrue(result["cleanup_allowed"])
self.assertIn("#691", result["audit_comment_body"])
self.assertIn(str(COMMENT_10749), result["audit_comment_body"])
self.assertIn(HEAD_A, result["audit_comment_body"])
self.assertIn(HEAD_B, result["audit_comment_body"])
def test_10_fresh_acquire_after_cleanup_marker(self):
now = datetime(2026, 7, 13, 5, 0, 0, tzinfo=timezone.utc)
expires = datetime(2026, 7, 13, 4, 23, 0, tzinfo=timezone.utc)
claim = _lease_comment(
expires_at=expires,
last_activity=expires - timedelta(hours=1),
comment_id=COMMENT_10749,
)
assessed = _assess([claim], now=now, apply=True)
self.assertTrue(assessed["cleanup_allowed"])
release = {
"id": 99999,
"body": assessed["release_body"],
"user": {"login": "sysadmin"},
}
# Newest terminal marker ends active lease.
active = leases.find_active_reviewer_lease(
[claim, release], pr_number=PR, now=now
)
self.assertIsNone(active)
acq = leases.assess_acquire_lease(
[claim, release],
pr_number=PR,
reviewer_identity="sysadmin",
profile="prgs-reviewer",
session_id="fresh-reviewer-1",
repo=REPO,
issue_number=ISSUE,
worktree="branches/review-pr-688-fresh",
candidate_head=HEAD_B,
target_branch="master",
target_branch_sha="b" * 40,
now=now,
)
self.assertTrue(acq["acquire_allowed"], acq["reasons"])
def test_11_no_validation_state_transfer(self):
now = datetime(2026, 7, 13, 5, 0, 0, tzinfo=timezone.utc)
expires = datetime(2026, 7, 13, 4, 23, 0, tzinfo=timezone.utc)
claim = _lease_comment(expires_at=expires, last_activity=expires - timedelta(hours=1))
result = _assess([claim], now=now, apply=True)
self.assertTrue(result["cleanup_allowed"])
# Release body keeps old candidate_head (no repoint).
self.assertIn(f"candidate_head: {HEAD_A}", result["release_body"])
self.assertNotIn(HEAD_B, result["release_body"].split("candidate_head:")[1].split("\n")[0])
self.assertIn("did not transfer validation", result["audit_comment_body"])
self.assertIn("did not repoint", result["audit_comment_body"])
# Session lease must remain unset by assessment (tool never seeds).
self.assertIsNone(leases.get_session_lease())
def test_12_repeated_cleanup_idempotent(self):
now = datetime(2026, 7, 13, 5, 0, 0, tzinfo=timezone.utc)
expires = datetime(2026, 7, 13, 4, 23, 0, tzinfo=timezone.utc)
claim = _lease_comment(expires_at=expires, last_activity=expires - timedelta(hours=1))
first = _assess([claim], now=now, apply=True)
self.assertTrue(first["cleanup_allowed"])
release = {"id": 100001, "body": first["release_body"], "user": {"login": "x"}}
second = _assess([claim, release], now=now, apply=True)
self.assertFalse(second["cleanup_allowed"])
self.assertEqual(second["classification"], "no_lease")
def test_13_malformed_expiry_fails_closed(self):
claim = _lease_comment()
# Corrupt expires_at in the body.
claim["body"] = claim["body"].replace(
claim["body"].split("expires_at:")[1].split("\n")[0].strip(),
"not-a-timestamp",
)
result = _assess([claim], formal_reviews=_reviews_for_head(HEAD_A))
self.assertFalse(result["cleanup_allowed"])
self.assertFalse(result["expires_parseable"])
self.assertTrue(any("expires_at" in r for r in result["reasons"]))
def test_14_wrong_identity_evidence_fails_closed(self):
now = datetime(2026, 7, 13, 5, 0, 0, tzinfo=timezone.utc)
expires = datetime(2026, 7, 13, 4, 23, 0, tzinfo=timezone.utc)
claim = _lease_comment(expires_at=expires, last_activity=expires - timedelta(hours=1))
bad_repo = _assess(
[claim],
now=now,
# force repo mismatch via expected_repo
)
# Patch via direct call with wrong expected_repo
bad = leases.assess_obsolete_reviewer_comment_lease_cleanup(
[claim],
pr_number=PR,
current_head_sha=HEAD_B,
formal_reviews=_reviews_for_head(HEAD_A),
expected_repo="Other-Org/Other-Repo",
requesting_session_id="fresh",
controller_recovery_authorized=True,
worktree_exists=True,
worktree_clean=True,
owner_process_alive=False,
now=now,
)
self.assertFalse(bad["cleanup_allowed"])
self.assertTrue(any("repository identity" in r for r in bad["reasons"]))
bad_pr = leases.assess_obsolete_reviewer_comment_lease_cleanup(
[claim],
pr_number=999,
current_head_sha=HEAD_B,
formal_reviews=_reviews_for_head(HEAD_A),
expected_repo=REPO,
requesting_session_id="fresh",
controller_recovery_authorized=True,
worktree_exists=True,
worktree_clean=True,
owner_process_alive=False,
expected_lease_comment_id=COMMENT_10749,
now=now,
)
self.assertFalse(bad_pr["cleanup_allowed"])
bad_head = _assess(
[claim],
now=now,
expected_leased_head="0" * 40,
)
self.assertFalse(bad_head["cleanup_allowed"])
bad_session = _assess(
[claim],
now=now,
expected_session_id="wrong-session",
)
self.assertFalse(bad_session["cleanup_allowed"])
def test_15_current_head_active_cannot_be_displaced(self):
now = datetime.now(timezone.utc)
claim = _lease_comment(
candidate_head=HEAD_B,
last_activity=now - timedelta(minutes=1),
expires_at=now + timedelta(hours=2),
)
result = _assess(
[claim],
current_head=HEAD_B,
formal_reviews=_reviews_for_head(HEAD_B),
now=now,
current_head_review_in_progress=True,
)
self.assertFalse(result["cleanup_allowed"])
# Acquire also blocked by foreign active lease.
acq = leases.assess_acquire_lease(
[claim],
pr_number=PR,
reviewer_identity="other",
profile="prgs-reviewer",
session_id="thief",
repo=REPO,
issue_number=ISSUE,
worktree="branches/steal",
candidate_head=HEAD_B,
target_branch="master",
target_branch_sha="b" * 40,
now=now,
)
self.assertFalse(acq["acquire_allowed"])
def test_regression_pr688_comment_10749_after_expiry(self):
"""Exact 10749 reproduction after recorded expires_at 2026-07-13T04:23:00Z."""
now = datetime(2026, 7, 13, 4, 30, 0, tzinfo=timezone.utc)
expires = datetime.fromisoformat(EXPIRES_10749.replace("Z", "+00:00"))
claim = _lease_comment(
session_id=SESSION_FOREIGN,
candidate_head=HEAD_A,
comment_id=COMMENT_10749,
last_activity=expires - timedelta(hours=2),
expires_at=expires,
)
# Diagnosis must not prescribe indefinite wait-only for this case.
diag = leases.diagnose_reviewer_pr_lease_handoff(
[claim],
pr_number=PR,
current_session_id="fresh-reviewer",
current_reviewer_identity="sysadmin",
current_head_sha=HEAD_B,
formal_reviews=_reviews_for_head(HEAD_A, "REQUEST_CHANGES"),
worktree_exists=True,
worktree_clean=True,
owner_process_alive=False,
instructed_session_id=SESSION_FOREIGN,
instructed_comment_id=COMMENT_10749,
now=now,
)
self.assertEqual(diag["classification"], "foreign_expired_superseded_head")
self.assertEqual(
diag["next_action"], leases.NEXT_ACTION_CLEANUP_OBSOLETE_LEASE
)
self.assertEqual(diag["cleanup_tool"], leases.CLEANUP_OBSOLETE_LEASE_TOOL)
self.assertIn("CLEANUP OBSOLETE REVIEWER LEASE 688", diag["required_confirmation"])
self.assertEqual(diag["mutation_eligibility"], "cleanup_only")
self.assertFalse(diag["mutation_allowed"])
# Assessment still returns the lease as active/stale class of block for acquire
# when unexpired path... here it's expired so find_active is None; acquire free
# only after cleanup if somehow still active. With expired, find_active is None:
active = leases.find_active_reviewer_lease([claim], pr_number=PR, now=now)
self.assertIsNone(active)
# But superseded unexpired still blocks acquire and diagnosis points to cleanup:
unexpired_now = datetime(2026, 7, 13, 1, 0, 0, tzinfo=timezone.utc)
claim2 = _lease_comment(
session_id=SESSION_FOREIGN,
candidate_head=HEAD_A,
comment_id=COMMENT_10749,
last_activity=unexpired_now - timedelta(minutes=45),
expires_at=expires,
)
active2 = leases.find_active_reviewer_lease(
[claim2], pr_number=PR, now=unexpired_now
)
self.assertIsNotNone(active2)
self.assertEqual(active2["freshness"], "stale_warning")
diag2 = leases.diagnose_reviewer_pr_lease_handoff(
[claim2],
pr_number=PR,
current_session_id="fresh-reviewer",
current_reviewer_identity="sysadmin",
current_head_sha=HEAD_B,
formal_reviews=_reviews_for_head(HEAD_A, "REQUEST_CHANGES"),
worktree_exists=True,
worktree_clean=True,
now=unexpired_now,
)
self.assertEqual(diag2["classification"], "foreign_completed_superseded_head")
self.assertEqual(
diag2["next_action"], leases.NEXT_ACTION_CLEANUP_OBSOLETE_LEASE
)
acq = leases.assess_acquire_lease(
[claim2],
pr_number=PR,
reviewer_identity="sysadmin",
profile="prgs-reviewer",
session_id="fresh-reviewer",
repo=REPO,
issue_number=ISSUE,
worktree="branches/review-pr-688-new",
candidate_head=HEAD_B,
target_branch="master",
target_branch_sha="b" * 40,
now=unexpired_now,
)
self.assertFalse(acq["acquire_allowed"])
def test_missing_controller_auth_denied(self):
now = datetime(2026, 7, 13, 5, 0, 0, tzinfo=timezone.utc)
expires = datetime(2026, 7, 13, 4, 23, 0, tzinfo=timezone.utc)
claim = _lease_comment(expires_at=expires, last_activity=expires - timedelta(hours=1))
result = _assess([claim], controller=False, now=now)
self.assertFalse(result["cleanup_allowed"])
def test_wrong_confirmation_denied_on_apply(self):
now = datetime(2026, 7, 13, 5, 0, 0, tzinfo=timezone.utc)
expires = datetime(2026, 7, 13, 4, 23, 0, tzinfo=timezone.utc)
claim = _lease_comment(expires_at=expires, last_activity=expires - timedelta(hours=1))
result = _assess(
[claim], now=now, apply=True, confirmation="MERGE PR 688"
)
self.assertFalse(result["cleanup_allowed"])
self.assertTrue(any("confirmation" in r for r in result["reasons"]))
def test_owner_session_must_use_owner_release(self):
now = datetime(2026, 7, 13, 5, 0, 0, tzinfo=timezone.utc)
expires = datetime(2026, 7, 13, 4, 23, 0, tzinfo=timezone.utc)
claim = _lease_comment(expires_at=expires, last_activity=expires - timedelta(hours=1))
result = _assess(
[claim],
now=now,
requesting_session=SESSION_FOREIGN,
)
self.assertFalse(result["cleanup_allowed"])
self.assertTrue(any("owns the lease" in r for r in result["reasons"]))
def test_superseded_without_terminal_review_denied(self):
now = datetime(2026, 7, 13, 5, 0, 0, tzinfo=timezone.utc)
expires = datetime(2026, 7, 13, 4, 23, 0, tzinfo=timezone.utc)
claim = _lease_comment(expires_at=expires, last_activity=expires - timedelta(hours=1))
result = _assess([claim], formal_reviews=[], now=now)
self.assertFalse(result["cleanup_allowed"])
self.assertEqual(result["classification"], "ambiguous_conflicting_evidence")
class TestIssue691DiagnoseClassifications(unittest.TestCase):
def setUp(self):
leases.clear_session_lease()
def test_foreign_active_current_head_still_wait(self):
now = datetime.now(timezone.utc)
claim = _lease_comment(
candidate_head=HEAD_B,
last_activity=now - timedelta(minutes=5),
expires_at=now + timedelta(hours=1),
)
claim["id"] = 1
result = leases.diagnose_reviewer_pr_lease_handoff(
[claim],
pr_number=PR,
current_session_id="other",
current_reviewer_identity="sysadmin",
current_head_sha=HEAD_B,
formal_reviews=[],
now=now,
)
self.assertEqual(result["classification"], "foreign_active_current_head")
self.assertEqual(result["next_action"], leases.NEXT_ACTION_WAIT)
self.assertFalse(result["mutation_allowed"])
if __name__ == "__main__":
unittest.main()