Gitea-Tools/issue_lock_store.py

"""Keyed, persistent issue-lock storage (#443) with flock hardening (#438).

Replaces the single global ``/tmp/gitea_issue_lock.json`` slot with per-issue
lock files under ``GITEA_ISSUE_LOCK_DIR`` (default
``~/.cache/gitea-tools/issue-locks``). Each MCP session binds its active lock
via a per-process pointer file so concurrent repos/issues never clobber each
other. Acquisition is serialized per issue with ``fcntl.flock``.
"""

from __future__ import annotations

import errno
import fcntl
import json
import os
import re
import tempfile
import uuid
from contextlib import contextmanager
from datetime import datetime, timedelta, timezone
from typing import Any

import lease_policy

LOCK_DIR_ENV = "GITEA_ISSUE_LOCK_DIR"
DEFAULT_LOCK_DIR = os.path.expanduser("~/.cache/gitea-tools/issue-locks")
AUTHOR_ISSUE_WORK_LEASE = "author_issue_work"

# Freshness classifications. ``STATUS_STALE`` remains the dead-PID band that
# #753 recovery keys on; the two bands below are new in #790 Slice A and apply
# only to leases minted under the heartbeat lifecycle.
STATUS_LIVE = "live"
STATUS_EXPIRED = "expired"
STATUS_ABSENT = "absent"
STATUS_STALE = "stale"
STATUS_STALE_MISSED_HEARTBEAT = "stale_missed_heartbeat"
STATUS_STALE_ABSOLUTE_CAP = "stale_absolute_cap"

_SAFE_SEGMENT_RE = re.compile(r"[^A-Za-z0-9._+-]+")


class LockContentionError(RuntimeError):
    """Raised when an exclusive per-issue lock cannot be acquired."""


def default_lock_dir() -> str:
    raw = (os.environ.get(LOCK_DIR_ENV) or DEFAULT_LOCK_DIR).strip()
    return raw or DEFAULT_LOCK_DIR


def _sanitize_segment(value: str) -> str:
    text = (value or "").strip()
    if not text:
        return "_"
    return _SAFE_SEGMENT_RE.sub("_", text)


def lock_key(
    *,
    remote: str,
    org: str,
    repo: str,
    issue_number: int,
) -> str:
    return "-".join(
        _sanitize_segment(part)
        for part in (remote, org, repo, str(issue_number))
    )


def lock_file_path(
    *,
    remote: str,
    org: str,
    repo: str,
    issue_number: int,
    lock_dir: str | None = None,
) -> str:
    root = (lock_dir or default_lock_dir()).strip()
    return os.path.join(root, f"{lock_key(remote=remote, org=org, repo=repo, issue_number=issue_number)}.json")


def session_pointer_path(lock_dir: str | None = None) -> str:
    root = (lock_dir or default_lock_dir()).strip()
    return os.path.join(root, f"session-{os.getpid()}.json")


def _ensure_lock_dir(lock_dir: str | None = None) -> str:
    root = (lock_dir or default_lock_dir()).strip()
    os.makedirs(root, mode=0o700, exist_ok=True)
    return root


def flock_path(json_path: str) -> str:
    return f"{json_path}.lock"


def is_process_alive(pid: int | None) -> bool:
    if not pid or pid <= 0:
        return False
    try:
        os.kill(int(pid), 0)
        return True
    except OSError as exc:
        return exc.errno != errno.ESRCH
    except (TypeError, ValueError):
        return False


@contextmanager
def _exclusive_file_lock(lock_path: str):
    os.makedirs(os.path.dirname(lock_path) or ".", exist_ok=True)
    fd = os.open(lock_path, os.O_CREAT | os.O_RDWR, 0o600)
    try:
        try:
            fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
        except BlockingIOError as exc:
            raise LockContentionError(
                f"could not acquire exclusive lock on '{lock_path}'"
            ) from exc
        yield fd
    finally:
        try:
            fcntl.flock(fd, fcntl.LOCK_UN)
        finally:
            os.close(fd)


def read_lock_file(path: str) -> dict[str, Any] | None:
    lock_path = (path or "").strip()
    if not lock_path or not os.path.exists(lock_path):
        return None
    try:
        with open(lock_path, encoding="utf-8") as handle:
            data = json.load(handle)
    except (OSError, json.JSONDecodeError):
        return None
    return data if isinstance(data, dict) else None


def save_lock_file(path: str, data: dict[str, Any]) -> None:
    lock_path = (path or "").strip()
    if not lock_path:
        raise ValueError("lock path is required (fail closed)")
    parent = os.path.dirname(lock_path) or "."
    os.makedirs(parent, mode=0o700, exist_ok=True)
    payload = json.dumps(data, indent=2, sort_keys=True) + "\n"
    fd, temp_path = tempfile.mkstemp(prefix=".lock-", suffix=".json", dir=parent)
    try:
        with os.fdopen(fd, "w", encoding="utf-8") as handle:
            handle.write(payload)
            handle.flush()
            os.fsync(handle.fileno())
        os.replace(temp_path, lock_path)
    finally:
        if os.path.exists(temp_path):
            try:
                os.remove(temp_path)
            except OSError:
                pass


def lock_generation(lock: dict[str, Any] | None) -> int:
    """Monotonic write counter for a durable lock record (#772 AC5).

    Absent or unusable values read as ``0`` so a lock written before generations
    existed still participates in compare-and-swap: its first recovery expects
    ``0`` and writes ``1``.
    """
    if not isinstance(lock, dict):
        return 0
    try:
        return int(lock.get("lock_generation") or 0)
    except (TypeError, ValueError):
        return 0


def bind_session_lock(
    lock_data: dict[str, Any],
    lock_dir: str | None = None,
    *,
    expected_generation: int | None = None,
    renewal_sanctioned: bool = False,
) -> str:
    """Persist a keyed lock and bind it to the current process session.

    ``expected_generation`` turns the write into a compare-and-swap (#772 AC5).
    Recovery decides it may take over a claim by reading the durable lock, but
    that read and this write are separate steps; without a CAS two replacement
    sessions can both observe the same dead owner, both pass assessment, and
    both write — the second silently clobbering the first. Passing the
    generation observed at assessment time makes exactly one of them win: the
    loser's expectation no longer matches and it fails closed.
    """
    remote = str(lock_data.get("remote") or "")
    org = str(lock_data.get("org") or "")
    repo = str(lock_data.get("repo") or "")
    issue_number = int(lock_data.get("issue_number") or 0)
    if not remote or not org or not repo or issue_number <= 0:
        raise ValueError("lock record must include remote, org, repo, and issue_number")

    root = _ensure_lock_dir(lock_dir)
    path = lock_file_path(
        remote=remote,
        org=org,
        repo=repo,
        issue_number=issue_number,
        lock_dir=root,
    )
    record = dict(lock_data)
    record["lock_file_path"] = path
    record["session_pid"] = os.getpid()
    record.setdefault("pid", os.getpid())

    pointer = {
        "pid": os.getpid(),
        "lock_file_path": path,
        "issue_number": issue_number,
        "branch_name": record.get("branch_name"),
        "remote": remote,
        "org": org,
        "repo": repo,
    }
    sentinel = flock_path(path)
    try:
        with _exclusive_file_lock(sentinel):
            existing = read_lock_file(path)
            overwrite_block = assess_foreign_lock_overwrite(existing, record)
            if overwrite_block:
                raise RuntimeError(overwrite_block)
            lease_block = assess_same_issue_lease_conflict(
                existing,
                issue_number=issue_number,
                branch_name=str(record.get("branch_name") or ""),
                worktree_path=str(record.get("worktree_path") or ""),
                renewal_sanctioned=renewal_sanctioned,
            )
            if lease_block:
                raise RuntimeError(lease_block)
            # #772 AC5: compare-and-swap inside the same critical section that
            # already serializes writers, so the check and the write cannot be
            # separated by another session's successful recovery.
            current_generation = lock_generation(existing)
            if (
                expected_generation is not None
                and current_generation != expected_generation
            ):
                raise RuntimeError(
                    f"Issue #{issue_number} lock generation changed: expected "
                    f"{expected_generation}, found {current_generation}; another "
                    "session already recovered or replaced this claim (fail closed)"
                )
            record["lock_generation"] = current_generation + 1
            save_lock_file(path, record)
            save_lock_file(session_pointer_path(root), pointer)
    except LockContentionError as exc:
        competing = read_lock_file(path)
        if competing:
            owner_pid = competing.get("session_pid") or competing.get("pid")
            raise RuntimeError(
                f"Issue #{issue_number} lock contention: {exc}; competing owner "
                f"pid={owner_pid} (fail closed)"
            ) from exc
        raise RuntimeError(f"Issue #{issue_number} lock contention: {exc} (fail closed)") from exc
    return path


def _ownership_refusals(
    lock: dict[str, Any],
    *,
    issue_number: int,
    branch_name: str,
    worktree_path: str,
    identity: str | None,
    profile: str | None,
) -> list[str]:
    """Exact-ownership mismatches between a durable lock and a live caller.

    Shared by the heartbeat writer and the legacy rebind path so the two cannot
    disagree about what "the same owner" means. Every field is compared against
    durable state; nothing is taken on the caller's word beyond the identity the
    server itself resolved.
    """
    reasons: list[str] = []
    if lock.get("issue_number") != issue_number:
        reasons.append(
            f"lock targets issue #{lock.get('issue_number')}, not #{issue_number}"
        )
    if str(lock.get("branch_name") or "") != str(branch_name or ""):
        reasons.append(
            f"lock branch '{lock.get('branch_name')}' does not match '{branch_name}'"
        )
    if not _same_realpath(str(lock.get("worktree_path") or ""), worktree_path):
        reasons.append(
            f"lock worktree '{lock.get('worktree_path')}' does not match "
            f"'{worktree_path}'"
        )
    lease = lock.get("work_lease") if isinstance(lock, dict) else None
    claimant = lease.get("claimant") if isinstance(lease, dict) else None
    claimant = claimant if isinstance(claimant, dict) else {}
    recorded_identity = str(claimant.get("username") or "").strip()
    recorded_profile = str(claimant.get("profile") or "").strip()
    if not recorded_identity or not recorded_profile:
        reasons.append("lock does not record both a claimant username and profile")
    if recorded_identity and recorded_identity != str(identity or "").strip():
        reasons.append(
            f"lock claimant '{recorded_identity}' does not match active identity "
            f"'{str(identity or '').strip() or 'unknown'}'"
        )
    if recorded_profile and recorded_profile != str(profile or "").strip():
        reasons.append(
            f"lock profile '{recorded_profile}' does not match active profile "
            f"'{str(profile or '').strip() or 'unknown'}'"
        )
    return reasons


def _refusal(reasons: list[str], **extra: Any) -> dict[str, Any]:
    return {"success": False, "performed": False, "reasons": reasons, **extra}


def heartbeat_session_lock(
    *,
    remote: str,
    org: str,
    repo: str,
    issue_number: int,
    branch_name: str,
    worktree_path: str,
    identity: str | None,
    profile: str | None,
    task_session_id: str,
    expected_generation: int | None = None,
    lock_dir: str | None = None,
    now: datetime | None = None,
) -> dict[str, Any]:
    """Slide a heartbeat-lifecycle lease forward (#790 Slice A, A4).

    The write happens inside the same per-issue ``flock`` that serializes
    acquisition, and under the #772 generation compare-and-swap, so a heartbeat
    can never race a concurrent reclaim: whichever lands first moves the
    generation and the other fails closed.

    Refuses — never revives — in every ambiguous case. A lease that has already
    lapsed past its grace is *not* heartbeatable: allowing that would let a
    session that stopped proving liveness restore ownership retroactively, which
    is precisely the revival AC-N5 forbids. Such a session must go through the
    sanctioned reclaim path, which mints a fresh generation.
    """
    current = _lease_now(now)
    root = _ensure_lock_dir(lock_dir)
    path = lock_file_path(
        remote=remote, org=org, repo=repo, issue_number=issue_number, lock_dir=root
    )
    declared_session = str(task_session_id or "").strip()
    if not declared_session:
        return _refusal(["no task_session_id supplied (fail closed)"])

    sentinel = flock_path(path)
    try:
        with _exclusive_file_lock(sentinel):
            lock = read_lock_file(path)
            if not lock:
                return _refusal([f"no durable lock for issue #{issue_number}"])

            if is_legacy_lease(lock):
                return _refusal(
                    [
                        "lock predates the heartbeat lifecycle; it must be rebound "
                        "by its exact owner before it can be heartbeated"
                    ],
                    lifecycle=lease_lifecycle_version(lock),
                    legacy_lease=True,
                )

            reasons = _ownership_refusals(
                lock,
                issue_number=issue_number,
                branch_name=branch_name,
                worktree_path=worktree_path,
                identity=identity,
                profile=profile,
            )
            recorded_session = lease_task_session_id(lock)
            if not recorded_session:
                reasons.append(
                    "lock declares the heartbeat lifecycle but records no "
                    "task_session_id (fail closed)"
                )
            elif recorded_session != declared_session:
                # A superseded session holding an old identifier cannot heartbeat
                # over the session that replaced it.
                reasons.append(
                    "task_session_id does not match the session recorded on the lock"
                )
            if reasons:
                return _refusal(reasons)

            current_generation = lock_generation(lock)
            if (
                expected_generation is not None
                and current_generation != expected_generation
            ):
                return _refusal(
                    [
                        f"lock generation changed: expected {expected_generation}, "
                        f"found {current_generation}; another session reclaimed or "
                        "replaced this claim (fail closed)"
                    ],
                    lock_generation=current_generation,
                )

            freshness = assess_lock_freshness(lock, now=current)
            if not freshness.get("live"):
                return _refusal(
                    [
                        f"lease is not live ({freshness.get('status')}): "
                        f"{freshness.get('reason')}; a lapsed lease must be "
                        "reclaimed, not heartbeated"
                    ],
                    freshness=freshness,
                )

            policy = lease_policy.policy_for(lease_task_class(lock))
            expires = current + timedelta(minutes=policy.initial_ttl_minutes)
            record = dict(lock)
            lease = dict(record.get("work_lease") or {})
            prior_heartbeat = lease.get("last_heartbeat_at")
            lease["last_heartbeat_at"] = _format_lease_timestamp(current)
            lease["expires_at"] = _format_lease_timestamp(expires)
            try:
                lease["heartbeat_count"] = int(lease.get("heartbeat_count") or 0) + 1
            except (TypeError, ValueError):
                lease["heartbeat_count"] = 1
            record["work_lease"] = lease
            record["lock_generation"] = current_generation + 1
            save_lock_file(path, record)
    except LockContentionError as exc:
        return _refusal([f"issue #{issue_number} lock contention: {exc} (fail closed)"])

    return {
        "success": True,
        "performed": True,
        "issue_number": issue_number,
        "branch_name": branch_name,
        "worktree_path": worktree_path,
        "task_session_id": declared_session,
        "lock_generation": record["lock_generation"],
        "prior_generation": current_generation,
        "prior_heartbeat_at": prior_heartbeat,
        "last_heartbeat_at": lease["last_heartbeat_at"],
        "expires_at": lease["expires_at"],
        "heartbeat_count": lease["heartbeat_count"],
        "lock_file_path": path,
        "policy": lease_policy.describe(lease_task_class(record)),
        "freshness": assess_lock_freshness(record, now=current),
    }


def rebind_legacy_lock(
    *,
    remote: str,
    org: str,
    repo: str,
    issue_number: int,
    branch_name: str,
    worktree_path: str,
    identity: str | None,
    profile: str | None,
    expected_generation: int | None = None,
    lock_dir: str | None = None,
    now: datetime | None = None,
) -> dict[str, Any]:
    """Move a legacy lock into the heartbeat lifecycle (#790 AC-N8).

    One of the two sanctioned exits from the preserved-expiry legacy state; the
    other is terminal retirement, which is Slice B. Only the exact recorded
    owner may rebind, and only while the legacy lock is still live under its
    original absolute expiry — an already-expired legacy lease belongs to the
    #760 renewal path or #601 reclaim, and this must not become a second, weaker
    way to revive one.

    The rebind mints a genuine task-session identifier and a genuine first
    heartbeat. It does not fabricate history: the original creation and expiry
    are preserved under ``legacy_origin`` for audit, and the new lifecycle's
    absolute cap runs from the rebind, not from the legacy claim.
    """
    current = _lease_now(now)
    root = _ensure_lock_dir(lock_dir)
    path = lock_file_path(
        remote=remote, org=org, repo=repo, issue_number=issue_number, lock_dir=root
    )
    sentinel = flock_path(path)
    try:
        with _exclusive_file_lock(sentinel):
            lock = read_lock_file(path)
            if not lock:
                return _refusal([f"no durable lock for issue #{issue_number}"])
            if not is_legacy_lease(lock):
                return _refusal(
                    [
                        "lock is already on the heartbeat lifecycle; use the "
                        "heartbeat path"
                    ],
                    lifecycle=lease_lifecycle_version(lock),
                    legacy_lease=False,
                )

            reasons = _ownership_refusals(
                lock,
                issue_number=issue_number,
                branch_name=branch_name,
                worktree_path=worktree_path,
                identity=identity,
                profile=profile,
            )
            if reasons:
                return _refusal(reasons)

            current_generation = lock_generation(lock)
            if (
                expected_generation is not None
                and current_generation != expected_generation
            ):
                return _refusal(
                    [
                        f"lock generation changed: expected {expected_generation}, "
                        f"found {current_generation} (fail closed)"
                    ],
                    lock_generation=current_generation,
                )

            freshness = assess_lock_freshness(lock, now=current)
            if not freshness.get("live"):
                return _refusal(
                    [
                        f"legacy lease is not live ({freshness.get('status')}): "
                        f"{freshness.get('reason')}; rebinding is not a recovery "
                        "path for a lapsed lease"
                    ],
                    freshness=freshness,
                )

            policy = lease_policy.policy_for(lease_task_class(lock))
            expires = current + timedelta(minutes=policy.initial_ttl_minutes)
            session_id = mint_task_session_id(lease_task_class(lock))
            record = dict(lock)
            lease = dict(record.get("work_lease") or {})
            legacy_origin = {
                "created_at": lease.get("created_at"),
                "expires_at": lease.get("expires_at"),
                "last_heartbeat_at": lease.get("last_heartbeat_at"),
                "lifecycle": lease_policy.LIFECYCLE_LEGACY,
            }
            lease["lifecycle_version"] = lease_policy.LIFECYCLE_HEARTBEAT_V1
            lease["task_session_id"] = session_id
            lease["created_at"] = _format_lease_timestamp(current)
            lease["last_heartbeat_at"] = _format_lease_timestamp(current)
            lease["expires_at"] = _format_lease_timestamp(expires)
            lease["heartbeat_count"] = 1
            record["work_lease"] = lease
            record["legacy_rebind"] = {
                "rebound_at": _format_lease_timestamp(current),
                "task_session_id": session_id,
                "prior_generation": current_generation,
                "legacy_origin": legacy_origin,
                "reason": (
                    "legacy lock rebound into the heartbeat lifecycle by its exact "
                    "recorded owner"
                ),
            }
            record["lock_generation"] = current_generation + 1
            save_lock_file(path, record)
    except LockContentionError as exc:
        return _refusal([f"issue #{issue_number} lock contention: {exc} (fail closed)"])

    return {
        "success": True,
        "performed": True,
        "issue_number": issue_number,
        "task_session_id": session_id,
        "lock_generation": record["lock_generation"],
        "prior_generation": current_generation,
        "lifecycle": lease_policy.LIFECYCLE_HEARTBEAT_V1,
        "legacy_rebind": record["legacy_rebind"],
        "expires_at": lease["expires_at"],
        "last_heartbeat_at": lease["last_heartbeat_at"],
        "lock_file_path": path,
        "freshness": assess_lock_freshness(record, now=current),
    }


def read_session_issue_lock(lock_dir: str | None = None) -> dict[str, Any] | None:
    root = (lock_dir or default_lock_dir()).strip()
    pointer = read_lock_file(session_pointer_path(root))
    if not pointer:
        return None
    lock_path = str(pointer.get("lock_file_path") or "").strip()
    if not lock_path:
        return None
    return read_lock_file(lock_path)


def load_issue_lock(
    *,
    remote: str,
    org: str,
    repo: str,
    issue_number: int,
    lock_dir: str | None = None,
) -> dict[str, Any] | None:
    return read_lock_file(
        lock_file_path(
            remote=remote,
            org=org,
            repo=repo,
            issue_number=issue_number,
            lock_dir=lock_dir,
        )
    )


def iter_lock_files(lock_dir: str | None = None) -> list[str]:
    root = (lock_dir or default_lock_dir()).strip()
    if not os.path.isdir(root):
        return []
    paths: list[str] = []
    for name in os.listdir(root):
        if not name.endswith(".json") or name.startswith("session-"):
            continue
        paths.append(os.path.join(root, name))
    return sorted(paths)


def find_lock_for_branch(
    *,
    remote: str,
    org: str,
    repo: str,
    branch_name: str,
    lock_dir: str | None = None,
) -> dict[str, Any] | None:
    target = (branch_name or "").strip()
    if not target:
        return None
    for path in iter_lock_files(lock_dir):
        lock = read_lock_file(path)
        if not lock:
            continue
        if (
            str(lock.get("remote") or "") == remote
            and str(lock.get("org") or "") == org
            and str(lock.get("repo") or "") == repo
            and str(lock.get("branch_name") or "").strip() == target
        ):
            lock = dict(lock)
            lock.setdefault("lock_file_path", path)
            return lock
    return None


def _lease_now(now: datetime | None = None) -> datetime:
    return now or datetime.now(timezone.utc)


def _parse_lease_timestamp(value: str | None) -> datetime | None:
    text = (value or "").strip()
    if not text:
        return None
    try:
        return datetime.fromisoformat(text.replace("Z", "+00:00")).astimezone(timezone.utc)
    except ValueError:
        return None


def _format_lease_timestamp(value: datetime) -> str:
    """Serialize a lease timestamp in the durable ``...Z`` form already on disk."""
    return (
        value.astimezone(timezone.utc)
        .replace(microsecond=0)
        .isoformat()
        .replace("+00:00", "Z")
    )


def lease_expires_at(lock: dict[str, Any] | None) -> datetime | None:
    if not lock:
        return None
    lease = lock.get("work_lease")
    if not isinstance(lease, dict):
        return None
    return _parse_lease_timestamp(lease.get("expires_at"))


def is_lease_expired(lock: dict[str, Any] | None, *, now: datetime | None = None) -> bool:
    expires = lease_expires_at(lock)
    if expires is None:
        return False
    return expires <= _lease_now(now)


def is_lease_live(lock: dict[str, Any] | None, *, now: datetime | None = None) -> bool:
    return assess_lock_freshness(lock, now=now)["live"]


def lease_task_class(lock_data: dict[str, Any] | None) -> str:
    """Policy task class for a durable lock; author work when unrecorded."""
    lease = lock_data.get("work_lease") if isinstance(lock_data, dict) else None
    if isinstance(lease, dict):
        recorded = str(lease.get("operation_type") or "").strip()
        if recorded:
            return recorded
    return AUTHOR_ISSUE_WORK_LEASE


def lease_lifecycle_version(lock_data: dict[str, Any] | None) -> str:
    """Read the durable lifecycle marker (#790 AC-N8).

    The marker is the *only* discriminator between a heartbeat-lifecycle lease
    and a legacy one. Timestamps are deliberately not consulted: a lock minted
    before this lifecycle existed has ``last_heartbeat_at == created_at``
    forever, and reading that equality as "recently heartbeated" would treat
    every never-heartbeated legacy lock as fresh — the precise inversion AC-N8
    forbids. A newly minted heartbeat lease also has the two equal, so the
    equality carries no information in either direction.
    """
    lease = lock_data.get("work_lease") if isinstance(lock_data, dict) else None
    if isinstance(lease, dict):
        recorded = str(lease.get("lifecycle_version") or "").strip()
        if recorded:
            return recorded
    return lease_policy.LIFECYCLE_LEGACY


def is_legacy_lease(lock_data: dict[str, Any] | None) -> bool:
    """True when a lock predates the shared heartbeat lifecycle."""
    return lease_lifecycle_version(lock_data) != lease_policy.LIFECYCLE_HEARTBEAT_V1


def lease_task_session_id(lock_data: dict[str, Any] | None) -> str:
    """Recorded per-task session identifier, or empty for a legacy lock."""
    lease = lock_data.get("work_lease") if isinstance(lock_data, dict) else None
    if isinstance(lease, dict):
        return str(lease.get("task_session_id") or "").strip()
    return ""


def mint_task_session_id(task_class: str = AUTHOR_ISSUE_WORK_LEASE) -> str:
    """Mint an ownership key for one task (#790 AC-N1).

    Deliberately contains no process identifier. The recorded PID belongs to the
    long-lived MCP daemon, which outlives any individual task and is reused by
    every task it serves, so PID digits cannot identify *which* task holds a
    claim. The PID is still recorded alongside this value as evidence.
    """
    prefix = _sanitize_segment(str(task_class or AUTHOR_ISSUE_WORK_LEASE))
    return f"{prefix}-{uuid.uuid4().hex[:16]}"


def _lease_heartbeat_at(lock_data: dict[str, Any] | None) -> datetime | None:
    lease = lock_data.get("work_lease") if isinstance(lock_data, dict) else None
    heartbeat_at = None
    if isinstance(lock_data, dict):
        heartbeat_at = _parse_lease_timestamp(lock_data.get("last_heartbeat_at"))
    if heartbeat_at is None and isinstance(lease, dict):
        heartbeat_at = _parse_lease_timestamp(lease.get("last_heartbeat_at"))
    return heartbeat_at


def assess_lock_freshness(
    lock_data: dict[str, Any] | None,
    *,
    now: datetime | None = None,
) -> dict[str, Any]:
    """Classify a lock as live, expired, stale, or absent.

    #790 Slice A makes the heartbeat load-bearing. Before this change
    ``last_heartbeat_at`` was parsed and then never consulted: liveness was
    decided entirely by the absolute ``expires_at`` and by PID liveness, and
    since the recorded PID is the long-lived MCP daemon, an abandoned author
    task stayed "live" for the full four-hour TTL.

    Two rules govern the rewrite:

    * **An alive PID never establishes freshness** (AC-N2). It proves the daemon
      is up, nothing about the task. It is recorded as evidence and no branch
      returns ``live`` because of it.
    * **A dead PID still corroborates staleness.** The dead-PID band is
      unchanged and still precedes every heartbeat evaluation, so #753
      dead-session recovery keys on exactly the classification it always did.

    Legacy leases (AC-N8) keep their recorded absolute expiry and are never
    evaluated against the short heartbeat grace, so deploying this change cannot
    make an existing claim instantly reclaimable.
    """
    current = _lease_now(now)
    if not lock_data:
        return {
            "status": STATUS_ABSENT,
            "live": False,
            "stale": False,
            "reason": "no lock record",
        }

    lease = lock_data.get("work_lease")
    expires_at = lease_expires_at(lock_data)
    heartbeat_at = _lease_heartbeat_at(lock_data)
    created_at = (
        _parse_lease_timestamp(lease.get("created_at"))
        if isinstance(lease, dict)
        else None
    )

    pid = lock_data.get("session_pid")
    if pid is None:
        pid = lock_data.get("pid")
    # Evidence only. Never consulted to grant liveness (AC-N2).
    pid_alive = is_process_alive(pid) if pid is not None else False

    lifecycle = lease_lifecycle_version(lock_data)
    legacy = lifecycle != lease_policy.LIFECYCLE_HEARTBEAT_V1
    policy = lease_policy.policy_for(lease_task_class(lock_data))

    evidence: dict[str, Any] = {
        "pid_alive": pid_alive,
        "lifecycle": lifecycle,
        "legacy_lease": legacy,
        "task_session_id": lease_task_session_id(lock_data) or None,
        "heartbeat_at": heartbeat_at.isoformat() if heartbeat_at else None,
        "expires_at": expires_at.isoformat() if expires_at else None,
    }

    def _result(status: str, *, live: bool, reason: str, **extra: Any) -> dict[str, Any]:
        return {
            "status": status,
            "live": live,
            "stale": not live and status != STATUS_ABSENT,
            "reason": reason,
            **evidence,
            **extra,
        }

    if legacy:
        # AC-N8: the preserved absolute expiry is the only clock for a lock
        # written before task-session heartbeats existed.
        if expires_at and expires_at <= current:
            return _result(
                STATUS_EXPIRED,
                live=False,
                reason=f"lease expired at {expires_at.isoformat()}",
            )
        if pid is not None and not pid_alive:
            return _result(
                STATUS_STALE, live=False, reason=f"owner pid {pid} is not alive"
            )
        return _result(
            STATUS_LIVE,
            live=True,
            reason=(
                "legacy lease is within its recorded absolute expiry; the "
                "heartbeat grace does not apply retroactively"
            ),
            legacy_expiry_preserved=True,
        )

    # ── Heartbeat lifecycle ──
    if pid is not None and not pid_alive:
        # Unchanged dead-PID band: #753 recovery depends on this exact status.
        return _result(STATUS_STALE, live=False, reason=f"owner pid {pid} is not alive")

    if heartbeat_at is None:
        # Contradictory: a heartbeat lease must carry a heartbeat. Fail closed.
        return _result(
            STATUS_STALE_MISSED_HEARTBEAT,
            live=False,
            reason=(
                f"lease declares lifecycle '{lifecycle}' but records no "
                "last_heartbeat_at (fail closed)"
            ),
        )

    if policy.absolute_cap_hours and created_at is not None:
        cap_at = created_at + timedelta(hours=policy.absolute_cap_hours)
        if cap_at <= current:
            return _result(
                STATUS_STALE_ABSOLUTE_CAP,
                live=False,
                reason=(
                    f"lease exceeded its {policy.absolute_cap_hours}h absolute cap "
                    f"at {cap_at.isoformat()}; canonical re-adoption is required"
                ),
                absolute_cap_at=cap_at.isoformat(),
            )

    grace_at = heartbeat_at + timedelta(minutes=policy.missed_heartbeat_grace_minutes)
    if grace_at <= current or (expires_at is not None and expires_at <= current):
        return _result(
            STATUS_STALE_MISSED_HEARTBEAT,
            live=False,
            reason=(
                f"no valid heartbeat since {heartbeat_at.isoformat()}; the "
                f"{policy.missed_heartbeat_grace_minutes}min grace lapsed at "
                f"{grace_at.isoformat()}"
            ),
            missed_heartbeat_since=grace_at.isoformat(),
        )

    warning_at = heartbeat_at + timedelta(minutes=policy.stale_warning_minutes)
    return _result(
        STATUS_LIVE,
        live=True,
        reason="lease heartbeat is fresh within the configured grace",
        heartbeat_warning=warning_at <= current,
    )


def _same_realpath(left: str | None, right: str | None) -> bool:
    if not left or not right:
        return False
    try:
        return os.path.realpath(left) == os.path.realpath(right)
    except OSError:
        return left == right



def assess_expired_lock_reclaim(
    existing_lock: dict[str, Any] | None,
    *,
    now: datetime | None = None,
) -> dict[str, Any]:
    """Decide whether an expired/stale author issue lock may be reclaimed (#601).

    Required proof (any reclaim of non-live lock):
    * lease not live (expired or dead pid)
    * owner process dead OR worktree missing
    * no force-delete of live foreign ownership
    """
    if not existing_lock:
        return {
            "reclaim_allowed": True,
            "reasons": ["no existing lock"],
            "freshness": assess_lock_freshness(None, now=now),
        }
    freshness = assess_lock_freshness(existing_lock, now=now)
    if freshness.get("live"):
        return {
            "reclaim_allowed": False,
            "reasons": ["lock is still live; cannot reclaim (fail closed)"],
            "freshness": freshness,
        }
    status = str(freshness.get("status") or "")
    if status in (STATUS_STALE_MISSED_HEARTBEAT, STATUS_STALE_ABSOLUTE_CAP):
        # #790: under the heartbeat lifecycle the heartbeat *is* the liveness
        # proof, so a session that stopped heartbeating past its grace has
        # released its claim by definition. Requiring a dead PID on top of that
        # would reinstate the original defect — the recorded PID is the shared
        # daemon, which stays alive across every abandoned task it ever served.
        #
        # This band is unreachable for a legacy lease (AC-N8), so no lock
        # written before this lifecycle can be reclaimed by this path.
        return {
            "reclaim_allowed": True,
            "reasons": [
                f"heartbeat-lifecycle lease is {status}: {freshness.get('reason')}"
            ],
            "freshness": freshness,
            "prior_branch": existing_lock.get("branch_name"),
            "prior_worktree": existing_lock.get("worktree_path"),
            "prior_pid": existing_lock.get("session_pid") or existing_lock.get("pid"),
            "prior_task_session_id": lease_task_session_id(existing_lock) or None,
        }
    pid = existing_lock.get("session_pid")
    if pid is None:
        pid = existing_lock.get("pid")
    dead = not is_process_alive(pid) if pid is not None else True
    wt = str(existing_lock.get("worktree_path") or "")
    missing_wt = (not wt) or (not os.path.isdir(os.path.realpath(wt)))
    if not (dead or missing_wt):
        return {
            "reclaim_allowed": False,
            "reasons": [
                "expired/stale lock still has live owner pid and present worktree; "
                "recovery review required (fail closed)"
            ],
            "freshness": freshness,
            "owner_pid_dead": dead,
            "worktree_missing": missing_wt,
        }
    return {
        "reclaim_allowed": True,
        "reasons": [
            "non-live lock with dead process and/or missing worktree; "
            "sanctioned reclaim allowed"
        ],
        "freshness": freshness,
        "owner_pid_dead": dead,
        "worktree_missing": missing_wt,
        "prior_branch": existing_lock.get("branch_name"),
        "prior_worktree": existing_lock.get("worktree_path"),
        "prior_pid": pid,
    }


def assess_same_issue_lease_conflict(
    existing_lock: dict[str, Any] | None,
    *,
    issue_number: int,
    branch_name: str,
    worktree_path: str,
    operation_type: str = AUTHOR_ISSUE_WORK_LEASE,
    renewal_sanctioned: bool = False,
    now: datetime | None = None,
) -> str | None:
    """Return a fail-closed error when a competing live lease blocks acquisition.

    ``renewal_sanctioned`` is set only when
    ``issue_lock_renewal.assess_exact_owner_lease_renewal`` has already proven,
    from the durable lock plus live server-side observation, that this session
    is the exact recorded owner of an *expired* lease (#760). It is never a
    caller-supplied parameter of any MCP tool (#760 AC14): the server computes
    it and passes it down. Left False, every pre-existing disposition is
    unchanged.
    """
    if not existing_lock:
        return None

    existing_issue = existing_lock.get("issue_number")
    lease = existing_lock.get("work_lease")
    existing_operation = (
        lease.get("operation_type")
        if isinstance(lease, dict)
        else AUTHOR_ISSUE_WORK_LEASE
    )
    if existing_issue != issue_number or existing_operation != operation_type:
        return None

    existing_branch = existing_lock.get("branch_name")
    existing_worktree = existing_lock.get("worktree_path")
    same_owner = (
        existing_branch == branch_name
        and _same_realpath(str(existing_worktree or ""), worktree_path)
    )
    expired = is_lease_expired(existing_lock, now=now)
    # #790 review #502/#516: a heartbeat-lifecycle lease that is non-live but
    # whose absolute expires_at is still in the future must enter the same
    # reclaim/renewal disposition as an expired lease. This happens for
    # stale_absolute_cap (a session that keeps heartbeating past the 8h cap has
    # expires_at = last_heartbeat + TTL in the future) and for
    # stale_missed_heartbeat under an independent TTL>grace policy. Keying this
    # gate on is_lease_expired alone left assess_expired_lock_reclaim — which
    # already permits exactly those bands — unreachable from the acquisition
    # path, so the load-bearing heartbeat was not load-bearing for foreign
    # reclaim: the precise abandonment scenario #790 exists to fix.
    # assess_foreign_lock_overwrite already keys on is_lease_live; this makes the
    # same-issue path consistent with it. Legacy leases keep their absolute-expiry
    # clock (AC-N8): is_lease_live already decides a legacy lease from expires_at
    # / dead-PID alone and is_legacy_lease excludes it here, so this widening is a
    # no-op for every pre-lifecycle lock.
    heartbeat_non_live_reclaimable = (
        not expired
        and not is_legacy_lease(existing_lock)
        and not is_lease_live(existing_lock, now=now)
    )
    if expired or heartbeat_non_live_reclaimable:
        # #760 AC1/AC2: exact-owner renewal is a different disposition from
        # foreign takeover and is evaluated first. Before this, both branches
        # below returned unconditionally, so the same_owner allowance further
        # down was unreachable for every expired lease — an owner could never
        # renew its own lock once the wall clock passed, no matter how complete
        # its ownership evidence. Requires BOTH the locally recomputed
        # same_owner match and the server-proven renewal waiver; either alone is
        # insufficient.
        if same_owner and renewal_sanctioned:
            return None
        reclaim = assess_expired_lock_reclaim(existing_lock, now=now)
        if reclaim.get("reclaim_allowed"):
            # #601: expired + dead pid / missing worktree may be reclaimed
            # through the normal lock path (sanctioned overwrite). #790: the new
            # stale heartbeat bands are reclaimable here on the same evidence.
            return None
        descriptor = "expired" if expired else "non-live"
        return (
            f"Issue #{issue_number} has an {descriptor} {operation_type} lease on "
            f"branch '{existing_branch}' from worktree '{existing_worktree}'. "
            "Recovery review is required before takeover (fail closed)"
        )
    if same_owner:
        return None
    return (
        f"Issue #{issue_number} already has an active {operation_type} lease on "
        f"branch '{existing_branch}' from worktree '{existing_worktree}' "
        "(fail closed)"
    )


def assess_foreign_lock_overwrite(
    existing_lock: dict[str, Any] | None,
    incoming_lock: dict[str, Any],
    *,
    now: datetime | None = None,
) -> str | None:
    """Block writes that would clobber an unrelated live lease on the same key."""
    if not existing_lock:
        return None

    same_issue = existing_lock.get("issue_number") == incoming_lock.get("issue_number")
    same_branch = existing_lock.get("branch_name") == incoming_lock.get("branch_name")
    same_worktree = _same_realpath(
        str(existing_lock.get("worktree_path") or ""),
        str(incoming_lock.get("worktree_path") or ""),
    )
    if same_issue and same_branch and same_worktree:
        return None
    if not is_lease_live(existing_lock, now=now):
        return None
    return (
        "Refusing to overwrite a live foreign issue lock "
        f"(issue #{existing_lock.get('issue_number')}, "
        f"branch '{existing_lock.get('branch_name')}', "
        f"worktree '{existing_lock.get('worktree_path')}') (fail closed)"
    )


def find_live_lock_for_branch(
    branch_name: str,
    lock_dir: str | None = None,
) -> dict[str, Any] | None:
    target = (branch_name or "").strip()
    if not target:
        return None
    for path in iter_lock_files(lock_dir):
        lock = read_lock_file(path)
        if not lock:
            continue
        if str(lock.get("branch_name") or "").strip() != target:
            continue
        if not is_lease_live(lock):
            continue
        record = dict(lock)
        record.setdefault("lock_file_path", path)
        return record
    return None


def resolve_locked_branch_for_session(
    branch_name: str | None = None,
    lock_dir: str | None = None,
) -> str:
    if branch_name:
        lock = find_live_lock_for_branch(branch_name, lock_dir)
        if lock:
            return str(lock.get("branch_name") or "")
    lock = read_session_issue_lock(lock_dir)
    return str((lock or {}).get("branch_name") or "")


def has_active_issue_lock(
    branch: str,
    *,
    lock_dir: str | None = None,
) -> bool:
    target = (branch or "").strip()
    if not target:
        return False
    for path in iter_lock_files(lock_dir):
        lock = read_lock_file(path)
        if not lock:
            continue
        if str(lock.get("branch_name") or "").strip() != target:
            continue
        if is_lease_live(lock):
            return True
    return False


def verify_lock_for_mutation(
    lock_data: dict[str, Any] | None,
    *,
    issue_number: int | None = None,
    branch_name: str | None = None,
    worktree_path: str | None = None,
) -> dict[str, Any]:
    """Re-check lock ownership immediately before a mutation (#438)."""
    reasons: list[str] = []
    if not lock_data:
        return {"proven": False, "block": True, "reasons": ["issue lock is missing (fail closed)"]}

    freshness = assess_lock_freshness(lock_data)
    if not freshness["live"]:
        reasons.append(f"issue lock is not live: {freshness['reason']} (fail closed)")

    if issue_number is not None and lock_data.get("issue_number") != issue_number:
        reasons.append(
            f"issue lock targets #{lock_data.get('issue_number')}, expected #{issue_number} (fail closed)"
        )

    if branch_name is not None and lock_data.get("branch_name") != branch_name:
        reasons.append(
            f"issue lock branch '{lock_data.get('branch_name')}' does not match "
            f"'{branch_name}' (fail closed)"
        )

    if worktree_path is not None:
        locked = os.path.realpath(str(lock_data.get("worktree_path") or ""))
        declared = os.path.realpath(worktree_path)
        if locked != declared:
            reasons.append(
                f"issue lock worktree '{locked}' does not match declared '{declared}' (fail closed)"
            )

    return {
        "proven": not reasons,
        "block": bool(reasons),
        "reasons": reasons,
        "freshness": freshness,
        "lock_proof": format_lock_proof(lock_data, freshness=freshness),
    }


def list_live_locks(
    *,
    lock_dir: str | None = None,
    now: datetime | None = None,
) -> list[dict[str, Any]]:
    """Return live per-issue locks for queue visibility."""
    live: list[dict[str, Any]] = []
    for path in iter_lock_files(lock_dir):
        record = read_lock_file(path)
        if not record:
            continue
        freshness = assess_lock_freshness(record, now=now)
        if not freshness["live"]:
            continue
        live.append(
            {
                "issue_number": record.get("issue_number"),
                "branch_name": record.get("branch_name"),
                "remote": record.get("remote"),
                "org": record.get("org"),
                "repo": record.get("repo"),
                "worktree_path": record.get("worktree_path"),
                "pid": record.get("session_pid") or record.get("pid"),
                "claimant": (
                    record.get("claimant")
                    or (record.get("work_lease") or {}).get("claimant")
                ),
                "freshness": freshness,
                "lock_path": record.get("lock_file_path") or path,
            }
        )
    return live


def format_lock_proof(
    lock_data: dict[str, Any] | None,
    *,
    freshness: dict[str, Any] | None = None,
    competing_live_locks: list[dict[str, Any]] | None = None,
    released: bool | None = None,
) -> str:
    """Canonical issue-lock proof string for final reports."""
    if not lock_data:
        return "issue lock proof: not acquired"
    fresh = freshness or assess_lock_freshness(lock_data)
    owner = lock_data.get("claimant") or {}
    if not owner and isinstance(lock_data.get("work_lease"), dict):
        owner = lock_data["work_lease"].get("claimant") or {}
    parts = [
        "issue lock proof:",
        f"acquired issue #{lock_data.get('issue_number')}",
        f"branch {lock_data.get('branch_name')}",
        f"owner {owner.get('profile') or 'unknown'}",
        f"pid {lock_data.get('session_pid') or lock_data.get('pid')}",
        f"freshness {fresh.get('status')}",
    ]
    if competing_live_locks is not None:
        parts.append(
            "no competing live lock"
            if not competing_live_locks
            else f"competing live locks {len(competing_live_locks)}"
        )
    if released is True:
        parts.append("lock released")
    elif released is False:
        parts.append("lock retained")
    return "; ".join(parts)