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
from contextlib import contextmanager
from datetime import datetime, timedelta, timezone
from typing import Any

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

_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,
    recovery_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, recovery_sanctioned=recovery_sanctioned
            )
            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,
                recovery_sanctioned=recovery_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 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 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 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."""
    current = _lease_now(now)
    if not lock_data:
        return {
            "status": "absent",
            "live": False,
            "stale": False,
            "reason": "no lock record",
        }

    expires_at = lease_expires_at(lock_data)
    lease = lock_data.get("work_lease")
    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"))

    pid = lock_data.get("session_pid")
    if pid is None:
        pid = lock_data.get("pid")
    if pid is None:
        pid = lock_data.get("owner_pid")
    pid_missing = pid is None or str(pid).strip() == ""
    try:
        pid_int = int(pid) if not pid_missing else None
        if pid_int is not None and pid_int <= 0:
            pid_missing = True
            pid_int = None
    except (TypeError, ValueError):
        pid_missing = True
        pid_int = None
    pid_alive = is_process_alive(pid_int) if pid_int is not None else False

    if expires_at and expires_at <= current:
        return {
            "status": "expired",
            "live": False,
            "stale": True,
            "reason": f"lease expired at {expires_at.isoformat()}",
            "pid_alive": pid_alive,
            "pid_missing": pid_missing,
        }

    # #860: a PID-less lock must never be considered live merely because
    # expiration / heartbeat fields are absent. Missing PID is insufficient
    # evidence of a live owner; treat as malformed/stale so recovery routes
    # can evaluate corroborating pins instead of blocking on a false live flag.
    if pid_missing:
        return {
            "status": "malformed",
            "live": False,
            "stale": True,
            "reason": (
                "lock has no usable session pid; cannot prove live ownership "
                "(PID-less locks are never live by missing expiry alone)"
            ),
            "pid_alive": False,
            "pid_missing": True,
            "heartbeat_at": heartbeat_at.isoformat() if heartbeat_at else None,
            "expires_at": expires_at.isoformat() if expires_at else None,
        }

    if pid_int is not None and not pid_alive:
        return {
            "status": "stale",
            "live": False,
            "stale": True,
            "reason": f"owner pid {pid_int} is not alive",
            "pid_alive": False,
            "pid_missing": False,
        }

    return {
        "status": "live",
        "live": True,
        "stale": False,
        "reason": "lock heartbeat and lease are fresh",
        "pid_alive": pid_alive,
        "pid_missing": False,
        "heartbeat_at": heartbeat_at.isoformat() if heartbeat_at else None,
        "expires_at": expires_at.isoformat() if expires_at else None,
    }


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,
        }
    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,
    recovery_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)
    )
    if recovery_sanctioned and existing_issue == issue_number and existing_branch == branch_name:
        return None
    if is_lease_expired(existing_lock, now=now):
        # #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).
            return None
        return (
            f"Issue #{issue_number} has an expired {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 _lock_claimant(lock: dict[str, Any] | None) -> dict[str, str]:
    if not isinstance(lock, dict):
        return {}
    claimant = lock.get("claimant")
    if not isinstance(claimant, dict):
        lease = lock.get("work_lease")
        claimant = lease.get("claimant") if isinstance(lease, dict) else None
    if not isinstance(claimant, dict):
        return {}
    return {
        "username": str(claimant.get("username") or ""),
        "profile": str(claimant.get("profile") or ""),
    }


def assess_foreign_lock_overwrite(
    existing_lock: dict[str, Any] | None,
    incoming_lock: dict[str, Any],
    *,
    recovery_sanctioned: bool = False,
    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

    existing_claimant = _lock_claimant(existing_lock)
    incoming_claimant = _lock_claimant(incoming_lock)
    same_claimant = (
        bool(existing_claimant.get("username"))
        and existing_claimant.get("username") == incoming_claimant.get("username")
        and existing_claimant.get("profile") == incoming_claimant.get("profile")
    )

    if recovery_sanctioned and same_issue and same_branch and same_claimant:
        return None

    if not is_lease_live(existing_lock, now=now):
        # #860 F8: A non-live or PID-less lock still blocks foreign overwrite
        # unless same claimant or sanctioned reclaim is proven.
        if not same_claimant and same_issue:
            reclaim = assess_expired_lock_reclaim(existing_lock, now=now)
            if not reclaim.get("reclaim_allowed"):
                return (
                    "Refusing foreign overwrite of non-live issue lock "
                    f"(issue #{existing_lock.get('issue_number')}, owner '{existing_claimant.get('username')}') "
                    "without sanctioned reclaim proof (fail closed)"
                )
        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)


# ── #871: durable linked-issue lock head refresh after branch synchronization ──
_FULL_SHA_RE = re.compile(r"^[0-9a-f]{40}$", re.IGNORECASE)

# Provenance recorded on the lock when the head is refreshed by a sanctioned
# merge-based branch synchronization (``gitea_update_pr_branch_by_merge``).
LOCK_HEAD_REFRESH_PROVENANCE_MERGE_SYNC = "gitea_update_pr_branch_by_merge"


def _norm_sha(value: Any) -> str | None:
    text = str(value or "").strip().lower()
    return text if _FULL_SHA_RE.match(text) else None


def _lock_claimant_view(lock: dict[str, Any] | None) -> dict[str, Any]:
    if not isinstance(lock, dict):
        return {}
    claimant = lock.get("claimant")
    if not isinstance(claimant, dict):
        lease = lock.get("work_lease")
        claimant = lease.get("claimant") if isinstance(lease, dict) else None
    return dict(claimant) if isinstance(claimant, dict) else {}


def assess_durable_lock_head_refresh(
    existing_lock: dict[str, Any] | None,
    *,
    remote: str,
    org: str,
    repo: str,
    issue_number: int,
    branch_name: str,
    worktree_path: str,
    pr_number: int | None,
    identity: str | None,
    profile: str | None,
    current_pid: int | None,
    expected_old_head: str | None,
    new_head: str | None,
    base_head: str | None = None,
) -> dict[str, Any]:
    """Fail-closed assessment for refreshing a durable lock's recorded head (#871).

    A successful ``gitea_update_pr_branch_by_merge`` advances the *remote* PR head
    but must also advance the durable linked-issue lock so a later dead-session
    recovery can prove ownership. This decides whether that refresh is permitted;
    it mutates nothing.

    Every element of durable ownership is re-verified against the persisted lock —
    repository, issue, branch, worktree, claimant identity/profile, and the live
    owning session — and the recorded head is compare-and-swapped: the lock's
    currently recorded synced head (if any) must equal ``expected_old_head``, so a
    lock whose head or provenance changed concurrently is never overwritten.
    """
    reasons: list[str] = []
    old = _norm_sha(expected_old_head)
    new = _norm_sha(new_head)
    evidence: dict[str, Any] = {
        "issue_number": issue_number,
        "branch_name": branch_name,
        "worktree_path": worktree_path,
        "pr_number": pr_number,
        "expected_old_head": old,
        "new_head": new,
        "base_head": _norm_sha(base_head),
    }

    if not isinstance(existing_lock, dict) or not existing_lock:
        reasons.append("no durable lock exists for this issue; nothing to refresh")
        return {"allowed": False, "reasons": reasons, "evidence": evidence,
                "expected_generation": 0}

    lock = dict(existing_lock)
    evidence["current_generation"] = lock_generation(lock)

    if lock.get("issue_number") != issue_number:
        reasons.append(
            f"durable lock targets issue #{lock.get('issue_number')}, not "
            f"#{issue_number}; refusing head refresh"
        )

    for field, expected in (("remote", remote), ("org", org), ("repo", repo)):
        actual = str(lock.get(field) or "").strip()
        if actual != str(expected or "").strip():
            reasons.append(
                f"lock {field} '{actual}' does not match requested "
                f"'{str(expected or '').strip()}'"
            )

    locked_branch = str(lock.get("branch_name") or "").strip()
    if locked_branch != str(branch_name or "").strip():
        reasons.append(
            f"lock branch '{locked_branch}' does not match requested "
            f"'{str(branch_name or '').strip()}'"
        )

    locked_worktree = str(lock.get("worktree_path") or "").strip()
    try:
        same_wt = bool(locked_worktree) and bool(worktree_path) and (
            os.path.realpath(locked_worktree) == os.path.realpath(worktree_path)
        )
    except OSError:
        same_wt = locked_worktree == (worktree_path or "")
    if not same_wt:
        reasons.append(
            f"lock worktree '{locked_worktree}' does not match declared "
            f"'{str(worktree_path or '').strip()}'"
        )

    claimant = _lock_claimant_view(lock)
    locked_identity = str(claimant.get("username") or "").strip()
    locked_profile = str(claimant.get("profile") or "").strip()
    if not locked_identity or not locked_profile:
        reasons.append(
            "durable lock does not record a claimant identity/profile; "
            "ownership could not be proven for head refresh"
        )
    if not str(identity or "").strip() or not str(profile or "").strip():
        reasons.append(
            "active session identity/profile is unknown; ownership could not be "
            "proven for head refresh"
        )
    if locked_identity and str(identity or "").strip() and locked_identity != str(identity).strip():
        reasons.append(
            f"lock claimant '{locked_identity}' does not match active identity "
            f"'{str(identity).strip()}'"
        )
    if locked_profile and str(profile or "").strip() and locked_profile != str(profile).strip():
        reasons.append(
            f"lock profile '{locked_profile}' does not match active profile "
            f"'{str(profile).strip()}'"
        )

    # The refresh is written by the LIVE owning author session. A refresh is not
    # a recovery: the current process must be the recorded owner.
    recorded_pid = lock.get("session_pid")
    if recorded_pid is None:
        recorded_pid = lock.get("pid")
    evidence["recorded_pid"] = recorded_pid
    evidence["current_pid"] = current_pid
    if current_pid is None:
        reasons.append("current session pid is unknown; cannot prove live ownership")
    else:
        try:
            if recorded_pid is None or int(recorded_pid) != int(current_pid):
                reasons.append(
                    f"durable lock is owned by pid {recorded_pid}, not the current "
                    f"session pid {current_pid}; head refresh requires the live owner"
                )
        except (TypeError, ValueError):
            reasons.append(
                "durable lock owner pid is malformed; cannot prove live ownership"
            )

    if not old:
        reasons.append("expected_old_head is not a full 40-char hex SHA (fail closed)")
    if not new:
        reasons.append("new_head is not a full 40-char hex SHA (fail closed)")
    if old and new and old == new:
        reasons.append(
            "new head equals the expected old head; a sync must advance the head"
        )

    # Compare-and-swap on the recorded head: if the lock already records a synced
    # head it must be exactly the expected old head, else another sync moved it.
    recorded_synced = _norm_sha(lock.get("synced_pr_head"))
    evidence["recorded_synced_pr_head"] = recorded_synced
    if recorded_synced is not None and old is not None and recorded_synced != old:
        reasons.append(
            f"durable lock already records synced head {recorded_synced}, not the "
            f"expected old head {old}; a concurrent sync changed it (CAS fail closed)"
        )

    if reasons:
        return {"allowed": False, "reasons": reasons, "evidence": evidence,
                "expected_generation": lock_generation(lock)}

    return {
        "allowed": True,
        "reasons": [
            f"durable lock for issue #{issue_number} branch '{locked_branch}' is "
            f"owned by the live session; refresh recorded head {old} -> {new}"
        ],
        "evidence": evidence,
        "expected_generation": lock_generation(lock),
    }


def apply_durable_lock_head_refresh(
    *,
    remote: str,
    org: str,
    repo: str,
    issue_number: int,
    branch_name: str,
    worktree_path: str,
    pr_number: int | None,
    identity: str | None,
    profile: str | None,
    current_pid: int | None,
    expected_old_head: str | None,
    new_head: str | None,
    synced_at: str,
    base_head: str | None = None,
    provenance: str = LOCK_HEAD_REFRESH_PROVENANCE_MERGE_SYNC,
    lock_dir: str | None = None,
) -> dict[str, Any]:
    """CAS-refresh the durable lock's recorded head after a branch sync (#871).

    Reads the durable lock from disk, re-asserts ownership via
    ``assess_durable_lock_head_refresh``, and — only when permitted — writes the
    new synced head through ``bind_session_lock`` with a generation compare-and-
    swap. Then re-reads the lock and proves it records the complete new head
    (read-after-write). Any failure at any step returns ``refreshed=False`` with
    reasons; the caller must treat that as a partial lifecycle failure and never
    report a fully successful synchronization.
    """
    existing = load_issue_lock(
        remote=remote, org=org, repo=repo, issue_number=issue_number, lock_dir=lock_dir
    )
    assessment = assess_durable_lock_head_refresh(
        existing,
        remote=remote,
        org=org,
        repo=repo,
        issue_number=issue_number,
        branch_name=branch_name,
        worktree_path=worktree_path,
        pr_number=pr_number,
        identity=identity,
        profile=profile,
        current_pid=current_pid,
        expected_old_head=expected_old_head,
        new_head=new_head,
        base_head=base_head,
    )
    result: dict[str, Any] = {
        "refreshed": False,
        "read_after_write_ok": False,
        "prior_head": _norm_sha(expected_old_head),
        "new_head": _norm_sha(new_head),
        "reasons": list(assessment.get("reasons") or []),
        "evidence": assessment.get("evidence"),
    }
    if not assessment.get("allowed"):
        return result

    new = _norm_sha(new_head)
    old = _norm_sha(expected_old_head)
    record = dict(existing or {})
    sync_block = {
        "last_synced_pr_head": new,
        "prior_pr_head": old,
        "base_head": _norm_sha(base_head),
        "pr_number": pr_number,
        "provenance": provenance,
        "synced_at": synced_at,
        "synced_by_pid": current_pid,
        "synced_by": {
            "username": str(identity or "").strip() or None,
            "profile": str(profile or "").strip() or None,
        },
    }
    record["synced_pr_head"] = new
    record["branch_sync"] = sync_block
    history = record.get("branch_sync_history")
    if not isinstance(history, list):
        history = []
    history = list(history)
    history.append(sync_block)
    record["branch_sync_history"] = history

    try:
        bind_session_lock(
            record,
            lock_dir=lock_dir,
            expected_generation=assessment.get("expected_generation"),
            renewal_sanctioned=True,
        )
    except Exception as exc:  # CAS miss or write failure — partial lifecycle failure
        result["reasons"].append(
            f"durable lock head refresh write failed (fail closed): {exc}"
        )
        return result

    after = load_issue_lock(
        remote=remote, org=org, repo=repo, issue_number=issue_number, lock_dir=lock_dir
    )
    after_head = _norm_sha((after or {}).get("synced_pr_head"))
    result["lock_generation_after"] = lock_generation(after)
    if after_head == new and new is not None:
        result["refreshed"] = True
        result["read_after_write_ok"] = True
        result["reasons"].append(
            f"durable lock recorded head refreshed to {new} and verified by "
            "read-after-write"
        )
    else:
        result["reasons"].append(
            "read-after-write verification failed: durable lock does not record "
            f"the new head {new} (found {after_head}); partial lifecycle failure"
        )
    return result