Gitea-Tools/lease_lifecycle.py

"""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,
    }