Review #515 F1: gitea_adopt_workflow_lease trusted a caller-supplied role ((role or active_role)), so any namespace holding gitea.read could consume an author-only cross-role handoff by passing role="author". - Derive the adopter role authoritatively from the active profile; reject any supplied role that does not exactly match (no silent accept). - Pass the profile-derived role and authoritative profile/namespace context to lease_lifecycle.adopt_lease; validate handoff provenance required_profile/required_namespace against it (fail closed). - Fail closed when the profile role cannot be derived (no author default). - Add MCP-boundary regression tests: reviewer/merger profiles cannot consume an author handoff via role="author"; the legitimate author profile still consumes; foreign required_profile rejected.
920 lines
32 KiB
Python
920 lines
32 KiB
Python
"""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"
|
|
SAFE_CONSUME_CROSS_ROLE = "consume_cross_role_handoff"
|
|
|
|
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],
|
|
}
|
|
handoff = is_pending_cross_role_handoff({"lease": lease})
|
|
if handoff:
|
|
return {
|
|
"safe_next_action": SAFE_CONSUME_CROSS_ROLE,
|
|
"reasons": [
|
|
f"controller allocation pending handoff (freshness={status}); "
|
|
"required-role worker may consume without abandon/reassign; "
|
|
f"required_role={handoff['required_role']}"
|
|
],
|
|
"block": False,
|
|
"same_owner": False,
|
|
"owner_session_id": owner,
|
|
"required_role": handoff["required_role"],
|
|
"cross_role_handoff": True,
|
|
"handoff_status": "pending",
|
|
"also_allowed": [SAFE_ABANDON_ALLOWED],
|
|
}
|
|
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":
|
|
# #843: pending cross-role handoff is consumable by required role
|
|
handoff = is_pending_cross_role_handoff({"lease": lease})
|
|
if handoff:
|
|
return {
|
|
"safe_next_action": SAFE_CONSUME_CROSS_ROLE,
|
|
"reasons": [
|
|
"controller cross-role allocation pending handoff; "
|
|
f"required_role={handoff['required_role']}; "
|
|
"consume via gitea_adopt_workflow_lease without "
|
|
"abandonment or sharing the controller session"
|
|
],
|
|
"block": False,
|
|
"same_owner": False,
|
|
"owner_session_id": owner,
|
|
"required_role": handoff["required_role"],
|
|
"cross_role_handoff": True,
|
|
"handoff_status": "pending",
|
|
}
|
|
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 parse_lease_provenance(lease_or_state: Mapping[str, Any] | None) -> dict[str, Any]:
|
|
"""Return durable lease provenance dict (empty when absent/unparseable)."""
|
|
if not lease_or_state:
|
|
return {}
|
|
if "provenance" in lease_or_state and isinstance(lease_or_state.get("provenance"), dict):
|
|
return dict(lease_or_state["provenance"])
|
|
raw = None
|
|
if "provenance_json" in lease_or_state:
|
|
raw = lease_or_state.get("provenance_json")
|
|
elif "lease" in lease_or_state and isinstance(lease_or_state.get("lease"), Mapping):
|
|
raw = lease_or_state["lease"].get("provenance_json")
|
|
if not raw:
|
|
return {}
|
|
if isinstance(raw, dict):
|
|
return dict(raw)
|
|
try:
|
|
loaded = json.loads(raw)
|
|
except (TypeError, json.JSONDecodeError):
|
|
return {}
|
|
return dict(loaded) if isinstance(loaded, dict) else {}
|
|
|
|
|
|
def is_pending_cross_role_handoff(
|
|
state: Mapping[str, Any] | None,
|
|
) -> dict[str, Any] | None:
|
|
"""Return handoff evidence when a controller allocation awaits consume (#843).
|
|
|
|
A pending handoff is identified by durable provenance written at
|
|
cross-role apply time — not by title heuristics or session-id guessing.
|
|
"""
|
|
if not state:
|
|
return None
|
|
lease = state.get("lease") if isinstance(state.get("lease"), Mapping) else state
|
|
if not isinstance(lease, Mapping):
|
|
return None
|
|
status = str(lease.get("status") or "").strip().lower()
|
|
if status in (LEASE_STATUS_ABANDONED, LEASE_STATUS_RELEASED, LEASE_STATUS_EXPIRED):
|
|
return None
|
|
prov = parse_lease_provenance(state)
|
|
if not prov and isinstance(lease, Mapping):
|
|
prov = parse_lease_provenance(lease)
|
|
if not prov.get("cross_role_handoff"):
|
|
return None
|
|
handoff_status = str(prov.get("handoff_status") or "pending").strip().lower()
|
|
if handoff_status != "pending":
|
|
return None
|
|
adopted_by = (
|
|
lease.get("adopted_by_session_id")
|
|
or prov.get("adopted_by_session_id")
|
|
or ""
|
|
)
|
|
if str(adopted_by).strip():
|
|
return None
|
|
required_role = str(
|
|
prov.get("required_role") or lease.get("role") or ""
|
|
).strip().lower()
|
|
if not required_role:
|
|
return None
|
|
return {
|
|
"cross_role_handoff": True,
|
|
"handoff_status": "pending",
|
|
"required_role": required_role,
|
|
"allocating_session_id": str(
|
|
prov.get("allocating_session_id") or lease.get("session_id") or ""
|
|
),
|
|
"allocating_role": str(prov.get("allocating_role") or "controller"),
|
|
"lease_id": str(lease.get("lease_id") or ""),
|
|
"assignment_id": (
|
|
str(state["assignment"]["assignment_id"])
|
|
if isinstance(state.get("assignment"), Mapping)
|
|
and state["assignment"].get("assignment_id")
|
|
else None
|
|
),
|
|
"provenance": prov,
|
|
}
|
|
|
|
|
|
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,
|
|
adopter_profile_name: str | None = None,
|
|
adopter_namespace: str | None = None,
|
|
) -> dict[str, Any]:
|
|
"""Sanctioned adopt path with provenance; never silent foreign steal.
|
|
|
|
#843 F1: for a pending cross-role handoff, ``role`` must be the
|
|
authoritative profile-derived role supplied by the MCP boundary — never
|
|
caller-asserted authority. When the handoff provenance declares
|
|
``required_profile`` / ``required_namespace`` and the caller context is
|
|
provided, both are validated exactly; a mismatch fails closed.
|
|
"""
|
|
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)
|
|
|
|
handoff = is_pending_cross_role_handoff(state)
|
|
adopter_role = (role or "").strip().lower()
|
|
|
|
if freshness["freshness"] in ("abandoned", "released"):
|
|
raise LeaseLifecycleError(
|
|
f"lease {lease_id} is {freshness['freshness']}; cannot adopt "
|
|
"(fail closed)"
|
|
)
|
|
|
|
if handoff and not same_owner:
|
|
# Terminal statuses already rejected above. Freshness may be
|
|
# active OR stale_dead_process (controller exited) — both are
|
|
# consumable without abandonment when handoff is still pending.
|
|
if freshness["freshness"] not in (
|
|
"active",
|
|
"stale_dead_process",
|
|
"stale_missing_worktree",
|
|
):
|
|
raise LeaseLifecycleError(
|
|
f"lease {lease_id} freshness={freshness['freshness']}; "
|
|
"terminal or non-active allocation cannot be handoff-consumed "
|
|
"(fail closed)"
|
|
)
|
|
required = handoff["required_role"]
|
|
if adopter_role != required:
|
|
raise LeaseLifecycleError(
|
|
f"wrong role for cross-role handoff consume of {lease_id}: "
|
|
f"required={required} adopter={adopter_role or 'none'} "
|
|
"(fail closed)"
|
|
)
|
|
# #843 F1: provenance profile/namespace restrictions are validated
|
|
# against the authoritative caller context when declared. Caller
|
|
# input can never widen authority; a mismatch fails closed.
|
|
handoff_prov = handoff.get("provenance") or {}
|
|
required_profile = str(
|
|
handoff_prov.get("required_profile") or ""
|
|
).strip()
|
|
if required_profile and adopter_profile_name is not None:
|
|
if str(adopter_profile_name).strip() != required_profile:
|
|
raise LeaseLifecycleError(
|
|
f"wrong profile for cross-role handoff consume of "
|
|
f"{lease_id}: required_profile={required_profile} "
|
|
f"adopter_profile={adopter_profile_name} (fail closed)"
|
|
)
|
|
required_namespace = str(
|
|
handoff_prov.get("required_namespace") or ""
|
|
).strip()
|
|
if required_namespace and adopter_namespace is not None:
|
|
if str(adopter_namespace).strip() != required_namespace:
|
|
raise LeaseLifecycleError(
|
|
f"wrong namespace for cross-role handoff consume of "
|
|
f"{lease_id}: required_namespace={required_namespace} "
|
|
f"adopter_namespace={adopter_namespace} (fail closed)"
|
|
)
|
|
reason = "cross-role-handoff-consume"
|
|
elif freshness["freshness"] == "active" and not same_owner:
|
|
raise LeaseLifecycleError(
|
|
f"refusing to steal active foreign lease {lease_id} owned by "
|
|
f"{owner} (fail closed)"
|
|
)
|
|
elif not same_owner and freshness["freshness"] in (
|
|
"expired",
|
|
"stale_dead_process",
|
|
"stale_missing_worktree",
|
|
):
|
|
# Expired or stale (non-handoff): require abandon-style safety before
|
|
# ownership transfer when not same owner; same owner may reclaim.
|
|
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)"
|
|
)
|
|
reason = "sanctioned-reclaim-adopt"
|
|
else:
|
|
reason = "owner-resume-adopt" if same_owner else "sanctioned-reclaim-adopt"
|
|
|
|
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=reason,
|
|
)
|
|
if handoff and not same_owner:
|
|
provenance["cross_role_handoff"] = True
|
|
provenance["handoff_status"] = "adopted"
|
|
provenance["required_role"] = handoff["required_role"]
|
|
provenance["allocating_session_id"] = handoff["allocating_session_id"]
|
|
provenance["allocating_role"] = handoff["allocating_role"]
|
|
|
|
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,
|
|
)
|
|
out = {
|
|
"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 [],
|
|
}
|
|
if handoff and not same_owner:
|
|
out["cross_role_handoff"] = True
|
|
out["handoff_status"] = "adopted"
|
|
out["required_role"] = handoff["required_role"]
|
|
out["adopted_by_session_id"] = adopter_session_id
|
|
out["adopted_from_session_id"] = owner
|
|
lease_row = result.get("lease") or {}
|
|
if isinstance(lease_row, Mapping):
|
|
out["read_after_write"] = {
|
|
"lease_id": lease_row.get("lease_id"),
|
|
"session_id": lease_row.get("session_id"),
|
|
"role": lease_row.get("role"),
|
|
"status": lease_row.get("status"),
|
|
"adopted_by_session_id": lease_row.get("adopted_by_session_id"),
|
|
"adopted_from_session_id": lease_row.get("adopted_from_session_id"),
|
|
"phase": lease_row.get("phase"),
|
|
}
|
|
return out
|
|
|
|
|
|
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,
|
|
}
|