"""First-class control-plane lease lifecycle (#601). Active leases are queryable workflow state. Canonical operations: * list / inspect * adopt (with provenance) * release (explicit, recorded) * expire / reclaim * abandon (requires proof: dead process and/or missing worktree, etc.) Authority model: * Control-plane DB is the coordination source for assignment/lease. * File locks and comment-only leases are **not** authoritative alone. * Reviewer/merger comment leases (``reviewer_pr_lease`` / merger adoption) remain for Gitea-thread durability; they must not bypass DB assignment when the control-plane path is in use. Fail closed on ambiguous ownership, active foreign leases, mismatched worktree, stale head, missing capability, and unsafe cleanup. """ from __future__ import annotations import json import os from dataclasses import dataclass, field from datetime import datetime, timezone from typing import Any, Callable, Mapping import control_plane_db as cpd # Outcomes / safe next actions (stable vocabulary for tools + tests). SAFE_OWNER_RESUME = "owner_resume" SAFE_WAIT_FOREIGN = "wait_foreign_active" SAFE_RECLAIM_EXPIRED = "reclaim_expired" SAFE_ABANDON_ALLOWED = "abandon_allowed" SAFE_RELEASE_OWNED = "release_owned" SAFE_STALE_PROMPT = "stale_prompt_lease" SAFE_UNKNOWN = "inspect_only" SAFE_NO_AUTHORITY = "file_or_comment_not_authoritative" LEASE_STATUS_ACTIVE = "active" LEASE_STATUS_RELEASED = "released" LEASE_STATUS_EXPIRED = "expired" LEASE_STATUS_ABANDONED = "abandoned" AUTHORITATIVE_SOURCE = "control_plane_db" class LeaseLifecycleError(RuntimeError): """Fail-closed lifecycle policy error.""" @dataclass(frozen=True) class AbandonProof: """Required evidence to abandon a non-owned or expired lease safely.""" dead_process: bool = False missing_worktree: bool = False same_owner: bool = False no_open_pr: bool = False no_live_mutation_risk: bool = False operator_authorized: bool = False worktree_path: str | None = None owner_pid: int | None = None notes: str = "" def as_dict(self) -> dict[str, Any]: return { "dead_process": self.dead_process, "missing_worktree": self.missing_worktree, "same_owner": self.same_owner, "no_open_pr": self.no_open_pr, "no_live_mutation_risk": self.no_live_mutation_risk, "operator_authorized": self.operator_authorized, "worktree_path": self.worktree_path, "owner_pid": self.owner_pid, "notes": self.notes, } def is_sufficient(self) -> bool: """Abandon requires process death or missing worktree + no mutation risk. Foreign active leases additionally need operator_authorized unless the owner process is dead and the worktree is missing. """ if not self.no_live_mutation_risk: return False if not (self.dead_process or self.missing_worktree): return False if self.same_owner: return True # Foreign abandon: operator flag OR (dead + missing worktree + no risk) if self.operator_authorized: return True return bool(self.dead_process and self.missing_worktree and self.no_open_pr) def is_process_alive(pid: int | None) -> bool: if pid is None: return False try: pid_i = int(pid) except (TypeError, ValueError): return False if pid_i <= 0: return False try: os.kill(pid_i, 0) return True except ProcessLookupError: return False except PermissionError: # Exists but not owned by us — treat as alive (fail closed). return True except OSError: return False def worktree_exists(path: str | None) -> bool: if not path: return False try: return os.path.isdir(os.path.realpath(path)) except OSError: return False def _utc_now() -> datetime: return datetime.now(timezone.utc) def _parse_ts(value: str | None) -> datetime | None: return cpd._parse_ts(value) def classify_lease_freshness( lease: Mapping[str, Any], *, now: datetime | None = None, pid_checker: Callable[[int | None], bool] = is_process_alive, worktree_checker: Callable[[str | None], bool] = worktree_exists, ) -> dict[str, Any]: """Classify a control-plane lease row for workflow tooling.""" moment = now or _utc_now() status = (lease.get("status") or "").strip().lower() expires = _parse_ts(lease.get("expires_at")) owner_pid = lease.get("owner_pid") if owner_pid is None: owner_pid = lease.get("session_pid") wt = lease.get("worktree_path") pid_alive = pid_checker(owner_pid) if owner_pid is not None else None wt_present = worktree_checker(wt) if wt else None expired_by_time = bool(expires and expires <= moment) if status == LEASE_STATUS_ABANDONED: freshness = "abandoned" elif status == LEASE_STATUS_RELEASED: freshness = "released" elif status == LEASE_STATUS_EXPIRED or expired_by_time: freshness = "expired" elif status != LEASE_STATUS_ACTIVE: freshness = status or "unknown" elif pid_alive is False: freshness = "stale_dead_process" elif wt is not None and wt_present is False: freshness = "stale_missing_worktree" else: freshness = "active" return { "freshness": freshness, "status": status, "expired_by_time": expired_by_time, "owner_pid": owner_pid, "owner_pid_alive": pid_alive, "worktree_path": wt, "worktree_present": wt_present, "expires_at": lease.get("expires_at"), "authoritative_source": AUTHORITATIVE_SOURCE, "file_lock_only": False, "comment_lease_only": False, } def decide_safe_next_action( *, lease: Mapping[str, Any] | None, caller_session_id: str, freshness: Mapping[str, Any] | None = None, caller_worktree: str | None = None, ) -> dict[str, Any]: """Return safe_next_action + reasons for a caller inspecting a lease.""" if not lease: return { "safe_next_action": SAFE_STALE_PROMPT, "reasons": ["lease not found in control-plane DB (stale prompt id?)"], "block": True, } fr = freshness or classify_lease_freshness(lease) owner = str(lease.get("session_id") or "") same_owner = owner == str(caller_session_id) status = fr.get("freshness") reasons: list[str] = [] # Worktree mismatch for owner resume lease_wt = lease.get("worktree_path") if ( same_owner and lease_wt and caller_worktree and os.path.realpath(str(lease_wt)) != os.path.realpath(str(caller_worktree)) ): return { "safe_next_action": SAFE_UNKNOWN, "reasons": [ f"worktree mismatch: lease has {lease_wt!r}, caller has " f"{caller_worktree!r} (fail closed)" ], "block": True, "same_owner": True, } if status in ("abandoned", "released"): return { "safe_next_action": SAFE_STALE_PROMPT, "reasons": [f"lease status is {status}; do not adopt blindly"], "block": True, "same_owner": same_owner, } if status == "expired" or fr.get("expired_by_time"): return { "safe_next_action": SAFE_RECLAIM_EXPIRED, "reasons": ["lease expired; reclaim via expire+assign or adopt reclaim path"], "block": False, "same_owner": same_owner, } if status in ("stale_dead_process", "stale_missing_worktree"): if same_owner: return { "safe_next_action": SAFE_OWNER_RESUME, "reasons": [ f"caller owns lease with freshness={status}; rebind via " "adopt/owner-resume or abandon with proof" ], "block": False, "same_owner": True, "also_allowed": [SAFE_ABANDON_ALLOWED, SAFE_RELEASE_OWNED], } return { "safe_next_action": SAFE_ABANDON_ALLOWED, "reasons": [ f"lease freshness={status}; abandon with required proof then reassign" ], "block": False, "same_owner": False, } if same_owner and status == "active": return { "safe_next_action": SAFE_OWNER_RESUME, "reasons": [ "caller owns active lease; resume via heartbeat/assign owner-resume " "or release explicitly" ], "block": False, "same_owner": True, "also_allowed": [SAFE_RELEASE_OWNED], } if not same_owner and status == "active": return { "safe_next_action": SAFE_WAIT_FOREIGN, "reasons": [ f"foreign active lease held by session {owner}; " "do not steal (fail closed)" ], "block": True, "same_owner": False, "owner_session_id": owner, } return { "safe_next_action": SAFE_UNKNOWN, "reasons": [f"unclassified freshness={status}"], "block": True, "same_owner": same_owner, } def non_db_lease_authority_report( *, file_lock_present: bool = False, comment_lease_present: bool = False, db_lease_present: bool = False, ) -> dict[str, Any]: """File/comment leases alone are not coordination authority (#601 / #600).""" authoritative = bool(db_lease_present) reasons: list[str] = [] if file_lock_present and not db_lease_present: reasons.append( "file lock present but control-plane DB lease absent — " "file lock is not authoritative alone" ) if comment_lease_present and not db_lease_present: reasons.append( "comment-only lease present but control-plane DB lease absent — " "comment lease is not authoritative alone" ) if authoritative: reasons.append("control-plane DB lease is the coordination authority") return { "authoritative_source": AUTHORITATIVE_SOURCE if authoritative else None, "db_lease_present": db_lease_present, "file_lock_present": file_lock_present, "comment_lease_present": comment_lease_present, "safe_next_action": ( SAFE_UNKNOWN if authoritative else SAFE_NO_AUTHORITY ), "reasons": reasons, "file_lock_only": bool(file_lock_present and not db_lease_present), "comment_lease_only": bool(comment_lease_present and not db_lease_present), } def build_adopt_provenance( *, adopted_from_session_id: str, adopted_by_session_id: str, work_kind: str, work_number: int, remote: str, org: str, repo: str, worktree_path: str | None, expected_head_sha: str | None, prior_lease_id: str | None, reason: str, ) -> dict[str, Any]: return { "adopted_from_session_id": adopted_from_session_id, "adopted_by_session_id": adopted_by_session_id, "work_kind": work_kind, "work_number": int(work_number), "remote": remote, "org": org, "repo": repo, "worktree_path": worktree_path, "expected_head_sha": expected_head_sha, "prior_lease_id": prior_lease_id, "reason": reason, "recorded_at": cpd._ts(), "source": "lease_lifecycle.adopt", } def inspect_lease( db: cpd.ControlPlaneDB, lease_id: str, *, caller_session_id: str, caller_worktree: str | None = None, ) -> dict[str, Any]: """Full first-class lease workflow state for one lease id.""" state = db.get_lease_workflow_state(lease_id) if not state: decision = decide_safe_next_action( lease=None, caller_session_id=caller_session_id ) return { "success": True, "found": False, "lease_id": lease_id, "lease": None, "freshness": None, **decision, "authoritative_source": AUTHORITATIVE_SOURCE, "file_lock_only": False, "comment_lease_only": False, } lease = state["lease"] freshness = classify_lease_freshness(lease) decision = decide_safe_next_action( lease=lease, caller_session_id=caller_session_id, freshness=freshness, caller_worktree=caller_worktree, ) return { "success": True, "found": True, "lease_id": lease_id, "lease": lease, "assignment": state.get("assignment"), "work_item": state.get("work_item"), "session": state.get("session"), "freshness": freshness, "provenance": state.get("provenance"), **decision, "authoritative_source": AUTHORITATIVE_SOURCE, "file_lock_only": False, "comment_lease_only": False, } def list_active_leases( db: cpd.ControlPlaneDB, *, remote: str | None = None, org: str | None = None, repo: str | None = None, role: str | None = None, include_non_active: bool = False, limit: int = 100, ) -> dict[str, Any]: statuses = None if include_non_active else (LEASE_STATUS_ACTIVE,) rows = db.list_leases( remote=remote, org=org, repo=repo, role=role, statuses=statuses, limit=limit, ) enriched = [] for row in rows: fr = classify_lease_freshness(row) enriched.append({**row, "freshness": fr}) return { "success": True, "count": len(enriched), "leases": enriched, "authoritative_source": AUTHORITATIVE_SOURCE, "file_lock_only": False, "comment_lease_only": False, "include_non_active": include_non_active, } def adopt_lease( db: cpd.ControlPlaneDB, *, lease_id: str, adopter_session_id: str, role: str, worktree_path: str | None = None, expected_head_sha: str | None = None, owner_pid: int | None = None, operator_authorized: bool = False, ) -> dict[str, Any]: """Sanctioned adopt path with provenance; never silent foreign steal.""" state = db.get_lease_workflow_state(lease_id) if not state: raise LeaseLifecycleError( f"unknown lease_id {lease_id}; stale prompt id (fail closed)" ) lease = state["lease"] work = state["work_item"] freshness = classify_lease_freshness(lease) owner = str(lease.get("session_id") or "") same_owner = owner == str(adopter_session_id) if freshness["freshness"] == "active" and not same_owner: raise LeaseLifecycleError( f"refusing to steal active foreign lease {lease_id} owned by " f"{owner} (fail closed)" ) if freshness["freshness"] in ("abandoned", "released"): raise LeaseLifecycleError( f"lease {lease_id} is {freshness['freshness']}; cannot adopt " "(fail closed)" ) # Expired or stale: require abandon-style safety before ownership transfer # when not same owner; same owner may reclaim. if not same_owner and freshness["freshness"] in ( "expired", "stale_dead_process", "stale_missing_worktree", ): if not operator_authorized and freshness["freshness"] == "expired": # Deterministic reclaim of expired foreign lease is allowed # without operator flag (sanctioned expire reclaim). pass elif freshness["freshness"] != "expired" and not operator_authorized: # stale active-looking requires abandon proof path raise LeaseLifecycleError( f"lease {lease_id} freshness={freshness['freshness']}; " "use abandon with proof before foreign adopt (fail closed)" ) provenance = build_adopt_provenance( adopted_from_session_id=owner, adopted_by_session_id=adopter_session_id, work_kind=str(work.get("kind")), work_number=int(work.get("number")), remote=str(work.get("remote")), org=str(work.get("org")), repo=str(work.get("repo")), worktree_path=worktree_path, expected_head_sha=expected_head_sha or lease.get("expected_head_sha"), prior_lease_id=lease_id, reason=( "owner-resume-adopt" if same_owner else "sanctioned-reclaim-adopt" ), ) result = db.adopt_lease( lease_id=lease_id, adopter_session_id=adopter_session_id, role=role, worktree_path=worktree_path, expected_head_sha=expected_head_sha, owner_pid=owner_pid if owner_pid is not None else os.getpid(), provenance=provenance, ) return { "success": True, "outcome": result.get("outcome"), "same_owner": same_owner, "prior_lease_id": lease_id, "lease": result.get("lease"), "assignment": result.get("assignment"), "provenance": provenance, "authoritative_source": AUTHORITATIVE_SOURCE, "file_lock_only": False, "comment_lease_only": False, "reasons": result.get("reasons") or [], } def release_lease( db: cpd.ControlPlaneDB, *, lease_id: str, session_id: str, ) -> dict[str, Any]: proof = db.release_lease_recorded(lease_id=lease_id, session_id=session_id) return { "success": True, "outcome": "released", "lease_id": lease_id, "session_id": session_id, "release_proof": proof, "authoritative_source": AUTHORITATIVE_SOURCE, "file_lock_only": False, "comment_lease_only": False, } def expire_leases(db: cpd.ControlPlaneDB) -> dict[str, Any]: n = db.expire_stale_leases() return { "success": True, "outcome": "expired", "expired_count": int(n), "authoritative_source": AUTHORITATIVE_SOURCE, "file_lock_only": False, "comment_lease_only": False, } def reclaim_expired_lease( db: cpd.ControlPlaneDB, *, lease_id: str, session_id: str, role: str, worktree_path: str | None = None, expected_head_sha: str | None = None, ) -> dict[str, Any]: """Expire-if-needed then assign to *session_id* for the same work item.""" state = db.get_lease_workflow_state(lease_id) if not state: raise LeaseLifecycleError(f"unknown lease_id {lease_id}") lease = state["lease"] work = state["work_item"] freshness = classify_lease_freshness(lease) if freshness["freshness"] == "active": if lease.get("session_id") != session_id: raise LeaseLifecycleError( f"cannot reclaim active foreign lease {lease_id} (fail closed)" ) # owner resume return adopt_lease( db, lease_id=lease_id, adopter_session_id=session_id, role=role, worktree_path=worktree_path, expected_head_sha=expected_head_sha, ) if freshness["freshness"] not in ( "expired", "stale_dead_process", "stale_missing_worktree", "abandoned", "released", ): raise LeaseLifecycleError( f"lease {lease_id} freshness={freshness['freshness']} not reclaimable" ) # Ensure expired marker is applied db.expire_stale_leases() # If still active due to clock skew / non-time stale, force expire this lease refreshed = db.get_lease_workflow_state(lease_id) if refreshed and refreshed["lease"].get("status") == "active": db.force_expire_lease(lease_id, reason="reclaim_expired_lease") head = expected_head_sha or work.get("current_head_sha") assigned = db.assign_and_lease( session_id=session_id, role=role, remote=str(work["remote"]), org=str(work["org"]), repo=str(work["repo"]), kind=str(work["kind"]), number=int(work["number"]), expected_head_sha=head, phase="reclaimed", worktree_path=worktree_path, owner_pid=os.getpid(), ) if assigned.outcome != "assigned": raise LeaseLifecycleError( f"reclaim assign failed: {assigned.outcome} — {assigned.reason}" ) provenance = build_adopt_provenance( adopted_from_session_id=str(lease.get("session_id")), adopted_by_session_id=session_id, work_kind=str(work["kind"]), work_number=int(work["number"]), remote=str(work["remote"]), org=str(work["org"]), repo=str(work["repo"]), worktree_path=worktree_path, expected_head_sha=head, prior_lease_id=lease_id, reason="reclaim_expired", ) db.attach_lease_provenance(assigned.lease_id, provenance) return { "success": True, "outcome": "reclaimed", "prior_lease_id": lease_id, "assignment": assigned.as_dict(), "provenance": provenance, "authoritative_source": AUTHORITATIVE_SOURCE, "file_lock_only": False, "comment_lease_only": False, } def abandon_lease( db: cpd.ControlPlaneDB, *, lease_id: str, requester_session_id: str, proof: AbandonProof, ) -> dict[str, Any]: state = db.get_lease_workflow_state(lease_id) if not state: raise LeaseLifecycleError(f"unknown lease_id {lease_id}") lease = state["lease"] owner = str(lease.get("session_id") or "") same_owner = owner == str(requester_session_id) # Enrich proof with live checks when not provided owner_pid = proof.owner_pid if owner_pid is None: owner_pid = lease.get("owner_pid") wt = proof.worktree_path if proof.worktree_path is not None else lease.get( "worktree_path" ) dead = proof.dead_process or ( owner_pid is not None and not is_process_alive(owner_pid) ) missing_wt = proof.missing_worktree or ( bool(wt) and not worktree_exists(str(wt)) ) enriched = AbandonProof( dead_process=bool(dead), missing_worktree=bool(missing_wt), same_owner=same_owner or proof.same_owner, no_open_pr=proof.no_open_pr, no_live_mutation_risk=proof.no_live_mutation_risk, operator_authorized=proof.operator_authorized, worktree_path=str(wt) if wt else None, owner_pid=int(owner_pid) if owner_pid is not None else None, notes=proof.notes, ) if not enriched.is_sufficient(): raise LeaseLifecycleError( "abandon proof insufficient: need (dead_process or missing_worktree) " "and no_live_mutation_risk; foreign abandon also needs " "operator_authorized or (dead+missing_worktree+no_open_pr). " f"proof={enriched.as_dict()}" ) # Active foreign with live process + present worktree: never abandon without # operator (already covered by is_sufficient). result = db.abandon_lease( lease_id=lease_id, requester_session_id=requester_session_id, proof=enriched.as_dict(), ) return { "success": True, "outcome": "abandoned", "lease_id": lease_id, "requester_session_id": requester_session_id, "prior_owner_session_id": owner, "abandon_proof": enriched.as_dict(), "audit": result, "authoritative_source": AUTHORITATIVE_SOURCE, "file_lock_only": False, "comment_lease_only": False, }