Merge pull request 'feat(mcp): use a 10-minute sliding TTL for reviewer and merger PR leases (Closes #747)' (#748) from feat/issue-747-sliding-lease-ttl into master

Reviewer and merger PR leases minted a fixed 120-minute expiry (#407 AC6/AC7)
and derived staleness from two further activity bands: stale at 30 minutes,
reclaimable at 60. A session that died — daemon crash, transport flap, client
restart — therefore kept a PR blocked for an hour before anyone could reclaim
it, and two hours before manual cleanup was sanctioned. #718 records the
resulting deadlock: a merge stalled behind a reviewer lease that had stopped
being live long before it stopped being authoritative.

The flaw was using a long fixed expiry as a proxy for "the owner is probably
still alive" instead of making the owner continuously prove liveness.

Sliding window
--------------
`LEASE_TTL_MINUTES = 10` now governs acquisition, and every write of the lease
marker re-derives `expires_at` from the moment of the write, so each heartbeat
slides the window forward. An actively heartbeating session is never evicted
and has no maximum lifetime; a dead one releases its hold within one TTL.

`LEASE_RENEWAL_MINUTES` is named separately from the acquisition TTL. Renewal
previously had no seam at all: the heartbeat slid the expiry only as a side
effect of re-defaulting the acquisition constant, so the two durations could
not be reasoned about or tuned independently. `format_lease_body` now takes an
explicit `ttl_minutes`, and the heartbeat passes the renewal window rather than
relying on that default.

Merger leases acquire through the same lease-body formatter, so they inherit
the identical window by construction rather than by a parallel constant.

Removal of the reclaim tier
---------------------------
`classify_lease_freshness` no longer returns `reclaimable`. Beyond being an
extra waiting tier, that band is unreachable under a sliding TTL: a heartbeat
stamps `last_activity` and `expires_at` together, so a lease idle for a full
TTL is necessarily already expired. Expiry is now the only takeover gate.

No reclaim path is lost. `find_active_reviewer_lease` already ignores expired
markers, so an expired foreign lease never gated acquisition; and the
`foreign_expired` classification carries the same
`NEXT_ACTION_RELEASE_EXPIRED_LEASE` the retired `foreign_reclaimable` did. The
two updated tests in `test_reviewer_pr_lease.py` assert exactly that: the
classification label changes, the sanctioned next action does not.

`STALE_WARNING_MINUTES` drops to 5 — half the window — so the warning still
fires while the owner can heartbeat and recover.

Diagnostics
-----------
Adds `lease_seconds_remaining`, and the heartbeat tool now returns
`ttl_minutes`, `expires_at`, and `seconds_remaining`, so an operator can
distinguish "held and live" from "held and dying" instead of only seeing that
a lease exists. All additions are additive; no existing key changed.

Also removes `pr_work_lease.DEFAULT_REVIEWER_LEASE_TTL_MINUTES`, a duplicate of
the reviewer TTL with no readers anywhere in the tree, which could only drift.

Legacy markers minted under the old 120-minute TTL still parse and are judged
against their own recorded `expires_at`, so no lease is retroactively expired
by this change.

Full suite: 3425 passed, 6 skipped, 2 failed. Both failures
(`test_issue_702_review_findings_f1_f6`, `test_reconciler_supersession_close`)
reproduce identically on clean master b05075fd25 and are pre-existing.

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
Claude-Session: https://claude.ai/code/session_01QxXHZ7rqXtLgTusngaWgKZ

docs/gitea-execution-profiles.md

@@ -317,44 +317,6 @@ Least-privilege constraints:
   canonical names such as `gitea.pr.close` (never bare `pr.close` /
   `issue.close`, which the production normalizer rejects or drops).
 
-### Post-merge moot-lease cleanup ownership (`gitea.pr.comment`)
-
-Neutralising a reviewer lease left behind on an already-merged/closed PR is
-reconciliation work too. `task_capability_map` maps
-`cleanup_post_merge_moot_lease` — and its tool-name alias
-`gitea_cleanup_post_merge_moot_lease` — to role `reconciler` with permission
-`gitea.pr.comment` (#745). Both names carry the **same** contract.
-
-The permission alone is deliberately not sufficient: author, reviewer and
-merger profiles all hold `gitea.pr.comment` for ordinary PR discussion, so the
-role gate — not the permission gate — is what keeps the terminal lease marker
-reconciler-owned.
-
-`gitea_cleanup_post_merge_moot_lease` splits its two modes on purpose:
-
-- **`apply=false` (assessment) requires only `gitea.read`, with no role gate.**
-  This matches `gitea_cleanup_stale_review_decision_lock` and
-  `gitea_cleanup_obsolete_reviewer_comment_lease`, whose assessment paths are
-  likewise read-gated, so an operator can diagnose a stuck lease from whichever
-  namespace happens to be attached without switching roles. The dry run
-  performs no mutation and records append-only evidence in-session.
-- **`apply=true` (mutation) requires all of the following**, in order: the
-  session must have resolved exactly `cleanup_post_merge_moot_lease` (resolving
-  any other task — including a sibling reconciler task — does not authorize
-  it); the active role must be `reconciler`; the profile must hold
-  `gitea.pr.comment`; the explicit `org`/`repo` must agree with the canonical
-  repository identity, which is derived from the session binding and can never
-  be overridden by request parameters; and matching dry-run evidence must show
-  `lease_moot`, `cleanup_allowed`, and the same PR, lease session, candidate
-  head and lease marker id that are live at apply time.
-
-Everything else fails closed: a live lease on an open PR, an already-terminal
-(idempotent) lease, a lease superseded between the dry run and the apply, a
-malformed lease missing session/head/marker, and any foreign-repository target.
-The cleanup only ever appends a terminal `phase: released` marker
-(`blocker: post-merge-moot`) — it never edits or deletes another session's
-comment, and it never merges or adopts a lease.
-
 Launch a static `gitea-reconciler` MCP namespace with
 `GITEA_MCP_PROFILE=prgs-reconciler`. Profile shape is validated by
 `reconciler_profile.assess_reconciler_profile` (#304). Use the

gitea_mcp_server.py

@@ -1829,7 +1829,6 @@ def _seed_session_context(
 import issue_work_duplicate_gate  # noqa: E402
 import issue_workflow_labels  # noqa: E402
 import reviewer_pr_lease  # noqa: E402
-import post_merge_moot_lease_gate  # noqa: E402  # #745 reconciler cleanup gate
 import merger_lease_adoption  # noqa: E402
 import merged_cleanup_reconcile  # noqa: E402
 import worktree_cleanup_audit  # noqa: E402
@@ -8860,14 +8859,13 @@ def gitea_review_pr(
         return out
 
 
-def _repository_binding_block(
+def _delete_branch_repository_binding_block(
     remote: str | None,
     *,
     org: str | None,
     repo: str | None,
-    required_permission: str = "gitea.branch.delete",
 ) -> dict | None:
-    """#733: validate an explicit mutation target against the workspace binding.
+    """#733: validate an explicit delete target against the workspace binding.
 
     The trusted repository identity comes only from the verified,
     workspace-aligned git remote (never ``REMOTES`` defaults, never
@@ -8900,7 +8898,7 @@ def _repository_binding_block(
         return {
             "success": False,
             "performed": False,
-            "required_permission": required_permission,
+            "required_permission": "gitea.branch.delete",
             "reasons": canonical_reasons,
             "blocker_kind": "repository_binding",
         }
@@ -8919,9 +8917,9 @@ def _repository_binding_block(
             return {
                 "success": False,
                 "performed": False,
-                "required_permission": required_permission,
+                "required_permission": "gitea.branch.delete",
                 "reasons": list(override.get("reasons") or [
-                    "target repository does not match the workspace "
+                    "delete target repository does not match the workspace "
                     "binding (fail closed)"
                 ]),
                 "blocker_kind": "repository_binding",
@@ -8931,51 +8929,17 @@ def _repository_binding_block(
         return {
             "success": False,
             "performed": False,
-            "required_permission": required_permission,
+            "required_permission": "gitea.branch.delete",
             "reasons": [
                 "repository binding unverified: no workspace repository "
                 "identity could be established to corroborate the explicit "
-                "target (fail closed)"
+                "delete target (fail closed)"
             ],
             "blocker_kind": "repository_binding",
         }
     return None
 
 
-def _delete_branch_repository_binding_block(
-    remote: str | None,
-    *,
-    org: str | None,
-    repo: str | None,
-) -> dict | None:
-    """Delete-branch view of :func:`_repository_binding_block` (#733).
-
-    Retained as the named entry point for the delete path so #733/#739
-    regression coverage keeps exercising the exact permission label that
-    ``gitea_delete_branch`` reports.
-    """
-    return _repository_binding_block(
-        remote, org=org, repo=repo,
-        required_permission="gitea.branch.delete",
-    )
-
-
-def _bound_repository_slug(remote: str | None) -> str | None:
-    """Canonical repository slug for the active session, or None (#745).
-
-    Same trust order as ``_repository_binding_block``: the configured canonical
-    root when the namespace declares one, otherwise the workspace-derived
-    identity. Request parameters are never consulted, so they cannot redirect a
-    mutation at a foreign repository.
-    """
-    canonical_slug, canonical_reasons = _canonical_repository_slug(
-        get_profile(), remote
-    )
-    if canonical_reasons:
-        return None
-    return canonical_slug or _workspace_repository_slug(remote)
-
-
 @mcp.tool()
 def gitea_delete_branch(
     branch: str,
@@ -12168,6 +12132,13 @@ def gitea_heartbeat_reviewer_pr_lease(
     verify_preflight_purity(remote, task="review_pr")
     h, o, r = _resolve(remote, host, org, repo)
     auth = _auth(h)
+    # Slide the sliding TTL forward (#747): the heartbeat is the liveness proof,
+    # so renewal is stated explicitly rather than inherited from the acquisition
+    # default.
+    beat_at = datetime.now(timezone.utc)
+    renewed_expiry = beat_at + timedelta(
+        minutes=reviewer_pr_lease.LEASE_RENEWAL_MINUTES
+    )
     body = reviewer_pr_lease.format_lease_body(
         repo=f"{o}/{r}",
         pr_number=pr_number,
@@ -12180,6 +12151,8 @@ def gitea_heartbeat_reviewer_pr_lease(
         candidate_head=candidate_head or session.get("candidate_head"),
         target_branch=session.get("target_branch") or "master",
         target_branch_sha=target_branch_sha or session.get("target_branch_sha"),
+        last_activity=beat_at,
+        ttl_minutes=reviewer_pr_lease.LEASE_RENEWAL_MINUTES,
     )
     comment_url = f"{repo_api_url(h, o, r)}/issues/{pr_number}/comments"
     with _audited(
@@ -12220,6 +12193,13 @@ def gitea_heartbeat_reviewer_pr_lease(
         "phase": phase,
         "comment_id": posted.get("id"),
         "session_lease": updated,
+        # Report the renewed window so an operator can tell "held and live"
+        # from "held and dying" (#747).
+        "ttl_minutes": reviewer_pr_lease.LEASE_RENEWAL_MINUTES,
+        "expires_at": renewed_expiry.replace(microsecond=0)
+        .isoformat()
+        .replace("+00:00", "Z"),
+        "seconds_remaining": reviewer_pr_lease.LEASE_RENEWAL_MINUTES * 60,
         "reasons": [],
     }
 
@@ -12388,10 +12368,6 @@ def gitea_diagnose_reviewer_pr_lease_handoff(
 def gitea_cleanup_post_merge_moot_lease(
     pr_number: int,
     apply: bool = False,
-    expected_session_id: str | None = None,
-    expected_candidate_head: str | None = None,
-    expected_lease_comment_id: int | None = None,
-    worktree_path: str | None = None,
     remote: str = "dadeschools",
     host: str | None = None,
     org: str | None = None,
@@ -12407,37 +12383,14 @@ def gitea_cleanup_post_merge_moot_lease(
     an append-only comment that neutralises the moot lease without deleting any
     other session's comment.
 
-    Role binding (#745). The two modes are gated differently, on purpose:
-
-    * ``apply=false`` (assessment) stays reachable under ``gitea.read`` for
-      **any** role, matching ``gitea_cleanup_stale_review_decision_lock`` and
-      ``gitea_cleanup_obsolete_reviewer_comment_lease``, so an operator can
-      diagnose a stuck lease from whichever namespace is attached. It performs
-      no mutation and records append-only dry-run evidence in-session.
-    * ``apply=true`` (mutation) additionally requires, in order: the session to
-      have resolved exactly ``cleanup_post_merge_moot_lease``; the
-      ``reconciler`` role; ``gitea.pr.comment``; a validated repository /
-      canonical-root / workspace binding; and matching dry-run evidence proving
-      ``lease_moot`` and ``cleanup_allowed`` for the same PR, lease session,
-      candidate head and lease marker. Author, reviewer and merger fail closed
-      here even though their profiles carry ``gitea.pr.comment``.
-
     Args:
         pr_number: The PR whose lingering lease to assess/clean.
         apply: When false (default) report only (read-only). When true, post the
-            terminal released marker if — and only if — cleanup is authorized.
-        expected_session_id: Optional lease session the caller expects; a
-            mismatch against the live lease fails closed.
-        expected_candidate_head: Optional leased head the caller expects; a
-            mismatch against the live lease fails closed.
-        expected_lease_comment_id: Optional lease marker id the caller expects;
-            a mismatch against the live lease fails closed.
-        worktree_path: Reconciler worktree to bind the apply-path preflight to.
+            terminal released marker if — and only if — cleanup is allowed.
         remote: Known instance — 'dadeschools' or 'prgs'.
         host: Override the Gitea host.
-        org: Override the owner/organization. Validated against the canonical
-            repository identity; it can never redirect the mutation.
-        repo: Override the repository name. Validated as above.
+        org: Override the owner/organization.
+        repo: Override the repository name.
 
     Returns:
         dict reporting PR merged/closed state, merge_commit_sha, linked-issue
@@ -12497,62 +12450,14 @@ def gitea_cleanup_post_merge_moot_lease(
         "reasons": assessment.get("reasons") or [],
     }
 
-    # The canonical repository identity comes from the session binding only —
-    # never from the org/repo request parameters (#745 requirement 10).
-    repository_slug = _bound_repository_slug(remote)
-    report["repository_slug"] = repository_slug
-    report["required_task"] = post_merge_moot_lease_gate.CLEANUP_TASK
-    report["required_role_kind"] = post_merge_moot_lease_gate.REQUIRED_ROLE
-
     if not apply:
-        # Append-only dry-run evidence. Recorded for every assessment, allowed
-        # or not, so a later apply can prove the lease it saw is the lease that
-        # is still there. Prior entries are never rewritten.
-        evidence = post_merge_moot_lease_gate.record_dry_run(
-            pr_number=pr_number,
-            repository_slug=repository_slug,
-            lease_moot=bool(assessment.get("is_moot")),
-            cleanup_allowed=bool(assessment.get("cleanup_allowed")),
-            session_id=active.get("session_id"),
-            candidate_head=active.get("candidate_head"),
-            lease_comment_id=active.get("comment_id"),
-        )
-        report["dry_run_evidence"] = evidence
         return report
-
-    # ---------------------------------------------------------------- apply --
-    # Preserve the existing dry-run safety assessment: a lease that is not moot
-    # (open PR, already-terminal, absent) is refused here exactly as before, and
-    # no mutation is attempted.
     if not assessment.get("cleanup_allowed"):
         report["cleanup_skipped_reason"] = (
             assessment.get("reasons") or ["cleanup not allowed"]
         )
         return report
 
-    # 1 + 2 + 6: exact resolved cleanup task, reconciler role, and matching
-    # append-only dry-run evidence. Permission alone is deliberately not enough.
-    authorization = post_merge_moot_lease_gate.assess_apply_authorization(
-        pr_number=pr_number,
-        repository_slug=repository_slug,
-        resolved_task=_preflight_resolved_task,
-        active_role_kind=_profile_role_kind(get_profile()),
-        assessment=assessment,
-        evidence=post_merge_moot_lease_gate.latest_dry_run(
-            pr_number=pr_number, repository_slug=repository_slug
-        ),
-        expected_session_id=expected_session_id,
-        expected_candidate_head=expected_candidate_head,
-        expected_lease_comment_id=expected_lease_comment_id,
-    )
-    report["authorization"] = authorization
-    if not authorization["allowed"]:
-        report["success"] = False
-        report["reasons"] = authorization["reasons"]
-        report["blocker_kind"] = authorization["blocker_kind"]
-        return report
-
-    # 3: the dedicated mutation permission.
     comment_block = _profile_operation_gate("gitea.pr.comment")
     if comment_block:
         report["success"] = False
@@ -12560,28 +12465,7 @@ def gitea_cleanup_post_merge_moot_lease(
         report["permission_report"] = _permission_block_report("gitea.pr.comment")
         return report
 
-    # 4: explicit target must agree with the canonical repository identity
-    # before any preflight or mutation (#733/#739).
-    repo_binding_block = _repository_binding_block(
-        remote, org=org, repo=repo,
-        required_permission="gitea.pr.comment",
-    )
-    if repo_binding_block:
-        report["success"] = False
-        report["reasons"] = repo_binding_block["reasons"]
-        report["blocker_kind"] = repo_binding_block["blocker_kind"]
-        return report
-
-    # 4 (cont.): canonical root, workspace binding and the resolved-task match
-    # are enforced by the shared preflight, with the explicit target forwarded
-    # so the #604 anti-stomp resolution validates the targeted repository.
-    verify_preflight_purity(
-        remote,
-        worktree_path=worktree_path,
-        task=post_merge_moot_lease_gate.CLEANUP_TASK,
-        org=org,
-        repo=repo,
-    )
+    verify_preflight_purity(remote)
     body = assessment["release_body"]
     comment_url = f"{repo_api_url(h, o, r)}/issues/{pr_number}/comments"
     with _audited(

post_merge_moot_lease_gate.py

@@ -1,279 +0,0 @@
-"""Reconciler authorization gate for post-merge moot-lease cleanup (#745).
-
-``gitea_cleanup_post_merge_moot_lease`` (#515) posts a terminal ``phase:
-released`` lease marker — a real, durable mutation of the PR lease ledger.
-Before #745 it was gated on permissions alone (``gitea.read`` to enter,
-``gitea.pr.comment`` to apply) with no canonical task and no role binding, so
-any profile carrying ``gitea.pr.comment`` reached the mutation path while the
-reconciler could not satisfy the operator-required resolve-exact-task ->
-mutation sequence.
-
-This module holds the pure half of that gate:
-
-* the canonical task name and its tool-name alias;
-* an **append-only** in-process ledger of read-only dry-run assessments;
-* ``assess_apply_authorization``, which decides whether an apply may proceed.
-
-Apply is authorized only when all of the following hold:
-
-* the session resolved exactly the cleanup task (no other task substitutes);
-* the active profile role is ``reconciler``;
-* a prior dry run in this session recorded ``lease_moot`` and
-  ``cleanup_allowed`` for the *same* repository, PR, lease session, candidate
-  head and lease marker id;
-* the live assessment still agrees with that evidence, so a lease superseded
-  between the dry run and the apply fails closed;
-* any caller-supplied expectations match the live lease exactly.
-
-Everything else fails closed. The ledger is only ever appended to — a
-superseded dry run stays visible as history instead of being rewritten — which
-keeps the cleanup audit trail append-only end to end.
-"""
-
-from __future__ import annotations
-
-from datetime import datetime, timezone
-from typing import Any
-
-CLEANUP_TASK = "cleanup_post_merge_moot_lease"
-CLEANUP_TOOL_ALIAS = "gitea_cleanup_post_merge_moot_lease"
-REQUIRED_ROLE = "reconciler"
-REQUIRED_PERMISSION = "gitea.pr.comment"
-
-# The read-only assessment stays reachable under gitea.read for every role —
-# the convention shared with cleanup_stale_review_decision_lock and
-# cleanup_obsolete_reviewer_comment_lease — so any namespace can diagnose a
-# stuck lease. Only the apply path demands CLEANUP_TASK + REQUIRED_ROLE.
-ASSESSMENT_PERMISSION = "gitea.read"
-
-_DRY_RUN_LEDGER: list[dict[str, Any]] = []
-
-
-def _norm(value: Any) -> str:
-    return str(value or "").strip()
-
-
-def _norm_comment_id(value: Any) -> int | None:
-    try:
-        return int(value)
-    except (TypeError, ValueError):
-        return None
-
-
-def record_dry_run(
-    *,
-    pr_number: int,
-    repository_slug: str | None,
-    lease_moot: bool,
-    cleanup_allowed: bool,
-    session_id: str | None,
-    candidate_head: str | None,
-    lease_comment_id: Any,
-    recorded_at: datetime | None = None,
-) -> dict[str, Any]:
-    """Append one read-only assessment to the dry-run ledger.
-
-    Never rewrites or removes a prior entry: repeated dry runs accumulate and
-    ``latest_dry_run`` returns the newest matching one.
-    """
-    entry = {
-        "task": CLEANUP_TASK,
-        "pr_number": int(pr_number),
-        "repository_slug": _norm(repository_slug) or None,
-        "lease_moot": bool(lease_moot),
-        "cleanup_allowed": bool(cleanup_allowed),
-        "session_id": _norm(session_id) or None,
-        "candidate_head": _norm(candidate_head) or None,
-        "lease_comment_id": _norm_comment_id(lease_comment_id),
-        "recorded_at": (recorded_at or datetime.now(timezone.utc)).isoformat(),
-    }
-    _DRY_RUN_LEDGER.append(entry)
-    return dict(entry)
-
-
-def dry_run_history() -> tuple[dict[str, Any], ...]:
-    """Immutable view of every recorded dry run, oldest first."""
-    return tuple(dict(entry) for entry in _DRY_RUN_LEDGER)
-
-
-def latest_dry_run(
-    *, pr_number: int, repository_slug: str | None
-) -> dict[str, Any] | None:
-    """Newest dry-run evidence for this repository + PR, or None."""
-    wanted_repo = _norm(repository_slug)
-    for entry in reversed(_DRY_RUN_LEDGER):
-        if entry["pr_number"] != int(pr_number):
-            continue
-        if _norm(entry.get("repository_slug")) != wanted_repo:
-            continue
-        return dict(entry)
-    return None
-
-
-def _reset_for_testing() -> None:
-    """Drop ledger state between tests. Never called by production paths."""
-    _DRY_RUN_LEDGER.clear()
-
-
-def assess_apply_authorization(
-    *,
-    pr_number: int,
-    repository_slug: str | None,
-    resolved_task: str | None,
-    active_role_kind: str | None,
-    assessment: dict[str, Any],
-    evidence: dict[str, Any] | None,
-    expected_session_id: str | None = None,
-    expected_candidate_head: str | None = None,
-    expected_lease_comment_id: Any = None,
-) -> dict[str, Any]:
-    """Decide whether a moot-lease cleanup apply is authorized (fail closed).
-
-    Returns ``{"allowed", "reasons", "blocker_kind", "evidence_matched", ...}``.
-    ``allowed`` is True only when every check passes; each failure contributes a
-    reason so the caller can report all of them together.
-    """
-    reasons: list[str] = []
-    blocker_kind: str | None = None
-
-    def _block(kind: str, reason: str) -> None:
-        nonlocal blocker_kind
-        reasons.append(reason)
-        if blocker_kind is None:
-            blocker_kind = kind
-
-    # 1. Exact resolved cleanup task. Resolving any other task — including a
-    #    sibling reconciler task — does not authorize this mutation.
-    if _norm(resolved_task) != CLEANUP_TASK:
-        _block(
-            "unresolved_cleanup_task",
-            "post-merge moot-lease cleanup requires the session to resolve "
-            f"task '{CLEANUP_TASK}' immediately before apply; resolved task is "
-            f"{resolved_task!r} (fail closed)",
-        )
-
-    # 2. Dedicated reconciler role, enforced independently of the permission.
-    if _norm(active_role_kind) != REQUIRED_ROLE:
-        _block(
-            "wrong_role",
-            f"profile role {active_role_kind!r} cannot apply post-merge "
-            f"moot-lease cleanup; required role is {REQUIRED_ROLE} even when "
-            f"{REQUIRED_PERMISSION} is present (fail closed)",
-        )
-
-    # 3. Canonical repository identity must be established, never inferred from
-    #    request parameters.
-    if not _norm(repository_slug):
-        _block(
-            "repository_binding",
-            "no canonical repository identity could be established for the "
-            "cleanup target (fail closed)",
-        )
-
-    # 4. The live safety assessment must still say the lease is moot/cleanable.
-    if not assessment.get("is_moot") or not assessment.get("cleanup_allowed"):
-        _block(
-            "lease_not_moot",
-            "live assessment does not report a moot, cleanable lease on PR "
-            f"#{pr_number} (lease_moot={bool(assessment.get('is_moot'))}, "
-            f"cleanup_allowed={bool(assessment.get('cleanup_allowed'))}) "
-            "(fail closed)",
-        )
-
-    live = assessment.get("active_lease") or {}
-    live_session = _norm(live.get("session_id"))
-    live_head = _norm(live.get("candidate_head"))
-    live_comment_id = _norm_comment_id(live.get("comment_id"))
-
-    # 5. A lease missing identifying fields is malformed and unsafe to act on.
-    if not live_session or not live_head or live_comment_id is None:
-        _block(
-            "malformed_lease",
-            "active lease is malformed: session_id / candidate_head / "
-            "comment_id must all be present to authorize cleanup "
-            f"(session_id={live.get('session_id')!r}, "
-            f"candidate_head={live.get('candidate_head')!r}, "
-            f"comment_id={live.get('comment_id')!r}) (fail closed)",
-        )
-
-    # 6. Caller expectations, when supplied, must match the live lease exactly.
-    if expected_session_id is not None and _norm(expected_session_id) != live_session:
-        _block(
-            "lease_mismatch",
-            f"expected lease session {expected_session_id!r} does not match the "
-            f"live lease session {live.get('session_id')!r} (fail closed)",
-        )
-    if (
-        expected_candidate_head is not None
-        and _norm(expected_candidate_head) != live_head
-    ):
-        _block(
-            "lease_mismatch",
-            f"expected candidate head {expected_candidate_head!r} does not "
-            f"match the live lease head {live.get('candidate_head')!r} "
-            "(fail closed)",
-        )
-    if expected_lease_comment_id is not None and (
-        _norm_comment_id(expected_lease_comment_id) != live_comment_id
-    ):
-        _block(
-            "lease_mismatch",
-            f"expected lease marker {expected_lease_comment_id!r} does not "
-            f"match the live lease marker {live.get('comment_id')!r} "
-            "(fail closed)",
-        )
-
-    # 7. Matching dry-run evidence recorded earlier in this session.
-    evidence_matched = False
-    if evidence is None:
-        _block(
-            "missing_dry_run_evidence",
-            "no read-only dry run recorded for this repository and PR; run the "
-            "tool with apply=false and confirm lease_moot / cleanup_allowed "
-            "before applying (fail closed)",
-        )
-    elif not evidence.get("lease_moot") or not evidence.get("cleanup_allowed"):
-        _block(
-            "dry_run_not_allowed",
-            "recorded dry run did not report an allowed cleanup "
-            f"(lease_moot={bool(evidence.get('lease_moot'))}, "
-            f"cleanup_allowed={bool(evidence.get('cleanup_allowed'))}) "
-            "(fail closed)",
-        )
-    elif int(evidence.get("pr_number") or -1) != int(pr_number) or _norm(
-        evidence.get("repository_slug")
-    ) != _norm(repository_slug):
-        _block(
-            "dry_run_mismatch",
-            "recorded dry run targets a different repository or PR "
-            f"({evidence.get('repository_slug')}#{evidence.get('pr_number')} vs "
-            f"{repository_slug}#{pr_number}) (fail closed)",
-        )
-    elif (
-        _norm(evidence.get("session_id")) != live_session
-        or _norm(evidence.get("candidate_head")) != live_head
-        or _norm_comment_id(evidence.get("lease_comment_id")) != live_comment_id
-    ):
-        _block(
-            "superseded_lease",
-            "the lease changed after the recorded dry run (dry run: "
-            f"session={evidence.get('session_id')!r}, "
-            f"head={evidence.get('candidate_head')!r}, "
-            f"marker={evidence.get('lease_comment_id')!r}; live: "
-            f"session={live.get('session_id')!r}, "
-            f"head={live.get('candidate_head')!r}, "
-            f"marker={live.get('comment_id')!r}); re-run the dry run "
-            "(fail closed)",
-        )
-    else:
-        evidence_matched = True
-
-    return {
-        "allowed": not reasons,
-        "reasons": reasons,
-        "blocker_kind": blocker_kind,
-        "evidence_matched": evidence_matched,
-        "required_task": CLEANUP_TASK,
-        "required_role_kind": REQUIRED_ROLE,
-        "required_permission": REQUIRED_PERMISSION,
-    }

pr_work_lease.py

@@ -36,7 +36,8 @@ _TERMINAL_CONFLICT_FIX_PHASES = frozenset({"released", "blocked", "done"})
 _ACTIVE_CONFLICT_FIX_PHASES = frozenset({"claimed", "pushing", "pushed"})
 
 DEFAULT_CONFLICT_FIX_TTL_MINUTES = 120
-DEFAULT_REVIEWER_LEASE_TTL_MINUTES = 120
+# The reviewer/merger PR-lease TTL lives in reviewer_pr_lease.LEASE_TTL_MINUTES
+# (#747). A second copy here had no readers and could only drift out of sync.
 
 
 def _parse_timestamp(value: str | None) -> datetime | None:

reviewer_pr_lease.py

@@ -29,9 +29,19 @@ _ACTIVE_PHASES = frozenset({
     "adopted",
 })
 
-DEFAULT_LEASE_TTL_MINUTES = 120
-STALE_WARNING_MINUTES = 30
-RECLAIMABLE_MINUTES = 60
+# Reviewer and merger PR leases use a short *sliding* window (#747): a lease
+# expires 10 minutes after its last heartbeat, and every heartbeat slides the
+# expiry forward. An actively heartbeating session is never evicted, while a
+# dead session releases its hold in at most one TTL instead of the two hours
+# the previous fixed 120-minute expiry allowed.
+LEASE_TTL_MINUTES = 10
+# Renewal is named separately from acquisition so the slide amount is tunable
+# without silently re-defining how long a fresh lease lives.
+LEASE_RENEWAL_MINUTES = 10
+# Retained for callers that imported the pre-#747 name.
+DEFAULT_LEASE_TTL_MINUTES = LEASE_TTL_MINUTES
+# Warn at half the window, while the owner can still heartbeat and recover.
+STALE_WARNING_MINUTES = 5
 
 _SESSION_LEASE: dict[str, Any] | None = None
 
@@ -80,10 +90,19 @@ def format_lease_body(
     target_branch_sha: str | None,
     last_activity: datetime | None = None,
     expires_at: datetime | None = None,
+    ttl_minutes: int = LEASE_TTL_MINUTES,
     blocker: str = "none",
 ) -> str:
+    """Serialize a lease marker.
+
+    Every write of this marker — acquisition, heartbeat, adoption — re-derives
+    ``expires_at`` from the moment of the write, which is what makes the TTL
+    slide (#747). Callers renewing an existing lease pass
+    ``ttl_minutes=LEASE_RENEWAL_MINUTES``; an explicit ``expires_at`` still
+    wins so a lease can be minted with a deliberate window.
+    """
     now = last_activity or datetime.now(timezone.utc)
-    expires = expires_at or (now + timedelta(minutes=DEFAULT_LEASE_TTL_MINUTES))
+    expires = expires_at or (now + timedelta(minutes=ttl_minutes))
     last_text = now.astimezone(timezone.utc).replace(microsecond=0).isoformat().replace(
         "+00:00", "Z"
     )
@@ -169,8 +188,30 @@ def _minutes_since_activity(lease: dict, *, now: datetime) -> float | None:
     return (now - last).total_seconds() / 60.0
 
 
+def lease_seconds_remaining(lease: dict, *, now: datetime | None = None) -> int | None:
+    """Seconds until *lease* expires, clamped at 0; ``None`` if unparsable.
+
+    Lets diagnostics distinguish "held and live" from "held and dying" (#747)
+    rather than only reporting that a lease exists.
+    """
+    expires_at = _parse_timestamp(lease.get("expires_at"))
+    if not expires_at:
+        return None
+    now = now or datetime.now(timezone.utc)
+    return max(0, int((expires_at - now).total_seconds()))
+
+
 def classify_lease_freshness(lease: dict, *, now: datetime | None = None) -> str:
-    """Return active, stale_warning, reclaimable, expired, or terminal."""
+    """Return active, stale_warning, expired, or terminal.
+
+    Expiry is the only takeover gate (#747). The pre-#747 ``reclaimable`` band
+    sat between "stale" and "expired" and made a dead lease wait out a second
+    timer before anyone could reclaim it. Under a sliding TTL that band is also
+    unreachable: a heartbeat stamps ``last_activity`` and ``expires_at``
+    together, so a lease idle for a full TTL is already expired. Foreign
+    expired leases are handled by the ``foreign_expired`` classification, which
+    carries the same sanctioned release next-action the old tier did.
+    """
     now = now or datetime.now(timezone.utc)
     phase = (lease.get("phase") or "").strip().lower()
     if phase in _TERMINAL_PHASES:
@@ -180,8 +221,6 @@ def classify_lease_freshness(lease: dict, *, now: datetime | None = None) -> str
     minutes = _minutes_since_activity(lease, now=now)
     if minutes is None:
         return "active"
-    if minutes >= RECLAIMABLE_MINUTES:
-        return "reclaimable"
     if minutes >= STALE_WARNING_MINUTES:
         return "stale_warning"
     return "active"

role_session_router.py

@@ -87,11 +87,6 @@ RECONCILER_TASKS = frozenset({
     # only to the reconciler profile). Raw gitea_delete_branch redirects here to
     # the guarded gitea_cleanup_merged_pr_branch path (#514/#687).
     "delete_branch",
-    # #745: post-merge moot reviewer-lease cleanup is reconciler-owned; the
-    # apply path posts a terminal lease marker. Kept in step with
-    # task_capability_map so map and router cannot disagree (#723 defect A).
-    "cleanup_post_merge_moot_lease",
-    "gitea_cleanup_post_merge_moot_lease",
     "reconcile_already_landed_pr",
     "reconcile_already_landed",
     "reconcile-landed-pr",
@@ -123,9 +118,6 @@ TASK_REQUIRED_ROLE = {
     "reconcile_already_landed": "reconciler",
     "reconcile-landed-pr": "reconciler",
     "cleanup_merged_pr_branch": "reconciler",
-    # #745: post-merge moot reviewer-lease cleanup (canonical task + tool alias).
-    "cleanup_post_merge_moot_lease": "reconciler",
-    "gitea_cleanup_post_merge_moot_lease": "reconciler",
     # #309: reconciler tasks close already-landed PRs/issues only.
     "reconcile_close_landed_pr": "reconciler",
     "reconcile_close_landed_issue": "reconciler",

task_capability_map.py

@@ -158,23 +158,6 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = {
         "permission": "gitea.pr.comment",
         "role": "reviewer",
     },
-    # #745: post-merge moot reviewer-lease cleanup is reconciler-owned. The
-    # apply path posts an append-only terminal `phase: released` lease marker
-    # (gitea.pr.comment), so holding the comment permission alone must not
-    # authorize it — author, reviewer and merger fail closed on the role gate
-    # even though their profiles carry gitea.pr.comment. The read-only
-    # `apply=false` assessment deliberately stays reachable under gitea.read
-    # inside the tool (the same convention as cleanup_stale_review_decision_lock
-    # below), so any namespace can diagnose a stuck lease; only apply requires
-    # this task plus the reconciler role.
-    "cleanup_post_merge_moot_lease": {
-        "permission": "gitea.pr.comment",
-        "role": "reconciler",
-    },
-    "gitea_cleanup_post_merge_moot_lease": {
-        "permission": "gitea.pr.comment",
-        "role": "reconciler",
-    },
     "blind_pr_queue_review": {
         "permission": "gitea.pr.review",
         "role": "reviewer",

tests/test_issue_745_moot_lease_reconciler_gate.py

@@ -1,658 +0,0 @@
-"""Reconciler role binding for post-merge moot-lease cleanup (#745).
-
-``gitea_cleanup_post_merge_moot_lease`` (#515) posts a terminal ``phase:
-released`` lease marker but had no capability-map entry and no role gate: entry
-required only ``gitea.read``, apply required only ``gitea.pr.comment``, so any
-profile holding the comment permission reached the mutation while the
-reconciler could not satisfy resolve-exact-task -> mutation.
-
-These tests pin the fixed contract:
-
-- the canonical task and its tool-name alias resolve identically
-  (``gitea.pr.comment`` + ``reconciler``);
-- only a reconciler profile satisfies permission AND role;
-- ``apply=false`` assessment stays reachable under ``gitea.read`` for any role
-  and mutates nothing (documented, deliberate divergence from the apply path);
-- ``apply=true`` requires the exact resolved task, the reconciler role, the
-  comment permission, a validated repository binding and matching dry-run
-  evidence;
-- live/non-moot/superseded/mismatched/malformed leases fail closed;
-- the dry-run ledger is append-only and cleanup stays idempotent.
-
-Every fixture is synthetic. No production PR, lease session or marker is used
-anywhere in this module (see ``TestNoProductionLeaseTouched``).
-"""
-
-import sys as _sys
-from pathlib import Path as _Path
-
-_sys.path.insert(0, str(_Path(__file__).resolve().parent.parent))
-
-import os  # noqa: E402
-import unittest  # noqa: E402
-from datetime import datetime, timezone  # noqa: E402
-from unittest.mock import patch  # noqa: E402
-
-import mcp_server  # noqa: E402
-import post_merge_moot_lease_gate as gate  # noqa: E402
-import reviewer_pr_lease as leases  # noqa: E402
-from mcp_server import gitea_cleanup_post_merge_moot_lease  # noqa: E402
-from role_session_router import RECONCILER_TASKS, TASK_REQUIRED_ROLE  # noqa: E402
-from task_capability_map import required_permission, required_role  # noqa: E402
-
-FAKE_AUTH = "Basic dGVzdDp0ZXN0"
-SLUG = "Scaled-Tech-Consulting/Gitea-Tools"
-PR = 487
-ISSUE = 485
-SESSION = "97274-676d20a825c4"
-HEAD_A = "a" * 40
-HEAD_B = "d" * 40
-LEASE_COMMENT_ID = 6603
-
-CANONICAL_TASK = "cleanup_post_merge_moot_lease"
-TOOL_ALIAS = "gitea_cleanup_post_merge_moot_lease"
-
-_BASE_OPS = "gitea.read,gitea.pr.comment"
-RECONCILER_ENV = {
-    "GITEA_PROFILE_NAME": "prgs-reconciler",
-    "GITEA_ALLOWED_OPERATIONS": _BASE_OPS + ",gitea.pr.close,gitea.branch.delete",
-}
-AUTHOR_ENV = {
-    "GITEA_PROFILE_NAME": "prgs-author",
-    "GITEA_ALLOWED_OPERATIONS": _BASE_OPS + ",gitea.pr.create,gitea.issue.create",
-}
-REVIEWER_ENV = {
-    "GITEA_PROFILE_NAME": "prgs-reviewer",
-    "GITEA_ALLOWED_OPERATIONS": _BASE_OPS + ",gitea.pr.review,gitea.pr.approve",
-}
-MERGER_ENV = {
-    "GITEA_PROFILE_NAME": "prgs-merger",
-    "GITEA_ALLOWED_OPERATIONS": _BASE_OPS + ",gitea.pr.merge",
-}
-
-# Permission shape of the configured role profiles (mirrors
-# tests/test_task_capability_role_invariants.py CANONICAL_ROLE_PROFILES).
-ROLE_PROFILE_PERMISSIONS = {
-    "author": {"gitea.read", "gitea.pr.comment", "gitea.pr.create",
-               "gitea.issue.create", "gitea.issue.comment", "gitea.issue.close",
-               "gitea.branch.create", "gitea.branch.push", "gitea.repo.commit"},
-    "reviewer": {"gitea.read", "gitea.pr.comment", "gitea.pr.review",
-                 "gitea.pr.approve", "gitea.pr.request_changes",
-                 "gitea.issue.comment"},
-    "merger": {"gitea.read", "gitea.pr.comment", "gitea.pr.merge",
-               "gitea.issue.comment"},
-    "reconciler": {"gitea.read", "gitea.pr.comment", "gitea.pr.close",
-                   "gitea.issue.close", "gitea.branch.delete"},
-}
-
-
-def _lease_comment(pr_number=PR, session_id=SESSION, *, phase="claimed",
-                   candidate_head=HEAD_A, comment_id=LEASE_COMMENT_ID):
-    body = leases.format_lease_body(
-        repo=SLUG,
-        pr_number=pr_number,
-        issue_number=ISSUE,
-        reviewer_identity="sysadmin",
-        profile="prgs-reviewer",
-        session_id=session_id,
-        worktree="branches/review-pr487",
-        phase=phase,
-        candidate_head=candidate_head,
-        target_branch="master",
-        target_branch_sha="b" * 40,
-        last_activity=datetime.now(timezone.utc),
-    )
-    return {"id": comment_id, "body": body, "user": {"login": "sysadmin"}}
-
-
-def _api_side_effect(*, pr_state, pr_merged, comments, posted_id=9999):
-    """api_request side effect keyed on method + url; records POSTs."""
-    calls = {"post": []}
-
-    def _side(method, url, auth=None, payload=None, *a, **k):
-        if (method or "").upper() == "POST":
-            calls["post"].append({"url": url, "payload": payload})
-            return {"id": posted_id}
-        if "/comments" in url:
-            return list(comments)
-        if "/pulls/" in url:
-            pr = {"state": pr_state, "number": PR, "merge_commit_sha": "c" * 40}
-            if pr_merged:
-                pr["merged"] = True
-                pr["merged_at"] = "2026-07-08T07:46:04Z"
-            return pr
-        if "/issues/" in url:
-            return {"state": "closed" if pr_merged else "open", "number": ISSUE}
-        return {}
-
-    return _side, calls
-
-
-def _assessment(comments, *, pr_merged=True, pr_state="closed"):
-    return leases.assess_post_merge_moot_lease(
-        comments, pr_number=PR, pr_merged=pr_merged, pr_state=pr_state,
-        merge_commit_sha="c" * 40,
-    )
-
-
-# --------------------------------------------------------------------------- #
-# 1-5. Capability map / router contract
-# --------------------------------------------------------------------------- #
-class TestCleanupTaskContract(unittest.TestCase):
-    def test_canonical_task_is_reconciler_owned(self):
-        """1. The reconciler is the role that can resolve the cleanup task."""
-        self.assertEqual(required_permission(CANONICAL_TASK), "gitea.pr.comment")
-        self.assertEqual(required_role(CANONICAL_TASK), "reconciler")
-
-    def test_tool_alias_resolves_to_identical_contract(self):
-        """5. Alias and canonical task must not diverge."""
-        self.assertEqual(
-            (required_permission(TOOL_ALIAS), required_role(TOOL_ALIAS)),
-            (required_permission(CANONICAL_TASK), required_role(CANONICAL_TASK)),
-        )
-
-    def test_author_reviewer_merger_cannot_resolve_the_task(self):
-        """2-4. No non-reconciler role satisfies permission AND role."""
-        for role in ("author", "reviewer", "merger"):
-            with self.subTest(role=role):
-                self.assertNotEqual(required_role(CANONICAL_TASK), role)
-                # They hold the permission — which is exactly why the role gate
-                # is required rather than optional.
-                self.assertIn(
-                    "gitea.pr.comment", ROLE_PROFILE_PERMISSIONS[role],
-                    "test premise: non-reconciler roles do hold pr.comment",
-                )
-
-    def test_reconciler_profile_satisfies_permission_and_role(self):
-        self.assertIn(
-            required_permission(CANONICAL_TASK),
-            ROLE_PROFILE_PERMISSIONS[required_role(CANONICAL_TASK)],
-        )
-
-    def test_router_agrees_with_capability_map(self):
-        for task in (CANONICAL_TASK, TOOL_ALIAS):
-            with self.subTest(task=task):
-                self.assertIn(task, RECONCILER_TASKS)
-                self.assertEqual(TASK_REQUIRED_ROLE[task], required_role(task))
-
-    def test_unknown_alias_still_rejected(self):
-        for bogus in ("cleanup_post_merge_moot_leases", "cleanup_moot_lease", ""):
-            with self.subTest(task=bogus):
-                with self.assertRaises(KeyError):
-                    required_role(bogus)
-                with self.assertRaises(KeyError):
-                    required_permission(bogus)
-
-
-# --------------------------------------------------------------------------- #
-# Authorization gate unit tests
-# --------------------------------------------------------------------------- #
-class TestApplyAuthorizationGate(unittest.TestCase):
-    def setUp(self):
-        gate._reset_for_testing()
-        self.addCleanup(gate._reset_for_testing)
-        self.assessment = _assessment([_lease_comment()])
-
-    def _evidence(self, **over):
-        base = dict(
-            pr_number=PR, repository_slug=SLUG, lease_moot=True,
-            cleanup_allowed=True, session_id=SESSION, candidate_head=HEAD_A,
-            lease_comment_id=LEASE_COMMENT_ID,
-        )
-        base.update(over)
-        return gate.record_dry_run(**base)
-
-    def _assess(self, **over):
-        kwargs = dict(
-            pr_number=PR, repository_slug=SLUG, resolved_task=CANONICAL_TASK,
-            active_role_kind="reconciler", assessment=self.assessment,
-            evidence=self._evidence(),
-        )
-        kwargs.update(over)
-        return gate.assess_apply_authorization(**kwargs)
-
-    def test_reconciler_with_matching_evidence_is_authorized(self):
-        result = self._assess()
-        self.assertTrue(result["allowed"], result["reasons"])
-        self.assertTrue(result["evidence_matched"])
-
-    def test_apply_without_exact_task_resolution_fails(self):
-        """8. Apply without exact task resolution fails preflight."""
-        result = self._assess(resolved_task=None)
-        self.assertFalse(result["allowed"])
-        self.assertEqual(result["blocker_kind"], "unresolved_cleanup_task")
-
-    def test_resolving_another_task_does_not_authorize_cleanup(self):
-        """9. A sibling reconciler task is not a substitute."""
-        for other in ("delete_branch", "reconcile_already_landed_pr",
-                      "cleanup_merged_pr_branch", "comment_pr"):
-            with self.subTest(task=other):
-                result = self._assess(resolved_task=other)
-                self.assertFalse(result["allowed"])
-                self.assertEqual(
-                    result["blocker_kind"], "unresolved_cleanup_task")
-
-    def test_non_reconciler_roles_fail_closed(self):
-        """10. Author/reviewer/merger cannot apply despite pr.comment."""
-        for role in ("author", "reviewer", "merger", None, ""):
-            with self.subTest(role=role):
-                result = self._assess(active_role_kind=role)
-                self.assertFalse(result["allowed"])
-                self.assertEqual(result["blocker_kind"], "wrong_role")
-
-    def test_missing_repository_identity_fails_closed(self):
-        result = self._assess(repository_slug=None)
-        self.assertFalse(result["allowed"])
-        self.assertEqual(result["blocker_kind"], "repository_binding")
-
-    def test_missing_dry_run_evidence_fails_closed(self):
-        result = self._assess(evidence=None)
-        self.assertFalse(result["allowed"])
-        self.assertEqual(result["blocker_kind"], "missing_dry_run_evidence")
-
-    def test_dry_run_that_disallowed_cleanup_fails_closed(self):
-        result = self._assess(evidence=self._evidence(cleanup_allowed=False))
-        self.assertFalse(result["allowed"])
-        self.assertEqual(result["blocker_kind"], "dry_run_not_allowed")
-
-    def test_dry_run_for_another_pr_or_repo_fails_closed(self):
-        """11. Wrong PR / repository fails closed."""
-        for over in ({"pr_number": PR + 1}, {"repository_slug": "Other/Repo"}):
-            with self.subTest(**over):
-                result = self._assess(evidence=self._evidence(**over))
-                self.assertFalse(result["allowed"])
-                self.assertEqual(result["blocker_kind"], "dry_run_mismatch")
-
-    def test_superseded_lease_fails_closed(self):
-        """12. Head/session/marker drift since the dry run fails closed."""
-        for over in ({"session_id": "other-session"},
-                     {"candidate_head": HEAD_B},
-                     {"lease_comment_id": 7777}):
-            with self.subTest(**over):
-                result = self._assess(evidence=self._evidence(**over))
-                self.assertFalse(result["allowed"])
-                self.assertEqual(result["blocker_kind"], "superseded_lease")
-
-    def test_expectation_mismatch_fails_closed(self):
-        """11. Wrong session / head / marker expectations fail closed."""
-        for over in ({"expected_session_id": "nope"},
-                     {"expected_candidate_head": HEAD_B},
-                     {"expected_lease_comment_id": 7777}):
-            with self.subTest(**over):
-                result = self._assess(**over)
-                self.assertFalse(result["allowed"])
-                self.assertEqual(result["blocker_kind"], "lease_mismatch")
-
-    def test_matching_expectations_are_authorized(self):
-        result = self._assess(
-            expected_session_id=SESSION,
-            expected_candidate_head=HEAD_A,
-            expected_lease_comment_id=LEASE_COMMENT_ID,
-        )
-        self.assertTrue(result["allowed"], result["reasons"])
-
-    def test_non_moot_lease_fails_closed(self):
-        """12. A live lease on an open PR is never cleanable."""
-        open_pr = _assessment(
-            [_lease_comment()], pr_merged=False, pr_state="open")
-        result = self._assess(assessment=open_pr)
-        self.assertFalse(result["allowed"])
-        self.assertIn(
-            result["blocker_kind"], ("lease_not_moot", "malformed_lease"))
-
-    def test_malformed_lease_fails_closed(self):
-        malformed = dict(self.assessment)
-        malformed["active_lease"] = {
-            "session_id": "", "candidate_head": None, "comment_id": None}
-        result = self._assess(assessment=malformed)
-        self.assertFalse(result["allowed"])
-        self.assertEqual(result["blocker_kind"], "malformed_lease")
-
-    def test_ledger_is_append_only(self):
-        """14. Recording never rewrites or drops prior entries."""
-        first = self._evidence()
-        second = self._evidence(candidate_head=HEAD_B, lease_comment_id=7777)
-        history = gate.dry_run_history()
-        self.assertEqual(len(history), 2)
-        self.assertEqual(history[0]["candidate_head"], first["candidate_head"])
-        self.assertEqual(history[1]["candidate_head"], HEAD_B)
-        # Newest-wins for lookup, but the older entry survives in history.
-        latest = gate.latest_dry_run(pr_number=PR, repository_slug=SLUG)
-        self.assertEqual(latest["lease_comment_id"], second["lease_comment_id"])
-        self.assertEqual(gate.dry_run_history()[0]["lease_comment_id"],
-                         LEASE_COMMENT_ID)
-
-    def test_history_view_cannot_mutate_the_ledger(self):
-        self._evidence()
-        snapshot = gate.dry_run_history()
-        snapshot[0]["pr_number"] = 999999
-        self.assertEqual(
-            gate.dry_run_history()[0]["pr_number"], PR,
-            "dry_run_history must hand out copies, not live rows")
-
-
-# --------------------------------------------------------------------------- #
-# Tool-level behavior
-# --------------------------------------------------------------------------- #
-class _ToolCase(unittest.TestCase):
-    def setUp(self):
-        leases.clear_session_lease()
-        gate._reset_for_testing()
-        self.addCleanup(gate._reset_for_testing)
-
-    def _run(self, env, *, apply, comments, pr_state="closed", pr_merged=True,
-             resolved_task=CANONICAL_TASK, slug=SLUG, **kwargs):
-        side, calls = _api_side_effect(
-            pr_state=pr_state, pr_merged=pr_merged, comments=comments)
-        with patch("mcp_server.api_request", side_effect=side), \
-                patch("mcp_server.get_auth_header", return_value=FAKE_AUTH), \
-                patch("mcp_server.verify_preflight_purity", return_value=None), \
-                patch("mcp_server._bound_repository_slug", return_value=slug), \
-                patch("mcp_server._repository_binding_block", return_value=None), \
-                patch.object(mcp_server, "_preflight_resolved_task",
-                             resolved_task), \
-                patch.dict(os.environ, env, clear=True):
-            result = gitea_cleanup_post_merge_moot_lease(
-                pr_number=PR, apply=apply, remote="prgs", **kwargs)
-        return result, calls
-
-
-class TestDryRunOpenToEveryRole(_ToolCase):
-    """7. Dry run performs no mutation and stays under the read capability."""
-
-    def test_dry_run_reports_moot_and_mutates_nothing_for_every_role(self):
-        for name, env in (("reconciler", RECONCILER_ENV), ("author", AUTHOR_ENV),
-                          ("reviewer", REVIEWER_ENV), ("merger", MERGER_ENV)):
-            with self.subTest(role=name):
-                gate._reset_for_testing()
-                result, calls = self._run(
-                    env, apply=False, comments=[_lease_comment()],
-                    resolved_task=None)
-                self.assertTrue(result["success"])
-                self.assertTrue(result["lease_moot"])
-                self.assertTrue(result["cleanup_allowed"])
-                self.assertFalse(result["cleanup_performed"])
-                self.assertEqual(result["mode"], "read_only")
-                self.assertEqual(calls["post"], [], "dry run must not mutate")
-
-    def test_dry_run_records_evidence(self):
-        result, _ = self._run(
-            RECONCILER_ENV, apply=False, comments=[_lease_comment()])
-        evidence = result["dry_run_evidence"]
-        self.assertEqual(evidence["pr_number"], PR)
-        self.assertEqual(evidence["repository_slug"], SLUG)
-        self.assertEqual(evidence["session_id"], SESSION)
-        self.assertEqual(evidence["candidate_head"], HEAD_A)
-        self.assertEqual(evidence["lease_comment_id"], LEASE_COMMENT_ID)
-        self.assertTrue(evidence["lease_moot"])
-        self.assertTrue(evidence["cleanup_allowed"])
-
-
-class TestApplyRequiresReconciler(_ToolCase):
-    def test_reconciler_apply_succeeds_after_matching_dry_run(self):
-        """7 (apply). Allowed dry run then apply posts exactly one marker."""
-        comments = [_lease_comment()]
-        dry, dry_calls = self._run(
-            RECONCILER_ENV, apply=False, comments=comments)
-        self.assertTrue(dry["cleanup_allowed"])
-        self.assertEqual(dry_calls["post"], [])
-
-        result, calls = self._run(
-            RECONCILER_ENV, apply=True, comments=comments)
-        self.assertTrue(result["success"], result.get("reasons"))
-        self.assertTrue(result["cleanup_performed"])
-        self.assertEqual(result["released_comment_id"], 9999)
-        self.assertEqual(len(calls["post"]), 1)
-        body = calls["post"][0]["payload"]["body"]
-        self.assertIn("phase: released", body)
-        self.assertIn("post-merge-moot", body)
-
-    def test_apply_without_dry_run_fails_closed(self):
-        """6. Apply must be preceded by a matching dry run."""
-        result, calls = self._run(
-            RECONCILER_ENV, apply=True, comments=[_lease_comment()])
-        self.assertFalse(result["success"])
-        self.assertFalse(result["cleanup_performed"])
-        self.assertEqual(result["blocker_kind"], "missing_dry_run_evidence")
-        self.assertEqual(calls["post"], [])
-
-    def test_apply_without_exact_task_resolution_fails_closed(self):
-        """8. No resolved cleanup task -> no mutation."""
-        comments = [_lease_comment()]
-        self._run(RECONCILER_ENV, apply=False, comments=comments)
-        result, calls = self._run(
-            RECONCILER_ENV, apply=True, comments=comments, resolved_task=None)
-        self.assertFalse(result["success"])
-        self.assertEqual(result["blocker_kind"], "unresolved_cleanup_task")
-        self.assertEqual(calls["post"], [])
-
-    def test_resolving_a_different_task_does_not_authorize_apply(self):
-        """9. Another resolved task is not a substitute."""
-        comments = [_lease_comment()]
-        self._run(RECONCILER_ENV, apply=False, comments=comments)
-        result, calls = self._run(
-            RECONCILER_ENV, apply=True, comments=comments,
-            resolved_task="delete_branch")
-        self.assertFalse(result["success"])
-        self.assertEqual(result["blocker_kind"], "unresolved_cleanup_task")
-        self.assertEqual(calls["post"], [])
-
-    def test_author_reviewer_merger_cannot_apply(self):
-        """10. Permission-only roles are refused at the role gate."""
-        for name, env in (("author", AUTHOR_ENV), ("reviewer", REVIEWER_ENV),
-                          ("merger", MERGER_ENV)):
-            with self.subTest(role=name):
-                gate._reset_for_testing()
-                comments = [_lease_comment()]
-                self._run(env, apply=False, comments=comments)
-                result, calls = self._run(env, apply=True, comments=comments)
-                self.assertFalse(result["success"])
-                self.assertFalse(result["cleanup_performed"])
-                self.assertEqual(result["blocker_kind"], "wrong_role")
-                self.assertEqual(
-                    calls["post"], [],
-                    f"{name} must not post a terminal lease marker")
-
-
-class TestApplyFailsClosedOnLeaseState(_ToolCase):
-    def test_open_pr_lease_is_never_force_cleaned(self):
-        """12. Non-moot: an active lease on an open PR stays untouched."""
-        comments = [_lease_comment()]
-        result, calls = self._run(
-            RECONCILER_ENV, apply=True, comments=comments,
-            pr_state="open", pr_merged=False)
-        self.assertFalse(result["cleanup_performed"])
-        self.assertFalse(result["pr_merged_or_closed"])
-        self.assertEqual(calls["post"], [], "never force-clean an open PR lease")
-        self.assertTrue(any(
-            "still open" in r for r in result.get("cleanup_skipped_reason", [])))
-
-    def test_superseded_lease_between_dry_run_and_apply_fails_closed(self):
-        """12. The lease moved on after the dry run -> refuse."""
-        self._run(RECONCILER_ENV, apply=False, comments=[_lease_comment()])
-        moved = [_lease_comment(session_id="fresh-session",
-                                candidate_head=HEAD_B, comment_id=7777)]
-        result, calls = self._run(RECONCILER_ENV, apply=True, comments=moved)
-        self.assertFalse(result["success"])
-        self.assertEqual(result["blocker_kind"], "superseded_lease")
-        self.assertEqual(calls["post"], [])
-
-    def test_expectation_mismatch_fails_closed(self):
-        """11. Wrong session / head / marker expectations refuse the apply."""
-        comments = [_lease_comment()]
-        for kwargs in ({"expected_session_id": "wrong-session"},
-                       {"expected_candidate_head": HEAD_B},
-                       {"expected_lease_comment_id": 7777}):
-            with self.subTest(**kwargs):
-                gate._reset_for_testing()
-                self._run(RECONCILER_ENV, apply=False, comments=comments)
-                result, calls = self._run(
-                    RECONCILER_ENV, apply=True, comments=comments, **kwargs)
-                self.assertFalse(result["success"])
-                self.assertEqual(result["blocker_kind"], "lease_mismatch")
-                self.assertEqual(calls["post"], [])
-
-    def test_already_terminal_cleanup_is_idempotent(self):
-        """13. A released lease reports nothing to clean and posts nothing."""
-        first = _assessment([_lease_comment()])
-        released = {"id": 7000, "body": first["release_body"],
-                    "user": {"login": "sysadmin"}}
-        comments = [_lease_comment(), released]
-        self._run(RECONCILER_ENV, apply=False, comments=comments)
-        result, calls = self._run(RECONCILER_ENV, apply=True, comments=comments)
-        self.assertFalse(result["cleanup_performed"])
-        self.assertFalse(result["lease_moot"])
-        self.assertEqual(calls["post"], [], "no second terminal marker")
-        self.assertTrue(any(
-            "already released/terminal" in r
-            for r in result.get("cleanup_skipped_reason", [])))
-
-    def test_apply_is_append_only_never_deletes(self):
-        """14. The only write is a POST; nothing is edited or deleted."""
-        comments = [_lease_comment()]
-        self._run(RECONCILER_ENV, apply=False, comments=comments)
-        side, _calls = _api_side_effect(
-            pr_state="closed", pr_merged=True, comments=comments)
-        seen = []
-
-        def _recording(method, url, auth=None, payload=None, *a, **k):
-            seen.append((method or "").upper())
-            return side(method, url, auth, payload, *a, **k)
-
-        with patch("mcp_server.api_request", side_effect=_recording), \
-                patch("mcp_server.get_auth_header", return_value=FAKE_AUTH), \
-                patch("mcp_server.verify_preflight_purity", return_value=None), \
-                patch("mcp_server._bound_repository_slug", return_value=SLUG), \
-                patch("mcp_server._repository_binding_block", return_value=None), \
-                patch.object(mcp_server, "_preflight_resolved_task",
-                             CANONICAL_TASK), \
-                patch.dict(os.environ, RECONCILER_ENV, clear=True):
-            result = gitea_cleanup_post_merge_moot_lease(
-                pr_number=PR, apply=True, remote="prgs")
-        self.assertTrue(result["cleanup_performed"])
-        self.assertNotIn("DELETE", seen)
-        self.assertNotIn("PATCH", seen)
-        self.assertNotIn("PUT", seen)
-        self.assertEqual(seen.count("POST"), 1)
-
-
-class TestRepositoryBinding(_ToolCase):
-    """11. Foreign-repository targets fail closed before any mutation."""
-
-    def test_explicit_foreign_repository_is_rejected(self):
-        comments = [_lease_comment()]
-        side, calls = _api_side_effect(
-            pr_state="closed", pr_merged=True, comments=comments)
-        with patch("mcp_server.api_request", side_effect=side), \
-                patch("mcp_server.get_auth_header", return_value=FAKE_AUTH), \
-                patch("mcp_server.verify_preflight_purity", return_value=None), \
-                patch("mcp_server._canonical_repository_slug",
-                      return_value=(None, [])), \
-                patch("mcp_server._workspace_repository_slug",
-                      return_value=SLUG), \
-                patch.object(mcp_server, "_preflight_resolved_task",
-                             CANONICAL_TASK), \
-                patch.dict(os.environ, RECONCILER_ENV, clear=True):
-            gitea_cleanup_post_merge_moot_lease(
-                pr_number=PR, apply=False, remote="prgs")
-            result = gitea_cleanup_post_merge_moot_lease(
-                pr_number=PR, apply=True, remote="prgs",
-                org="Some-Other-Org", repo="Some-Other-Repo")
-        self.assertFalse(result["success"])
-        self.assertEqual(result["blocker_kind"], "repository_binding")
-        self.assertEqual(calls["post"], [])
-
-    def test_unresolvable_canonical_root_fails_closed(self):
-        comments = [_lease_comment()]
-        side, calls = _api_side_effect(
-            pr_state="closed", pr_merged=True, comments=comments)
-        with patch("mcp_server.api_request", side_effect=side), \
-                patch("mcp_server.get_auth_header", return_value=FAKE_AUTH), \
-                patch("mcp_server.verify_preflight_purity", return_value=None), \
-                patch("mcp_server._canonical_repository_slug",
-                      return_value=(None, ["canonical root unresolvable"])), \
-                patch.object(mcp_server, "_preflight_resolved_task",
-                             CANONICAL_TASK), \
-                patch.dict(os.environ, RECONCILER_ENV, clear=True):
-            gitea_cleanup_post_merge_moot_lease(
-                pr_number=PR, apply=False, remote="prgs")
-            result = gitea_cleanup_post_merge_moot_lease(
-                pr_number=PR, apply=True, remote="prgs")
-        self.assertFalse(result["success"])
-        self.assertEqual(result["blocker_kind"], "repository_binding")
-        self.assertEqual(calls["post"], [])
-
-
-class TestPreflightBinding(_ToolCase):
-    """4. The apply path binds the shared preflight to the exact task."""
-
-    def test_apply_forwards_task_and_target_to_preflight(self):
-        comments = [_lease_comment()]
-        self._run(RECONCILER_ENV, apply=False, comments=comments)
-        side, _calls = _api_side_effect(
-            pr_state="closed", pr_merged=True, comments=comments)
-        seen = {}
-
-        def _purity(remote=None, worktree_path=None, task=None, **kw):
-            seen.update({"remote": remote, "worktree_path": worktree_path,
-                         "task": task, **kw})
-            return None
-
-        with patch("mcp_server.api_request", side_effect=side), \
-                patch("mcp_server.get_auth_header", return_value=FAKE_AUTH), \
-                patch("mcp_server.verify_preflight_purity", _purity), \
-                patch("mcp_server._bound_repository_slug", return_value=SLUG), \
-                patch("mcp_server._repository_binding_block", return_value=None), \
-                patch.object(mcp_server, "_preflight_resolved_task",
-                             CANONICAL_TASK), \
-                patch.dict(os.environ, RECONCILER_ENV, clear=True):
-            result = gitea_cleanup_post_merge_moot_lease(
-                pr_number=PR, apply=True, remote="prgs",
-                worktree_path="/tmp/branches/reconciler-745",
-                org="Scaled-Tech-Consulting", repo="Gitea-Tools")
-        self.assertTrue(result["cleanup_performed"])
-        self.assertEqual(seen["task"], CANONICAL_TASK)
-        self.assertEqual(seen["worktree_path"], "/tmp/branches/reconciler-745")
-        self.assertEqual(seen["org"], "Scaled-Tech-Consulting")
-        self.assertEqual(seen["repo"], "Gitea-Tools")
-
-    def test_dry_run_does_not_require_preflight(self):
-        def _boom(*a, **k):
-            raise AssertionError("dry run must not run mutation preflight")
-
-        side, calls = _api_side_effect(
-            pr_state="closed", pr_merged=True, comments=[_lease_comment()])
-        with patch("mcp_server.api_request", side_effect=side), \
-                patch("mcp_server.get_auth_header", return_value=FAKE_AUTH), \
-                patch("mcp_server.verify_preflight_purity", _boom), \
-                patch("mcp_server._bound_repository_slug", return_value=SLUG), \
-                patch.dict(os.environ, AUTHOR_ENV, clear=True):
-            result = gitea_cleanup_post_merge_moot_lease(
-                pr_number=PR, apply=False, remote="prgs")
-        self.assertTrue(result["success"])
-        self.assertEqual(calls["post"], [])
-
-
-class TestNoProductionLeaseTouched(unittest.TestCase):
-    """16. No real production lease or PR is referenced by these tests."""
-
-    PRODUCTION_PR = 744
-    PRODUCTION_SESSION = "33673-1d54887a0415"
-    PRODUCTION_MARKER = 12452
-
-    def test_fixtures_are_synthetic(self):
-        self.assertNotEqual(PR, self.PRODUCTION_PR)
-        self.assertNotEqual(SESSION, self.PRODUCTION_SESSION)
-        self.assertNotEqual(LEASE_COMMENT_ID, self.PRODUCTION_MARKER)
-
-    def test_module_source_never_names_the_production_lease(self):
-        source = _Path(__file__).read_text()
-        for token in (self.PRODUCTION_SESSION, str(self.PRODUCTION_MARKER)):
-            self.assertEqual(
-                source.count(token), 1,
-                f"{token!r} must appear only in this guard's own constants",
-            )
-
-
-if __name__ == "__main__":
-    unittest.main()

tests/test_issue_747_sliding_lease_ttl.py

@@ -0,0 +1,210 @@
+"""Tests for the 10-minute sliding TTL on reviewer and merger PR leases (#747).
+
+The lease ledger previously minted a fixed 120-minute expiry and derived
+staleness from separate 30/60-minute activity bands. A dead session therefore
+held a PR for up to two hours. These tests pin the sliding-window contract:
+acquisition mints a 10-minute expiry, every heartbeat slides it forward, and an
+expired lease is immediately reclaimable with no intermediate waiting tier.
+"""
+
+import sys
+import unittest
+from datetime import datetime, timedelta, timezone
+
+sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
+
+import reviewer_pr_lease as leases
+
+
+def _body(
+    *,
+    session_id: str = "session-a",
+    pr_number: int = 747,
+    phase: str = "claimed",
+    last_activity: datetime | None = None,
+    expires_at: datetime | None = None,
+    ttl_minutes: int | None = None,
+) -> str:
+    kwargs = {}
+    if ttl_minutes is not None:
+        kwargs["ttl_minutes"] = ttl_minutes
+    return leases.format_lease_body(
+        repo="Scaled-Tech-Consulting/Gitea-Tools",
+        pr_number=pr_number,
+        issue_number=747,
+        reviewer_identity="rev1",
+        profile="prgs-reviewer",
+        session_id=session_id,
+        worktree="branches/review-pr747",
+        phase=phase,
+        candidate_head="a" * 40,
+        target_branch="master",
+        target_branch_sha="b" * 40,
+        last_activity=last_activity,
+        expires_at=expires_at,
+        **kwargs,
+    )
+
+
+def _comment(**kwargs) -> dict:
+    return {"id": 1, "body": _body(**kwargs), "user": {"login": "rev1"}}
+
+
+def _minutes_ago(minutes: int) -> datetime:
+    return datetime.now(timezone.utc) - timedelta(minutes=minutes)
+
+
+class TestSlidingTTLConstant(unittest.TestCase):
+    """AC6: one named constant per lease kind, no duplicated literals."""
+
+    def test_ttl_is_ten_minutes(self):
+        self.assertEqual(leases.LEASE_TTL_MINUTES, 10)
+
+    def test_renewal_window_is_separately_named(self):
+        self.assertEqual(leases.LEASE_RENEWAL_MINUTES, 10)
+
+
+class TestAcquisitionTTL(unittest.TestCase):
+    """AC1 / AC2: reviewer and merger acquisition both mint now + 10 minutes."""
+
+    def test_acquire_mints_ten_minute_expiry(self):
+        now = datetime(2026, 7, 18, 12, 0, 0, tzinfo=timezone.utc)
+        lease = leases.parse_lease_comment(_body(last_activity=now))
+        expires = leases._parse_timestamp(lease["expires_at"])
+        self.assertEqual(expires, now + timedelta(minutes=10))
+
+    def test_merger_acquisition_shares_the_same_window(self):
+        # Merger acquisition funnels through the same lease-body formatter, so
+        # the reviewer TTL is the merger TTL by construction.
+        now = datetime(2026, 7, 18, 12, 0, 0, tzinfo=timezone.utc)
+        lease = leases.parse_lease_comment(_body(phase="merging", last_activity=now))
+        expires = leases._parse_timestamp(lease["expires_at"])
+        self.assertEqual(expires, now + timedelta(minutes=10))
+
+
+class TestHeartbeatSlides(unittest.TestCase):
+    """AC3: a heartbeat slides expires_at to now + 10 minutes."""
+
+    def test_heartbeat_slides_expiry_forward(self):
+        acquired = datetime(2026, 7, 18, 12, 0, 0, tzinfo=timezone.utc)
+        beat = acquired + timedelta(minutes=7)
+        first = leases.parse_lease_comment(_body(last_activity=acquired))
+        renewed = leases.parse_lease_comment(_body(last_activity=beat))
+
+        first_expiry = leases._parse_timestamp(first["expires_at"])
+        renewed_expiry = leases._parse_timestamp(renewed["expires_at"])
+
+        self.assertEqual(renewed_expiry, beat + timedelta(minutes=10))
+        self.assertGreater(renewed_expiry, first_expiry)
+
+    def test_renewal_window_is_independently_tunable(self):
+        # The renewal amount must not be hardwired to the acquisition TTL;
+        # format_lease_body accepts an explicit window.
+        now = datetime(2026, 7, 18, 12, 0, 0, tzinfo=timezone.utc)
+        lease = leases.parse_lease_comment(_body(last_activity=now, ttl_minutes=3))
+        expires = leases._parse_timestamp(lease["expires_at"])
+        self.assertEqual(expires, now + timedelta(minutes=3))
+
+
+class TestFreshnessBands(unittest.TestCase):
+    """AC5: expiry is the only gate; no intermediate reclaim tier."""
+
+    def test_fresh_lease_is_active(self):
+        lease = leases.parse_lease_comment(_body(last_activity=_minutes_ago(1)))
+        self.assertEqual(leases.classify_lease_freshness(lease), "active")
+
+    def test_idle_past_half_ttl_warns_before_expiry(self):
+        lease = leases.parse_lease_comment(_body(last_activity=_minutes_ago(6)))
+        self.assertEqual(leases.classify_lease_freshness(lease), "stale_warning")
+
+    def test_lease_expires_after_ten_idle_minutes(self):
+        lease = leases.parse_lease_comment(_body(last_activity=_minutes_ago(11)))
+        self.assertEqual(leases.classify_lease_freshness(lease), "expired")
+
+    def test_no_separate_reclaimable_tier_remains(self):
+        # The old 60-minute reclaim band sat between "stale" and "expired" and
+        # blocked acquisition. Under a sliding TTL an idle lease is already
+        # expired, so the tier must not reappear at any idle duration.
+        for minutes in (11, 30, 65, 121, 600):
+            lease = leases.parse_lease_comment(_body(last_activity=_minutes_ago(minutes)))
+            self.assertEqual(
+                leases.classify_lease_freshness(lease),
+                "expired",
+                f"idle {minutes}m should be expired, not a waiting tier",
+            )
+
+
+class TestExpiredLeaseIsImmediatelyReclaimable(unittest.TestCase):
+    """AC5: another session takes over an expired lease with no extra wait."""
+
+    def setUp(self):
+        leases.clear_session_lease()
+
+    def _acquire_against(self, comments: list[dict]) -> dict:
+        return leases.assess_acquire_lease(
+            comments,
+            pr_number=747,
+            reviewer_identity="rev2",
+            profile="prgs-reviewer",
+            session_id="session-b",
+            repo="Scaled-Tech-Consulting/Gitea-Tools",
+            issue_number=747,
+            worktree="branches/review-pr747-b",
+            candidate_head="c" * 40,
+            target_branch="master",
+            target_branch_sha="d" * 40,
+        )
+
+    def test_expired_foreign_lease_does_not_block_acquisition(self):
+        comments = [_comment(session_id="dead-session", last_activity=_minutes_ago(11))]
+        result = self._acquire_against(comments)
+        self.assertTrue(result["acquire_allowed"], result["reasons"])
+
+    def test_live_foreign_lease_still_blocks_acquisition(self):
+        comments = [_comment(session_id="live-session", last_activity=_minutes_ago(2))]
+        result = self._acquire_against(comments)
+        self.assertFalse(result["acquire_allowed"])
+        self.assertTrue(any("already has active" in r for r in result["reasons"]))
+
+
+class TestRemainingTimeReporting(unittest.TestCase):
+    """AC7: diagnostics can distinguish 'held and live' from 'held and dying'."""
+
+    def test_seconds_remaining_on_live_lease(self):
+        now = datetime(2026, 7, 18, 12, 0, 0, tzinfo=timezone.utc)
+        lease = leases.parse_lease_comment(_body(last_activity=now))
+        remaining = leases.lease_seconds_remaining(lease, now=now + timedelta(minutes=4))
+        self.assertEqual(remaining, 360)
+
+    def test_seconds_remaining_is_zero_when_expired(self):
+        now = datetime(2026, 7, 18, 12, 0, 0, tzinfo=timezone.utc)
+        lease = leases.parse_lease_comment(_body(last_activity=now))
+        remaining = leases.lease_seconds_remaining(lease, now=now + timedelta(minutes=30))
+        self.assertEqual(remaining, 0)
+
+    def test_seconds_remaining_is_none_without_parsable_expiry(self):
+        self.assertIsNone(leases.lease_seconds_remaining({"expires_at": "not-a-time"}))
+
+
+class TestLegacyLeaseRows(unittest.TestCase):
+    """AC9: leases minted under the old 120-minute TTL still evaluate."""
+
+    def test_legacy_two_hour_expiry_is_honoured_until_it_passes(self):
+        now = datetime.now(timezone.utc)
+        legacy = leases.parse_lease_comment(
+            _body(last_activity=now - timedelta(minutes=90), expires_at=now + timedelta(minutes=30))
+        )
+        # Still inside its originally minted window: not expired, but idle long
+        # enough to warn.
+        self.assertEqual(leases.classify_lease_freshness(legacy), "stale_warning")
+
+    def test_legacy_row_past_its_own_expiry_is_expired(self):
+        now = datetime.now(timezone.utc)
+        legacy = leases.parse_lease_comment(
+            _body(last_activity=now - timedelta(minutes=180), expires_at=now - timedelta(minutes=60))
+        )
+        self.assertEqual(leases.classify_lease_freshness(legacy), "expired")
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_post_merge_moot_lease.py

@@ -19,8 +19,6 @@ from unittest.mock import patch
 
 sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
 
-import mcp_server  # noqa: E402
-import post_merge_moot_lease_gate as moot_gate  # noqa: E402
 import reviewer_pr_lease as leases  # noqa: E402
 from mcp_server import (  # noqa: E402
     gitea_acquire_reviewer_pr_lease,
@@ -32,13 +30,6 @@ MERGER_ENV = {
     "GITEA_PROFILE_NAME": "prgs-merger",
     "GITEA_ALLOWED_OPERATIONS": "gitea.read,gitea.pr.comment",
 }
-# #745: applying the terminal marker is reconciler-owned.
-RECONCILER_ENV = {
-    "GITEA_PROFILE_NAME": "prgs-reconciler",
-    "GITEA_ALLOWED_OPERATIONS": "gitea.read,gitea.pr.comment,gitea.pr.close",
-}
-CLEANUP_TASK = moot_gate.CLEANUP_TASK
-REPO_SLUG = "Scaled-Tech-Consulting/Gitea-Tools"
 PR = 487
 ISSUE = 485
 SESSION = "97274-676d20a825c4"
@@ -213,8 +204,6 @@ class TestAcquireToolRefusesMergedPR(unittest.TestCase):
 class TestCleanupTool(unittest.TestCase):
     def setUp(self):
         leases.clear_session_lease()
-        moot_gate._reset_for_testing()
-        self.addCleanup(moot_gate._reset_for_testing)
 
     @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
     @patch("mcp_server.api_request")
@@ -234,53 +223,23 @@ class TestCleanupTool(unittest.TestCase):
         self.assertEqual(calls["post"], [])
 
     @patch("mcp_server.verify_preflight_purity", return_value=None)
-    @patch("mcp_server._repository_binding_block", return_value=None)
-    @patch("mcp_server._bound_repository_slug", return_value=REPO_SLUG)
     @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
     @patch("mcp_server.api_request")
     def test_apply_posts_released_marker_on_merged_pr(
-            self, mock_api, _auth, _slug, _binding, _purity):
-        """#745: apply is reconciler-only and needs a matching dry run first."""
+            self, mock_api, _auth, _purity):
         side, calls = _api_side_effect(
             pr_state="closed", pr_merged=True, comments=[_lease_comment()])
         mock_api.side_effect = side
-        with patch.object(mcp_server, "_preflight_resolved_task",
-                          CLEANUP_TASK), \
-                patch.dict(os.environ, RECONCILER_ENV, clear=True):
-            gitea_cleanup_post_merge_moot_lease(
-                pr_number=PR, apply=False, remote="prgs")
+        with patch.dict(os.environ, MERGER_ENV, clear=True):
             result = gitea_cleanup_post_merge_moot_lease(
                 pr_number=PR, apply=True, remote="prgs")
-        self.assertTrue(result["success"], result.get("reasons"))
+        self.assertTrue(result["success"])
         self.assertTrue(result["cleanup_performed"])
         self.assertEqual(result["released_comment_id"], 9999)
         self.assertEqual(len(calls["post"]), 1)
         self.assertIn("phase: released", calls["post"][0]["payload"]["body"])
         self.assertIn("post-merge-moot", calls["post"][0]["payload"]["body"])
 
-    @patch("mcp_server.verify_preflight_purity", return_value=None)
-    @patch("mcp_server._repository_binding_block", return_value=None)
-    @patch("mcp_server._bound_repository_slug", return_value=REPO_SLUG)
-    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
-    @patch("mcp_server.api_request")
-    def test_merger_can_no_longer_apply(
-            self, mock_api, _auth, _slug, _binding, _purity):
-        """#745: holding gitea.pr.comment is no longer sufficient to apply."""
-        side, calls = _api_side_effect(
-            pr_state="closed", pr_merged=True, comments=[_lease_comment()])
-        mock_api.side_effect = side
-        with patch.object(mcp_server, "_preflight_resolved_task",
-                          CLEANUP_TASK), \
-                patch.dict(os.environ, MERGER_ENV, clear=True):
-            gitea_cleanup_post_merge_moot_lease(
-                pr_number=PR, apply=False, remote="prgs")
-            result = gitea_cleanup_post_merge_moot_lease(
-                pr_number=PR, apply=True, remote="prgs")
-        self.assertFalse(result["success"])
-        self.assertFalse(result["cleanup_performed"])
-        self.assertEqual(result["blocker_kind"], "wrong_role")
-        self.assertEqual(calls["post"], [])
-
     @patch("mcp_server.verify_preflight_purity", return_value=None)
     @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
     @patch("mcp_server.api_request")

tests/test_reviewer_pr_lease.py

@@ -79,22 +79,26 @@ class TestReviewerLeaseAcquire(unittest.TestCase):
 
 
 class TestReviewerLeaseFreshness(unittest.TestCase):
-    def test_stale_warning_after_30_minutes(self):
+    def test_stale_warning_at_half_the_sliding_window(self):
+        # #747 warns at half the 10-minute window, while the owner can still
+        # heartbeat and keep the lease.
         lease = leases.parse_lease_comment(
-            _lease_comment(382, "session-a", minutes_ago=35)["body"]
+            _lease_comment(382, "session-a", minutes_ago=6)["body"]
         )
         self.assertEqual(
             leases.classify_lease_freshness(lease),
             "stale_warning",
         )
 
-    def test_reclaimable_after_60_minutes(self):
+    def test_expired_once_the_sliding_window_lapses(self):
+        # Pre-#747 a 65-minute-idle lease was "reclaimable" and had to wait out
+        # a second timer. It is now simply expired and immediately takeable.
         lease = leases.parse_lease_comment(
             _lease_comment(382, "session-a", minutes_ago=65)["body"]
         )
         self.assertEqual(
             leases.classify_lease_freshness(lease),
-            "reclaimable",
+            "expired",
         )
 
 
@@ -281,7 +285,10 @@ class TestReviewerLeaseHandoffDiagnose(unittest.TestCase):
         self.assertEqual(result["active_lease"]["comment_id"], 8647)
         self.assertFalse(result["mutation_allowed"])
 
-    def test_foreign_reclaimable_release_expired(self):
+    def test_foreign_expired_release_expired(self):
+        # Pre-#747 this classified as "foreign_reclaimable" after the 60-minute
+        # activity band. Under the sliding TTL the lease is simply expired, and
+        # the sanctioned next action is unchanged.
         reclaim = _lease_comment(
             592, "foreign-old", phase="claimed", minutes_ago=65
         )
@@ -293,7 +300,7 @@ class TestReviewerLeaseHandoffDiagnose(unittest.TestCase):
             current_reviewer_identity="sysadmin",
             proposed_worktree="branches/review-pr-592",
         )
-        self.assertEqual(result["classification"], "foreign_reclaimable")
+        self.assertEqual(result["classification"], "foreign_expired")
         self.assertEqual(
             result["next_action"], leases.NEXT_ACTION_RELEASE_EXPIRED_LEASE
         )