feat(lease): make the author task heartbeat load-bearing (#790 Slice A)

Slice A of Issue #790, per the controller reassessment in comment 13958. Does
not close the issue: terminal retirement (Slice B) and the read-side generation
check plus the #760 renewal re-scope (Slice C) are deliberately not implemented.

The defect. `issue_lock_store.assess_lock_freshness` parsed `last_heartbeat_at`
and then never consulted it. Liveness was decided by an absolute four-hour
`expires_at` and by PID liveness, and the recorded PID is the long-lived MCP
daemon rather than the authoring task, so an abandoned claim stayed live for the
full four hours. A tree-wide search found the field written in exactly one place
and advanced by nothing. Issue #787 / PR #789 hit this; Issue #760 / PR #791 hit
it again, blocking reconciliation for over five hours after its work had landed.

A1 — central policy. New `lease_policy` declares every duration for every task
class in one place: author initial/sliding TTL 10 minutes, heartbeat cadence 2,
stale warning 5, missed-heartbeat grace 10, absolute cap 8 hours, recovery grace
10, terminal race-drain 2. It ships first so the first heartbeat and TTL
behavior to run reads from it (AC-N7). The duplicated four-hour literal is gone
from both `issue_lock_store` and `gitea_mcp_server`. Reviewer, merger, and
conflict-fix classes are declared but not rewired — Slice C moves those call
sites — and a test asserts the declaration still equals the constants #747 and
`pr_work_lease` own, so the two cannot drift apart unnoticed.

A2 — load-bearing freshness, with two deliberate asymmetries. An alive PID never
establishes freshness anywhere (AC-N2); it is recorded as evidence and no branch
returns live because of it. A dead PID still marks a lease stale, and that band
still precedes every heartbeat evaluation, so #753 dead-session recovery keys on
exactly the classification it always did. New bands `stale_missed_heartbeat` and
`stale_absolute_cap` are classified in `branch_cleanup_guard` rather than
falling through to unknown-status, and still block unless the ownership record
proves `reclaim_allowed is True`. A heartbeat lease carrying no heartbeat is
contradictory and fails closed. `assess_expired_lock_reclaim` accepts a lapsed
heartbeat as reclaim grounds for heartbeat-lifecycle leases only: under this
lifecycle the heartbeat is the liveness proof, and also requiring a dead PID
would reinstate the original defect.

A3/A4 — task-session identity and the writer. `mint_task_session_id` produces an
ownership key containing no process identifier, since the daemon PID is reused
by every task it serves and identifies none of them. `heartbeat_session_lock`
writes inside the existing per-issue flock under the #772 generation
compare-and-swap, verifying exact issue, branch, realpath-normalized worktree,
claimant username, claimant profile, and recorded session identifier. It cannot
acquire, take over, or revive: a lease past its grace is refused and must use
the reclaim path, so a session that stopped proving liveness cannot restore
ownership retroactively. New `gitea_heartbeat_issue_lock` gates on the same
authority as `lock_issue`, being strictly narrower.

A5 — legacy compatibility (AC-N8). The explicit `lifecycle_version` marker, never
a timestamp comparison, discriminates legacy from heartbeat leases: a legacy lock
has `last_heartbeat_at == created_at` forever precisely because nothing advanced
it, and a freshly minted heartbeat lease has them equal too, so the equality
carries no information in either direction. Legacy locks keep their recorded
absolute expiry and are never evaluated against the short grace, so deployment
cannot make an existing claim instantly reclaimable. They leave that state only
by terminal retirement (Slice B) or by `rebind_legacy_lock`, which re-verifies
the exact owner and mints a genuine identifier and first heartbeat while
preserving the original claim under `legacy_origin`. Rebinding a lapsed legacy
lease is refused; that belongs to #760 renewal or #601 reclaim.

A6 — native coverage. Review #499 proved assessor-level tests miss discard
points, so `tests/test_issue_790_heartbeat_mcp_path.py` drives the real tools
against a real git repository and a real durable lock: lock creation and
read-back, policy window, freshness, survival of `verify_lock_for_mutation`,
invariance of the duplicate-work and linked-open-PR gates, CAS rejection,
foreign-session and foreign-claimant refusal, alive-PID-only refusal, missed
heartbeat, legacy protection on deployment, and legacy rebinding.

Tests. New suites 55 passed. Lock and lease regression set (issue_lock_store,
lease_lifecycle, #753, #755, #760 x2, #768, #772, lock registration, worktree,
adoption, duplicate gate, branch cleanup guard, capability invariants, claim
heartbeat, worktrees) 383 passed with 98 subtests. Full suite 4295 passed, 11
failed, 6 skipped, 499 subtests passed, against a clean master baseline worktree
at 620ed6e9 that reports 11 failed and 4240 passed — the same eleven node IDs.
The 55-test delta is exactly the new suites; no new failures.

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

branch_cleanup_guard.py

@@ -163,7 +163,19 @@ _TERMINAL_OWNERSHIP_STATUSES = frozenset(
     {"released", "abandoned", "done", "blocked", "terminal", "closed"}
 )
 _EXPIRED_STATUSES = frozenset({"expired"})
-_STALE_STATUSES = frozenset({"stale", "stale_dead_process", "stale_missing_worktree"})
+_STALE_STATUSES = frozenset(
+    {
+        "stale",
+        "stale_dead_process",
+        "stale_missing_worktree",
+        # #790 Slice A heartbeat-lifecycle bands. Listed here so they are
+        # *classified* rather than falling through to the unknown-status branch;
+        # they still block unless the ownership record proves
+        # ``reclaim_allowed is True``, so the O2 fail-closed rule is unchanged.
+        "stale_missed_heartbeat",
+        "stale_absolute_cap",
+    }
+)
 
 
 def _norm_str(value: Any) -> str:

docs/mcp-tool-inventory.md

@@ -100,6 +100,7 @@ that gates each call, not which tools exist.
 - `gitea_get_profile`
 - `gitea_get_runtime_context`
 - `gitea_get_shell_health`
+- `gitea_heartbeat_issue_lock`
 - `gitea_heartbeat_reviewer_pr_lease`
 - `gitea_inspect_workflow_lease`
 - `gitea_issue_irrecoverable_provenance_authorization`

gitea_mcp_server.py

@@ -2007,6 +2007,7 @@ import allocator_dependencies  # noqa: E402
 import dependency_graph  # noqa: E402  # #784 durable dependency edges
 import control_plane_db  # noqa: E402
 import lease_lifecycle  # noqa: E402
+import lease_policy  # noqa: E402
 import workflow_dashboard  # noqa: E402  # #605 live queue/lease dashboard
 import incident_bridge  # noqa: E402
 import sentry_observability  # noqa: E402  (#606 optional Sentry observability)
@@ -2248,7 +2249,6 @@ import canonical_comment_validator as ccv  # noqa: E402
 # GITEA_ISSUE_LOCK_DIR, bound to the current MCP session via a per-PID pointer.
 # Legacy global path retained only for test/doc references — do not seed manually.
 ISSUE_LOCK_FILE = "/tmp/gitea_issue_lock.json"
-WORK_LEASE_TTL_HOURS = 4
 AUTHOR_ISSUE_WORK_LEASE = "author_issue_work"
 VALID_WORK_LEASE_OPERATIONS = frozenset({
     AUTHOR_ISSUE_WORK_LEASE,
@@ -2548,7 +2548,12 @@ def _build_author_issue_work_lease(
     host: str | None,
 ) -> dict:
     created = _work_lease_now()
-    expires = created + timedelta(hours=WORK_LEASE_TTL_HOURS)
+    # #790 Slice A: the window comes from the central policy, not a literal here.
+    # It is also now a *sliding* window — the lease lives ``initial_ttl_minutes``
+    # past its last valid heartbeat rather than a fixed four hours past its
+    # creation, so an abandoned task stops holding the claim within one TTL.
+    policy = lease_policy.policy_for(lease_policy.TASK_CLASS_AUTHOR_ISSUE_WORK)
+    expires = created + timedelta(minutes=policy.initial_ttl_minutes)
     return {
         "operation_type": AUTHOR_ISSUE_WORK_LEASE,
         "issue_number": issue_number,
@@ -2559,6 +2564,15 @@ def _build_author_issue_work_lease(
         "created_at": _work_lease_timestamp(created),
         "expires_at": _work_lease_timestamp(expires),
         "last_heartbeat_at": _work_lease_timestamp(created),
+        # #790 AC-N1: the ownership key for this task. Distinct from the recorded
+        # PID, which is the shared daemon and identifies no individual task.
+        "task_session_id": issue_lock_store.mint_task_session_id(
+            AUTHOR_ISSUE_WORK_LEASE
+        ),
+        # #790 AC-N8: the explicit lifecycle marker. Its absence — never a
+        # timestamp comparison — is what makes a lock legacy.
+        "lifecycle_version": lease_policy.LIFECYCLE_HEARTBEAT_V1,
+        "heartbeat_count": 1,
     }
 
 
@@ -4328,6 +4342,138 @@ def gitea_lock_issue(
     return result
 
 
+@mcp.tool()
+def gitea_heartbeat_issue_lock(
+    issue_number: int,
+    branch_name: str,
+    task_session_id: str | None = None,
+    remote: str = "dadeschools",
+    host: str | None = None,
+    org: str | None = None,
+    repo: str | None = None,
+    worktree_path: str | None = None,
+    expected_generation: int | None = None,
+) -> dict:
+    """Prove an owned author issue lease is still active (#790 Slice A).
+
+    The task-liveness signal the lifecycle was missing. Before this, an author
+    lease carried a fixed four-hour expiry that nothing could shorten, and the
+    only liveness evidence was the recorded PID — the long-lived MCP daemon,
+    which stays alive across every task it serves and so proved nothing about
+    whether the authoring task still held the work.
+
+    Each successful call slides the lease ``initial_ttl_minutes`` past *now*
+    from the central policy, so an actively heartbeating session is never
+    evicted while an abandoned one releases its claim within one TTL.
+
+    What this tool cannot do, by construction:
+
+    * **Acquire.** It refuses when no durable lock exists.
+    * **Take over.** Exact issue, branch, realpath-normalized worktree,
+      claimant username, claimant profile, and recorded task-session identifier
+      must all match; a superseded session holding an older identifier is
+      refused.
+    * **Revive.** A lease already past its grace is not heartbeatable — that
+      would let a session restore ownership it had stopped proving. It must use
+      the sanctioned reclaim path, which mints a new generation.
+
+    A lock predating the heartbeat lifecycle is rebound rather than heartbeated:
+    its exact owner is re-verified and a genuine task-session identifier and
+    first heartbeat are minted (#790 AC-N8). The rebind is decided server-side
+    from the durable lifecycle marker; there is no caller-facing switch.
+
+    Args:
+        issue_number: The locked issue number.
+        branch_name: The branch recorded on the lock.
+        task_session_id: The identifier this session received when it acquired
+            or rebound the lock. It is a fencing token, not an ownership
+            assertion: it is compared against durable state and can only ever
+            cause a refusal, never grant anything. Omitted only when rebinding a
+            legacy lock, which has no identifier yet and mints one.
+        remote: Known instance — 'dadeschools' or 'prgs'.
+        host: Override the Gitea host.
+        org: Override the owner/organization.
+        repo: Override the repository name.
+        worktree_path: Author worktree recorded on the lock.
+        expected_generation: Optional fencing value. The per-issue flock already
+            serializes the read and the write, so this is for a caller that
+            wants to pin the generation it last observed across calls; a moved
+            generation fails closed.
+
+    Returns:
+        dict with 'success', 'performed', the sliding 'expires_at',
+        'last_heartbeat_at', 'lock_generation', 'task_session_id', the applied
+        'policy', and post-write 'freshness'; on refusal 'success'/'performed'
+        False with 'reasons' naming exactly what did not match.
+    """
+    blocked = _profile_permission_block(
+        task_capability_map.required_permission("heartbeat_issue_lock"),
+        issue_number=issue_number,
+        remote=remote,
+        host=host,
+        org=org,
+        repo=repo,
+        org_explicit=org is not None,
+        repo_explicit=repo is not None,
+    )
+    if blocked:
+        return blocked
+
+    resolved_worktree = issue_lock_worktree.resolve_author_worktree_path(
+        worktree_path, _canonical_local_git_root()
+    )
+    h, o, r = _resolve(remote, host, org, repo)
+    claimant = _work_lease_claimant(h)
+    identity = claimant.get("username")
+    profile = claimant.get("profile")
+
+    existing = _load_existing_issue_lock(
+        remote=remote, org=o, repo=r, issue_number=issue_number
+    )
+    if not existing:
+        return {
+            "success": False,
+            "performed": False,
+            "issue_number": issue_number,
+            "reasons": [
+                f"no durable lock for issue #{issue_number}; heartbeat cannot "
+                "acquire a claim (fail closed)"
+            ],
+        }
+
+    if issue_lock_store.is_legacy_lease(existing):
+        # AC-N8 exit route one: canonical exact-owner rebinding. The other exit
+        # is terminal retirement, which is Slice B.
+        outcome = issue_lock_store.rebind_legacy_lock(
+            remote=remote,
+            org=o,
+            repo=r,
+            issue_number=issue_number,
+            branch_name=branch_name,
+            worktree_path=resolved_worktree,
+            identity=identity,
+            profile=profile,
+            expected_generation=expected_generation,
+        )
+        outcome["operation"] = "legacy_rebind"
+        return outcome
+
+    outcome = issue_lock_store.heartbeat_session_lock(
+        remote=remote,
+        org=o,
+        repo=r,
+        issue_number=issue_number,
+        branch_name=branch_name,
+        worktree_path=resolved_worktree,
+        identity=identity,
+        profile=profile,
+        task_session_id=str(task_session_id or ""),
+        expected_generation=expected_generation,
+    )
+    outcome["operation"] = "heartbeat"
+    return outcome
+
+
 @mcp.tool()
 def gitea_assess_work_issue_duplicate(
     issue_number: int,
@@ -12536,39 +12682,10 @@ def _try_auto_switch_for_operation(op: str, host: str | None = None) -> bool:
     return False
 
 
-def _git_default_remote_name(root: str) -> str:
-    """First configured git remote name for *root*, defaulting to 'origin'.
-
-    Used to resolve the live remote master target for parity (#610). Best
-    effort: any failure falls back to 'origin' so callers never raise.
-    """
-    try:
-        res = subprocess.run(
-            ["git", "-C", root, "remote"],
-            capture_output=True, text=True, check=False,
-        )
-    except Exception:
-        return "origin"
-    if res.returncode != 0:
-        return "origin"
-    names = [n.strip() for n in (res.stdout or "").splitlines() if n.strip()]
-    return names[0] if names else "origin"
-
-
 def _current_master_parity() -> dict:
-    """Assess this process's code against local and live remote master (#420/#610).
-
-    Compares the daemon's startup commit, the on-disk checkout HEAD, and the
-    live remote master target. A stale daemon relative to live master fails
-    closed for mutations even when the local checkout HEAD still matches the
-    startup commit. The live-remote read is best effort: an unresolved live
-    head leaves read-only diagnostics unblocked but is never mutation-safe.
-    """
+    """Assess this process's code against the on-disk master HEAD (#420)."""
     current_head = master_parity_gate.read_git_head(PROJECT_ROOT)
-    live_head = master_parity_gate.read_remote_master_head(
-        PROJECT_ROOT, remote=_git_default_remote_name(PROJECT_ROOT))
-    return master_parity_gate.assess_master_parity(
-        _STARTUP_PARITY, current_head, live_remote_head=live_head)
+    return master_parity_gate.assess_master_parity(_STARTUP_PARITY, current_head)
 
 
 def _current_runtime_mode_report(refresh: bool = False) -> dict:
@@ -16509,34 +16626,15 @@ def gitea_get_runtime_context(
         "restart_required": parity["restart_required"],
         "startup_head": parity["startup_head"],
         "current_head": parity["current_head"],
-        # #610 distinguished mutation-safety signals:
-        "daemon_start_head": parity["daemon_start_head"],
-        "local_head": parity["local_head"],
-        "live_remote_head": parity["live_remote_head"],
-        "live_known": parity["live_known"],
-        "live_stale": parity["live_stale"],
-        "mutation_safe": parity["mutation_safe"],
         "summary": master_parity_gate.format_parity(parity),
         "mutation_gate_enforced": not master_parity_gate.gate_disabled(),
-        # #610: the capability resolver is authoritative for mutation safety;
-        # local parity alone must never authorize a mutation.
-        "resolver_authoritative_for_mutation_safety": True,
     }
-    if parity["restart_required"] and not master_parity_gate.gate_disabled():
-        if parity["live_stale"]:
-            safe_next_action = (
-                "Daemon is stale relative to LIVE remote master "
-                f"(started {parity['startup_head'][:12] if parity['startup_head'] else 'unknown'}, "
-                f"live master {parity['live_remote_head'][:12] if parity['live_remote_head'] else 'unknown'}); "
-                "restart/reconnect the Gitea MCP server before mutating. The "
-                "capability resolver is authoritative for mutation safety."
-            )
-        else:
-            safe_next_action = (
-                "Server code is stale relative to master; restart the Gitea MCP "
-                "server to load current capability gates before mutating. "
-                f"({master_parity_gate.format_parity(parity)})"
-            )
+    if parity["stale"] and not master_parity_gate.gate_disabled():
+        safe_next_action = (
+            "Server code is stale relative to master; restart the Gitea MCP "
+            "server to load current capability gates before mutating. "
+            f"({master_parity_gate.format_parity(parity)})"
+        )
         result["safe_next_action"] = safe_next_action
 
     if reveal and h:
@@ -16585,13 +16683,6 @@ def gitea_assess_master_parity(
         "determinable": parity["determinable"],
         "startup_head": parity["startup_head"],
         "current_head": parity["current_head"],
-        # #610 distinguished mutation-safety signals:
-        "daemon_start_head": parity["daemon_start_head"],
-        "local_head": parity["local_head"],
-        "live_remote_head": parity["live_remote_head"],
-        "live_known": parity["live_known"],
-        "live_stale": parity["live_stale"],
-        "mutation_safe": parity["mutation_safe"],
         "mutation_gate_enforced": enforced,
         "summary": master_parity_gate.format_parity(parity),
         "reasons": parity["reasons"],
@@ -16608,7 +16699,7 @@ def gitea_assess_master_parity(
             source=canonical_source,
         ),
     }
-    if parity["restart_required"] and enforced:
+    if parity["stale"] and enforced:
         out["report"] = master_parity_gate.parity_report(parity)
     return out
 

issue_lock_store.py

@@ -15,15 +15,27 @@ import json
 import os
 import re
 import tempfile
+import uuid
 from contextlib import contextmanager
 from datetime import datetime, timedelta, timezone
 from typing import Any
 
+import lease_policy
+
 LOCK_DIR_ENV = "GITEA_ISSUE_LOCK_DIR"
 DEFAULT_LOCK_DIR = os.path.expanduser("~/.cache/gitea-tools/issue-locks")
-WORK_LEASE_TTL_HOURS = 4
 AUTHOR_ISSUE_WORK_LEASE = "author_issue_work"
 
+# Freshness classifications. ``STATUS_STALE`` remains the dead-PID band that
+# #753 recovery keys on; the two bands below are new in #790 Slice A and apply
+# only to leases minted under the heartbeat lifecycle.
+STATUS_LIVE = "live"
+STATUS_EXPIRED = "expired"
+STATUS_ABSENT = "absent"
+STATUS_STALE = "stale"
+STATUS_STALE_MISSED_HEARTBEAT = "stale_missed_heartbeat"
+STATUS_STALE_ABSOLUTE_CAP = "stale_absolute_cap"
+
 _SAFE_SEGMENT_RE = re.compile(r"[^A-Za-z0-9._+-]+")
 
 
@@ -253,6 +265,331 @@ def bind_session_lock(
     return path
 
 
+def _ownership_refusals(
+    lock: dict[str, Any],
+    *,
+    issue_number: int,
+    branch_name: str,
+    worktree_path: str,
+    identity: str | None,
+    profile: str | None,
+) -> list[str]:
+    """Exact-ownership mismatches between a durable lock and a live caller.
+
+    Shared by the heartbeat writer and the legacy rebind path so the two cannot
+    disagree about what "the same owner" means. Every field is compared against
+    durable state; nothing is taken on the caller's word beyond the identity the
+    server itself resolved.
+    """
+    reasons: list[str] = []
+    if lock.get("issue_number") != issue_number:
+        reasons.append(
+            f"lock targets issue #{lock.get('issue_number')}, not #{issue_number}"
+        )
+    if str(lock.get("branch_name") or "") != str(branch_name or ""):
+        reasons.append(
+            f"lock branch '{lock.get('branch_name')}' does not match '{branch_name}'"
+        )
+    if not _same_realpath(str(lock.get("worktree_path") or ""), worktree_path):
+        reasons.append(
+            f"lock worktree '{lock.get('worktree_path')}' does not match "
+            f"'{worktree_path}'"
+        )
+    lease = lock.get("work_lease") if isinstance(lock, dict) else None
+    claimant = lease.get("claimant") if isinstance(lease, dict) else None
+    claimant = claimant if isinstance(claimant, dict) else {}
+    recorded_identity = str(claimant.get("username") or "").strip()
+    recorded_profile = str(claimant.get("profile") or "").strip()
+    if not recorded_identity or not recorded_profile:
+        reasons.append("lock does not record both a claimant username and profile")
+    if recorded_identity and recorded_identity != str(identity or "").strip():
+        reasons.append(
+            f"lock claimant '{recorded_identity}' does not match active identity "
+            f"'{str(identity or '').strip() or 'unknown'}'"
+        )
+    if recorded_profile and recorded_profile != str(profile or "").strip():
+        reasons.append(
+            f"lock profile '{recorded_profile}' does not match active profile "
+            f"'{str(profile or '').strip() or 'unknown'}'"
+        )
+    return reasons
+
+
+def _refusal(reasons: list[str], **extra: Any) -> dict[str, Any]:
+    return {"success": False, "performed": False, "reasons": reasons, **extra}
+
+
+def heartbeat_session_lock(
+    *,
+    remote: str,
+    org: str,
+    repo: str,
+    issue_number: int,
+    branch_name: str,
+    worktree_path: str,
+    identity: str | None,
+    profile: str | None,
+    task_session_id: str,
+    expected_generation: int | None = None,
+    lock_dir: str | None = None,
+    now: datetime | None = None,
+) -> dict[str, Any]:
+    """Slide a heartbeat-lifecycle lease forward (#790 Slice A, A4).
+
+    The write happens inside the same per-issue ``flock`` that serializes
+    acquisition, and under the #772 generation compare-and-swap, so a heartbeat
+    can never race a concurrent reclaim: whichever lands first moves the
+    generation and the other fails closed.
+
+    Refuses — never revives — in every ambiguous case. A lease that has already
+    lapsed past its grace is *not* heartbeatable: allowing that would let a
+    session that stopped proving liveness restore ownership retroactively, which
+    is precisely the revival AC-N5 forbids. Such a session must go through the
+    sanctioned reclaim path, which mints a fresh generation.
+    """
+    current = _lease_now(now)
+    root = _ensure_lock_dir(lock_dir)
+    path = lock_file_path(
+        remote=remote, org=org, repo=repo, issue_number=issue_number, lock_dir=root
+    )
+    declared_session = str(task_session_id or "").strip()
+    if not declared_session:
+        return _refusal(["no task_session_id supplied (fail closed)"])
+
+    sentinel = flock_path(path)
+    try:
+        with _exclusive_file_lock(sentinel):
+            lock = read_lock_file(path)
+            if not lock:
+                return _refusal([f"no durable lock for issue #{issue_number}"])
+
+            if is_legacy_lease(lock):
+                return _refusal(
+                    [
+                        "lock predates the heartbeat lifecycle; it must be rebound "
+                        "by its exact owner before it can be heartbeated"
+                    ],
+                    lifecycle=lease_lifecycle_version(lock),
+                    legacy_lease=True,
+                )
+
+            reasons = _ownership_refusals(
+                lock,
+                issue_number=issue_number,
+                branch_name=branch_name,
+                worktree_path=worktree_path,
+                identity=identity,
+                profile=profile,
+            )
+            recorded_session = lease_task_session_id(lock)
+            if not recorded_session:
+                reasons.append(
+                    "lock declares the heartbeat lifecycle but records no "
+                    "task_session_id (fail closed)"
+                )
+            elif recorded_session != declared_session:
+                # A superseded session holding an old identifier cannot heartbeat
+                # over the session that replaced it.
+                reasons.append(
+                    "task_session_id does not match the session recorded on the lock"
+                )
+            if reasons:
+                return _refusal(reasons)
+
+            current_generation = lock_generation(lock)
+            if (
+                expected_generation is not None
+                and current_generation != expected_generation
+            ):
+                return _refusal(
+                    [
+                        f"lock generation changed: expected {expected_generation}, "
+                        f"found {current_generation}; another session reclaimed or "
+                        "replaced this claim (fail closed)"
+                    ],
+                    lock_generation=current_generation,
+                )
+
+            freshness = assess_lock_freshness(lock, now=current)
+            if not freshness.get("live"):
+                return _refusal(
+                    [
+                        f"lease is not live ({freshness.get('status')}): "
+                        f"{freshness.get('reason')}; a lapsed lease must be "
+                        "reclaimed, not heartbeated"
+                    ],
+                    freshness=freshness,
+                )
+
+            policy = lease_policy.policy_for(lease_task_class(lock))
+            expires = current + timedelta(minutes=policy.initial_ttl_minutes)
+            record = dict(lock)
+            lease = dict(record.get("work_lease") or {})
+            prior_heartbeat = lease.get("last_heartbeat_at")
+            lease["last_heartbeat_at"] = _format_lease_timestamp(current)
+            lease["expires_at"] = _format_lease_timestamp(expires)
+            try:
+                lease["heartbeat_count"] = int(lease.get("heartbeat_count") or 0) + 1
+            except (TypeError, ValueError):
+                lease["heartbeat_count"] = 1
+            record["work_lease"] = lease
+            record["lock_generation"] = current_generation + 1
+            save_lock_file(path, record)
+    except LockContentionError as exc:
+        return _refusal([f"issue #{issue_number} lock contention: {exc} (fail closed)"])
+
+    return {
+        "success": True,
+        "performed": True,
+        "issue_number": issue_number,
+        "branch_name": branch_name,
+        "worktree_path": worktree_path,
+        "task_session_id": declared_session,
+        "lock_generation": record["lock_generation"],
+        "prior_generation": current_generation,
+        "prior_heartbeat_at": prior_heartbeat,
+        "last_heartbeat_at": lease["last_heartbeat_at"],
+        "expires_at": lease["expires_at"],
+        "heartbeat_count": lease["heartbeat_count"],
+        "lock_file_path": path,
+        "policy": lease_policy.describe(lease_task_class(record)),
+        "freshness": assess_lock_freshness(record, now=current),
+    }
+
+
+def rebind_legacy_lock(
+    *,
+    remote: str,
+    org: str,
+    repo: str,
+    issue_number: int,
+    branch_name: str,
+    worktree_path: str,
+    identity: str | None,
+    profile: str | None,
+    expected_generation: int | None = None,
+    lock_dir: str | None = None,
+    now: datetime | None = None,
+) -> dict[str, Any]:
+    """Move a legacy lock into the heartbeat lifecycle (#790 AC-N8).
+
+    One of the two sanctioned exits from the preserved-expiry legacy state; the
+    other is terminal retirement, which is Slice B. Only the exact recorded
+    owner may rebind, and only while the legacy lock is still live under its
+    original absolute expiry — an already-expired legacy lease belongs to the
+    #760 renewal path or #601 reclaim, and this must not become a second, weaker
+    way to revive one.
+
+    The rebind mints a genuine task-session identifier and a genuine first
+    heartbeat. It does not fabricate history: the original creation and expiry
+    are preserved under ``legacy_origin`` for audit, and the new lifecycle's
+    absolute cap runs from the rebind, not from the legacy claim.
+    """
+    current = _lease_now(now)
+    root = _ensure_lock_dir(lock_dir)
+    path = lock_file_path(
+        remote=remote, org=org, repo=repo, issue_number=issue_number, lock_dir=root
+    )
+    sentinel = flock_path(path)
+    try:
+        with _exclusive_file_lock(sentinel):
+            lock = read_lock_file(path)
+            if not lock:
+                return _refusal([f"no durable lock for issue #{issue_number}"])
+            if not is_legacy_lease(lock):
+                return _refusal(
+                    [
+                        "lock is already on the heartbeat lifecycle; use the "
+                        "heartbeat path"
+                    ],
+                    lifecycle=lease_lifecycle_version(lock),
+                    legacy_lease=False,
+                )
+
+            reasons = _ownership_refusals(
+                lock,
+                issue_number=issue_number,
+                branch_name=branch_name,
+                worktree_path=worktree_path,
+                identity=identity,
+                profile=profile,
+            )
+            if reasons:
+                return _refusal(reasons)
+
+            current_generation = lock_generation(lock)
+            if (
+                expected_generation is not None
+                and current_generation != expected_generation
+            ):
+                return _refusal(
+                    [
+                        f"lock generation changed: expected {expected_generation}, "
+                        f"found {current_generation} (fail closed)"
+                    ],
+                    lock_generation=current_generation,
+                )
+
+            freshness = assess_lock_freshness(lock, now=current)
+            if not freshness.get("live"):
+                return _refusal(
+                    [
+                        f"legacy lease is not live ({freshness.get('status')}): "
+                        f"{freshness.get('reason')}; rebinding is not a recovery "
+                        "path for a lapsed lease"
+                    ],
+                    freshness=freshness,
+                )
+
+            policy = lease_policy.policy_for(lease_task_class(lock))
+            expires = current + timedelta(minutes=policy.initial_ttl_minutes)
+            session_id = mint_task_session_id(lease_task_class(lock))
+            record = dict(lock)
+            lease = dict(record.get("work_lease") or {})
+            legacy_origin = {
+                "created_at": lease.get("created_at"),
+                "expires_at": lease.get("expires_at"),
+                "last_heartbeat_at": lease.get("last_heartbeat_at"),
+                "lifecycle": lease_policy.LIFECYCLE_LEGACY,
+            }
+            lease["lifecycle_version"] = lease_policy.LIFECYCLE_HEARTBEAT_V1
+            lease["task_session_id"] = session_id
+            lease["created_at"] = _format_lease_timestamp(current)
+            lease["last_heartbeat_at"] = _format_lease_timestamp(current)
+            lease["expires_at"] = _format_lease_timestamp(expires)
+            lease["heartbeat_count"] = 1
+            record["work_lease"] = lease
+            record["legacy_rebind"] = {
+                "rebound_at": _format_lease_timestamp(current),
+                "task_session_id": session_id,
+                "prior_generation": current_generation,
+                "legacy_origin": legacy_origin,
+                "reason": (
+                    "legacy lock rebound into the heartbeat lifecycle by its exact "
+                    "recorded owner"
+                ),
+            }
+            record["lock_generation"] = current_generation + 1
+            save_lock_file(path, record)
+    except LockContentionError as exc:
+        return _refusal([f"issue #{issue_number} lock contention: {exc} (fail closed)"])
+
+    return {
+        "success": True,
+        "performed": True,
+        "issue_number": issue_number,
+        "task_session_id": session_id,
+        "lock_generation": record["lock_generation"],
+        "prior_generation": current_generation,
+        "lifecycle": lease_policy.LIFECYCLE_HEARTBEAT_V1,
+        "legacy_rebind": record["legacy_rebind"],
+        "expires_at": lease["expires_at"],
+        "last_heartbeat_at": lease["last_heartbeat_at"],
+        "lock_file_path": path,
+        "freshness": assess_lock_freshness(record, now=current),
+    }
+
+
 def read_session_issue_lock(lock_dir: str | None = None) -> dict[str, Any] | None:
     root = (lock_dir or default_lock_dir()).strip()
     pointer = read_lock_file(session_pointer_path(root))
@@ -336,6 +673,16 @@ def _parse_lease_timestamp(value: str | None) -> datetime | None:
         return None
 
 
+def _format_lease_timestamp(value: datetime) -> str:
+    """Serialize a lease timestamp in the durable ``...Z`` form already on disk."""
+    return (
+        value.astimezone(timezone.utc)
+        .replace(microsecond=0)
+        .isoformat()
+        .replace("+00:00", "Z")
+    )
+
+
 def lease_expires_at(lock: dict[str, Any] | None) -> datetime | None:
     if not lock:
         return None
@@ -356,60 +703,216 @@ def is_lease_live(lock: dict[str, Any] | None, *, now: datetime | None = None) -
     return assess_lock_freshness(lock, now=now)["live"]
 
 
+def lease_task_class(lock_data: dict[str, Any] | None) -> str:
+    """Policy task class for a durable lock; author work when unrecorded."""
+    lease = lock_data.get("work_lease") if isinstance(lock_data, dict) else None
+    if isinstance(lease, dict):
+        recorded = str(lease.get("operation_type") or "").strip()
+        if recorded:
+            return recorded
+    return AUTHOR_ISSUE_WORK_LEASE
+
+
+def lease_lifecycle_version(lock_data: dict[str, Any] | None) -> str:
+    """Read the durable lifecycle marker (#790 AC-N8).
+
+    The marker is the *only* discriminator between a heartbeat-lifecycle lease
+    and a legacy one. Timestamps are deliberately not consulted: a lock minted
+    before this lifecycle existed has ``last_heartbeat_at == created_at``
+    forever, and reading that equality as "recently heartbeated" would treat
+    every never-heartbeated legacy lock as fresh — the precise inversion AC-N8
+    forbids. A newly minted heartbeat lease also has the two equal, so the
+    equality carries no information in either direction.
+    """
+    lease = lock_data.get("work_lease") if isinstance(lock_data, dict) else None
+    if isinstance(lease, dict):
+        recorded = str(lease.get("lifecycle_version") or "").strip()
+        if recorded:
+            return recorded
+    return lease_policy.LIFECYCLE_LEGACY
+
+
+def is_legacy_lease(lock_data: dict[str, Any] | None) -> bool:
+    """True when a lock predates the shared heartbeat lifecycle."""
+    return lease_lifecycle_version(lock_data) != lease_policy.LIFECYCLE_HEARTBEAT_V1
+
+
+def lease_task_session_id(lock_data: dict[str, Any] | None) -> str:
+    """Recorded per-task session identifier, or empty for a legacy lock."""
+    lease = lock_data.get("work_lease") if isinstance(lock_data, dict) else None
+    if isinstance(lease, dict):
+        return str(lease.get("task_session_id") or "").strip()
+    return ""
+
+
+def mint_task_session_id(task_class: str = AUTHOR_ISSUE_WORK_LEASE) -> str:
+    """Mint an ownership key for one task (#790 AC-N1).
+
+    Deliberately contains no process identifier. The recorded PID belongs to the
+    long-lived MCP daemon, which outlives any individual task and is reused by
+    every task it serves, so PID digits cannot identify *which* task holds a
+    claim. The PID is still recorded alongside this value as evidence.
+    """
+    prefix = _sanitize_segment(str(task_class or AUTHOR_ISSUE_WORK_LEASE))
+    return f"{prefix}-{uuid.uuid4().hex[:16]}"
+
+
+def _lease_heartbeat_at(lock_data: dict[str, Any] | None) -> datetime | None:
+    lease = lock_data.get("work_lease") if isinstance(lock_data, dict) else None
+    heartbeat_at = None
+    if isinstance(lock_data, dict):
+        heartbeat_at = _parse_lease_timestamp(lock_data.get("last_heartbeat_at"))
+    if heartbeat_at is None and isinstance(lease, dict):
+        heartbeat_at = _parse_lease_timestamp(lease.get("last_heartbeat_at"))
+    return heartbeat_at
+
+
 def assess_lock_freshness(
     lock_data: dict[str, Any] | None,
     *,
     now: datetime | None = None,
 ) -> dict[str, Any]:
-    """Classify a lock as live, expired, stale, or absent."""
+    """Classify a lock as live, expired, stale, or absent.
+
+    #790 Slice A makes the heartbeat load-bearing. Before this change
+    ``last_heartbeat_at`` was parsed and then never consulted: liveness was
+    decided entirely by the absolute ``expires_at`` and by PID liveness, and
+    since the recorded PID is the long-lived MCP daemon, an abandoned author
+    task stayed "live" for the full four-hour TTL.
+
+    Two rules govern the rewrite:
+
+    * **An alive PID never establishes freshness** (AC-N2). It proves the daemon
+      is up, nothing about the task. It is recorded as evidence and no branch
+      returns ``live`` because of it.
+    * **A dead PID still corroborates staleness.** The dead-PID band is
+      unchanged and still precedes every heartbeat evaluation, so #753
+      dead-session recovery keys on exactly the classification it always did.
+
+    Legacy leases (AC-N8) keep their recorded absolute expiry and are never
+    evaluated against the short heartbeat grace, so deploying this change cannot
+    make an existing claim instantly reclaimable.
+    """
     current = _lease_now(now)
     if not lock_data:
         return {
-            "status": "absent",
+            "status": 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"))
+    expires_at = lease_expires_at(lock_data)
+    heartbeat_at = _lease_heartbeat_at(lock_data)
+    created_at = (
+        _parse_lease_timestamp(lease.get("created_at"))
+        if isinstance(lease, dict)
+        else None
+    )
 
     pid = lock_data.get("session_pid")
     if pid is None:
         pid = lock_data.get("pid")
+    # Evidence only. Never consulted to grant liveness (AC-N2).
     pid_alive = is_process_alive(pid) if pid is not None else False
 
-    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,
-        }
+    lifecycle = lease_lifecycle_version(lock_data)
+    legacy = lifecycle != lease_policy.LIFECYCLE_HEARTBEAT_V1
+    policy = lease_policy.policy_for(lease_task_class(lock_data))
 
-    if pid is not None and not pid_alive:
-        return {
-            "status": "stale",
-            "live": False,
-            "stale": True,
-            "reason": f"owner pid {pid} is not alive",
-            "pid_alive": False,
-        }
-
-    return {
-        "status": "live",
-        "live": True,
-        "stale": False,
-        "reason": "lock heartbeat and lease are fresh",
+    evidence: dict[str, Any] = {
         "pid_alive": pid_alive,
+        "lifecycle": lifecycle,
+        "legacy_lease": legacy,
+        "task_session_id": lease_task_session_id(lock_data) or None,
         "heartbeat_at": heartbeat_at.isoformat() if heartbeat_at else None,
         "expires_at": expires_at.isoformat() if expires_at else None,
     }
 
+    def _result(status: str, *, live: bool, reason: str, **extra: Any) -> dict[str, Any]:
+        return {
+            "status": status,
+            "live": live,
+            "stale": not live and status != STATUS_ABSENT,
+            "reason": reason,
+            **evidence,
+            **extra,
+        }
+
+    if legacy:
+        # AC-N8: the preserved absolute expiry is the only clock for a lock
+        # written before task-session heartbeats existed.
+        if expires_at and expires_at <= current:
+            return _result(
+                STATUS_EXPIRED,
+                live=False,
+                reason=f"lease expired at {expires_at.isoformat()}",
+            )
+        if pid is not None and not pid_alive:
+            return _result(
+                STATUS_STALE, live=False, reason=f"owner pid {pid} is not alive"
+            )
+        return _result(
+            STATUS_LIVE,
+            live=True,
+            reason=(
+                "legacy lease is within its recorded absolute expiry; the "
+                "heartbeat grace does not apply retroactively"
+            ),
+            legacy_expiry_preserved=True,
+        )
+
+    # ── Heartbeat lifecycle ──
+    if pid is not None and not pid_alive:
+        # Unchanged dead-PID band: #753 recovery depends on this exact status.
+        return _result(STATUS_STALE, live=False, reason=f"owner pid {pid} is not alive")
+
+    if heartbeat_at is None:
+        # Contradictory: a heartbeat lease must carry a heartbeat. Fail closed.
+        return _result(
+            STATUS_STALE_MISSED_HEARTBEAT,
+            live=False,
+            reason=(
+                f"lease declares lifecycle '{lifecycle}' but records no "
+                "last_heartbeat_at (fail closed)"
+            ),
+        )
+
+    if policy.absolute_cap_hours and created_at is not None:
+        cap_at = created_at + timedelta(hours=policy.absolute_cap_hours)
+        if cap_at <= current:
+            return _result(
+                STATUS_STALE_ABSOLUTE_CAP,
+                live=False,
+                reason=(
+                    f"lease exceeded its {policy.absolute_cap_hours}h absolute cap "
+                    f"at {cap_at.isoformat()}; canonical re-adoption is required"
+                ),
+                absolute_cap_at=cap_at.isoformat(),
+            )
+
+    grace_at = heartbeat_at + timedelta(minutes=policy.missed_heartbeat_grace_minutes)
+    if grace_at <= current or (expires_at is not None and expires_at <= current):
+        return _result(
+            STATUS_STALE_MISSED_HEARTBEAT,
+            live=False,
+            reason=(
+                f"no valid heartbeat since {heartbeat_at.isoformat()}; the "
+                f"{policy.missed_heartbeat_grace_minutes}min grace lapsed at "
+                f"{grace_at.isoformat()}"
+            ),
+            missed_heartbeat_since=grace_at.isoformat(),
+        )
+
+    warning_at = heartbeat_at + timedelta(minutes=policy.stale_warning_minutes)
+    return _result(
+        STATUS_LIVE,
+        live=True,
+        reason="lease heartbeat is fresh within the configured grace",
+        heartbeat_warning=warning_at <= current,
+    )
+
 
 def _same_realpath(left: str | None, right: str | None) -> bool:
     if not left or not right:
@@ -446,6 +949,27 @@ def assess_expired_lock_reclaim(
             "reasons": ["lock is still live; cannot reclaim (fail closed)"],
             "freshness": freshness,
         }
+    status = str(freshness.get("status") or "")
+    if status in (STATUS_STALE_MISSED_HEARTBEAT, STATUS_STALE_ABSOLUTE_CAP):
+        # #790: under the heartbeat lifecycle the heartbeat *is* the liveness
+        # proof, so a session that stopped heartbeating past its grace has
+        # released its claim by definition. Requiring a dead PID on top of that
+        # would reinstate the original defect — the recorded PID is the shared
+        # daemon, which stays alive across every abandoned task it ever served.
+        #
+        # This band is unreachable for a legacy lease (AC-N8), so no lock
+        # written before this lifecycle can be reclaimed by this path.
+        return {
+            "reclaim_allowed": True,
+            "reasons": [
+                f"heartbeat-lifecycle lease is {status}: {freshness.get('reason')}"
+            ],
+            "freshness": freshness,
+            "prior_branch": existing_lock.get("branch_name"),
+            "prior_worktree": existing_lock.get("worktree_path"),
+            "prior_pid": existing_lock.get("session_pid") or existing_lock.get("pid"),
+            "prior_task_session_id": lease_task_session_id(existing_lock) or None,
+        }
     pid = existing_lock.get("session_pid")
     if pid is None:
         pid = existing_lock.get("pid")

lease_policy.py

@@ -0,0 +1,212 @@
+"""Central lease policy configuration (#790 Slice A, AC-N7).
+
+The single authoritative source for every lease duration in the project. Before
+this module the numbers were scattered: a four-hour author TTL was declared
+twice (``issue_lock_store`` and ``gitea_mcp_server``), the reviewer/merger
+sliding window lived in ``reviewer_pr_lease``, the conflict-fix window in
+``pr_work_lease``, and the control-plane default in ``control_plane_db``.
+Nothing tied them together, so tuning one class silently diverged from the
+others and no reader could answer "how long does a lease live?" without
+grepping four files.
+
+AC-N7 requires that this configuration exist *before* the first heartbeat and
+TTL behavior that reads from it, so it ships in Slice A rather than trailing the
+code it governs.
+
+Deliberate boundaries:
+
+* **Declaration is not rewiring.** Every task class is declared here, but only
+  those with ``heartbeat_lifecycle_active`` were migrated onto the shared
+  heartbeat lifecycle in Slice A — currently ``author_issue_work`` alone.
+  Reviewer, merger, and conflict-fix leases keep their own existing behavior
+  until Slice C moves them; their numbers are recorded here so the two cannot
+  drift apart unnoticed, and ``tests/test_issue_790_lease_policy.py`` asserts
+  the recorded values still equal the constants those modules use.
+* **No policy decision lives here.** This module answers "how long", never "may
+  this session proceed". Freshness, reclaim, and renewal dispositions stay in
+  ``issue_lock_store``.
+"""
+
+from __future__ import annotations
+
+import os
+from dataclasses import dataclass
+from typing import Any
+
+# Task classes. Only the first is migrated onto the shared lifecycle in Slice A.
+TASK_CLASS_AUTHOR_ISSUE_WORK = "author_issue_work"
+TASK_CLASS_REVIEWER_PR = "reviewer_pr"
+TASK_CLASS_MERGER_PR = "merger_pr"
+TASK_CLASS_CONFLICT_FIX = "conflict_fix"
+
+# Durable marker for a lease minted under the shared heartbeat lifecycle.
+#
+# #790 AC-N8: this explicit marker — never a timestamp comparison — is what
+# distinguishes a heartbeat-lifecycle lease from a legacy one. A lock written
+# before this lifecycle existed carries no marker and reads as
+# ``LIFECYCLE_LEGACY``.
+LIFECYCLE_HEARTBEAT_V1 = "heartbeat-v1"
+LIFECYCLE_LEGACY = "legacy"
+
+_ENV_PREFIX = "GITEA_LEASE_POLICY"
+
+
+@dataclass(frozen=True)
+class LeasePolicy:
+    """Durations governing one task class.
+
+    All intervals are minutes except ``absolute_cap_hours``. ``None`` for the
+    cap means the class has no maximum continuous duration.
+    """
+
+    task_class: str
+    initial_ttl_minutes: float
+    heartbeat_cadence_minutes: float
+    stale_warning_minutes: float
+    missed_heartbeat_grace_minutes: float
+    absolute_cap_hours: float | None
+    recovery_grace_minutes: float
+    terminal_race_drain_minutes: float
+    terminal_retirement_eligible: bool
+    heartbeat_lifecycle_active: bool
+
+
+# Defaults. ``author_issue_work`` adopts the reviewer window proven by #747
+# rather than inventing new numbers: a lease expires 10 minutes after its last
+# valid heartbeat, warns at half that, and an actively heartbeating session is
+# never evicted. The prior value was a fixed four hours (240 minutes) that no
+# heartbeat could shorten — the defect this issue exists to correct.
+_DEFAULTS: dict[str, LeasePolicy] = {
+    TASK_CLASS_AUTHOR_ISSUE_WORK: LeasePolicy(
+        task_class=TASK_CLASS_AUTHOR_ISSUE_WORK,
+        initial_ttl_minutes=10.0,
+        heartbeat_cadence_minutes=2.0,
+        stale_warning_minutes=5.0,
+        missed_heartbeat_grace_minutes=10.0,
+        absolute_cap_hours=8.0,
+        recovery_grace_minutes=10.0,
+        terminal_race_drain_minutes=2.0,
+        terminal_retirement_eligible=True,
+        heartbeat_lifecycle_active=True,
+    ),
+    # Declared, not rewired. These mirror reviewer_pr_lease.LEASE_TTL_MINUTES
+    # and STALE_WARNING_MINUTES; Slice C migrates the call sites.
+    TASK_CLASS_REVIEWER_PR: LeasePolicy(
+        task_class=TASK_CLASS_REVIEWER_PR,
+        initial_ttl_minutes=10.0,
+        heartbeat_cadence_minutes=2.0,
+        stale_warning_minutes=5.0,
+        missed_heartbeat_grace_minutes=10.0,
+        absolute_cap_hours=None,
+        recovery_grace_minutes=10.0,
+        terminal_race_drain_minutes=2.0,
+        terminal_retirement_eligible=False,
+        heartbeat_lifecycle_active=False,
+    ),
+    TASK_CLASS_MERGER_PR: LeasePolicy(
+        task_class=TASK_CLASS_MERGER_PR,
+        initial_ttl_minutes=10.0,
+        heartbeat_cadence_minutes=2.0,
+        stale_warning_minutes=5.0,
+        missed_heartbeat_grace_minutes=10.0,
+        absolute_cap_hours=None,
+        recovery_grace_minutes=10.0,
+        terminal_race_drain_minutes=2.0,
+        terminal_retirement_eligible=False,
+        heartbeat_lifecycle_active=False,
+    ),
+    # Mirrors pr_work_lease.DEFAULT_CONFLICT_FIX_TTL_MINUTES. Deliberately left
+    # at its current window; shortening it is Slice C's call, not this slice's.
+    TASK_CLASS_CONFLICT_FIX: LeasePolicy(
+        task_class=TASK_CLASS_CONFLICT_FIX,
+        initial_ttl_minutes=120.0,
+        heartbeat_cadence_minutes=2.0,
+        stale_warning_minutes=5.0,
+        missed_heartbeat_grace_minutes=10.0,
+        absolute_cap_hours=None,
+        recovery_grace_minutes=10.0,
+        terminal_race_drain_minutes=2.0,
+        terminal_retirement_eligible=False,
+        heartbeat_lifecycle_active=False,
+    ),
+}
+
+_NUMERIC_FIELDS = (
+    "initial_ttl_minutes",
+    "heartbeat_cadence_minutes",
+    "stale_warning_minutes",
+    "missed_heartbeat_grace_minutes",
+    "absolute_cap_hours",
+    "recovery_grace_minutes",
+    "terminal_race_drain_minutes",
+)
+
+
+def env_var_name(task_class: str, field: str) -> str:
+    """Environment variable that overrides one field of one task class."""
+    return f"{_ENV_PREFIX}_{task_class.upper()}_{field.upper()}"
+
+
+def _override(task_class: str, field: str, default: float | None) -> float | None:
+    """Read one override, falling back to *default* on anything unusable.
+
+    A malformed or non-positive override is ignored rather than raised: a typo
+    in an environment variable must not be able to mint a zero-length lease that
+    makes every claim instantly reclaimable, nor crash the server at import.
+    """
+    raw = (os.environ.get(env_var_name(task_class, field)) or "").strip()
+    if not raw:
+        return default
+    try:
+        value = float(raw)
+    except (TypeError, ValueError):
+        return default
+    if value <= 0:
+        return default
+    return value
+
+
+def policy_for(task_class: str) -> LeasePolicy:
+    """Return the effective policy for *task_class*.
+
+    Unknown task classes fall back to the author policy, which is the most
+    conservative migrated class, rather than raising — a new caller must never
+    be able to crash a lock write by naming a class this table has not learned.
+    """
+    key = str(task_class or "").strip() or TASK_CLASS_AUTHOR_ISSUE_WORK
+    base = _DEFAULTS.get(key) or _DEFAULTS[TASK_CLASS_AUTHOR_ISSUE_WORK]
+    resolved = {
+        field: _override(base.task_class, field, getattr(base, field))
+        for field in _NUMERIC_FIELDS
+    }
+    if all(resolved[field] == getattr(base, field) for field in _NUMERIC_FIELDS):
+        return base
+    return LeasePolicy(
+        task_class=base.task_class,
+        terminal_retirement_eligible=base.terminal_retirement_eligible,
+        heartbeat_lifecycle_active=base.heartbeat_lifecycle_active,
+        **resolved,
+    )
+
+
+def known_task_classes() -> tuple[str, ...]:
+    """Every declared task class, migrated or not."""
+    return tuple(_DEFAULTS)
+
+
+def describe(task_class: str) -> dict[str, Any]:
+    """Serializable view of a policy, for audit records and tool payloads."""
+    policy = policy_for(task_class)
+    return {
+        "task_class": policy.task_class,
+        "initial_ttl_minutes": policy.initial_ttl_minutes,
+        "heartbeat_cadence_minutes": policy.heartbeat_cadence_minutes,
+        "stale_warning_minutes": policy.stale_warning_minutes,
+        "missed_heartbeat_grace_minutes": policy.missed_heartbeat_grace_minutes,
+        "absolute_cap_hours": policy.absolute_cap_hours,
+        "recovery_grace_minutes": policy.recovery_grace_minutes,
+        "terminal_race_drain_minutes": policy.terminal_race_drain_minutes,
+        "terminal_retirement_eligible": policy.terminal_retirement_eligible,
+        "heartbeat_lifecycle_active": policy.heartbeat_lifecycle_active,
+        "lifecycle_version": LIFECYCLE_HEARTBEAT_V1,
+    }

master_parity_gate.py

@@ -24,50 +24,12 @@ from __future__ import annotations
 
 import os
 import subprocess
-import time
-
-# Live-remote head cache: the parity gate runs on every mutation and every
-# runtime-context read, so the ``git ls-remote`` result is cached briefly to
-# avoid a network round-trip per call (#610). Keyed by (root, remote, branch).
-_REMOTE_HEAD_CACHE: dict[tuple[str, str, str], tuple[float, str | None]] = {}
-_REMOTE_HEAD_TTL = 60.0
-
-# When True, ``read_remote_master_head`` never performs ``git ls-remote`` unless
-# ``GITEA_TEST_LIVE_REMOTE_HEAD`` is set. Conftest enables this suite-wide so
-# feature worktrees (whose HEAD differs from live master) cannot flip legacy
-# runtime-context assertions to live_stale, and so unit tests never depend on
-# a live network (PR #788 F1/F2 / issue #610). Module-level (not env-only) so
-# ``patch.dict(os.environ, …, clear=True)`` cannot re-enable the probe.
-_HERMETIC_TEST_MODE: bool = False
-
-
-def _clear_remote_head_cache() -> None:
-    """Reset the live-remote head cache (test isolation / forced refresh)."""
-    _REMOTE_HEAD_CACHE.clear()
-
-
-def set_hermetic_test_mode(enabled: bool) -> None:
-    """Enable or disable suite-wide hermetic live-remote reads (tests only)."""
-    global _HERMETIC_TEST_MODE
-    _HERMETIC_TEST_MODE = bool(enabled)
-    _clear_remote_head_cache()
-
-
-def hermetic_test_mode() -> bool:
-    """Return whether hermetic live-remote reads are active."""
-    return bool(_HERMETIC_TEST_MODE)
-
 
 # Environment escape hatches (ops + tests):
 #   GITEA_MCP_DISABLE_PARITY_GATE  -> disable enforcement entirely (fail open).
 #   GITEA_TEST_CURRENT_HEAD        -> force the "current" HEAD read, for tests.
 ENV_DISABLE = "GITEA_MCP_DISABLE_PARITY_GATE"
 ENV_TEST_CURRENT_HEAD = "GITEA_TEST_CURRENT_HEAD"
-# GITEA_TEST_LIVE_REMOTE_HEAD -> force the live remote master read, for tests.
-ENV_TEST_LIVE_REMOTE_HEAD = "GITEA_TEST_LIVE_REMOTE_HEAD"
-# GITEA_TEST_ALLOW_LIVE_REMOTE_PROBE -> opt a single test into a real ls-remote
-# even when hermetic mode is on (rare; prefer ENV_TEST_LIVE_REMOTE_HEAD).
-ENV_TEST_ALLOW_LIVE_REMOTE_PROBE = "GITEA_TEST_ALLOW_LIVE_REMOTE_PROBE"
 
 
 def read_git_head(root: str) -> str | None:
@@ -96,75 +58,6 @@ def read_git_head(root: str) -> str | None:
     return (res.stdout or "").strip() or None
 
 
-def read_remote_master_head(
-    root: str,
-    remote: str = "origin",
-    branch: str = "master",
-    ttl: float = _REMOTE_HEAD_TTL,
-) -> str | None:
-    """Return the live remote ``branch`` commit SHA, or ``None`` (#610).
-
-    Resolves the *live* target commit via ``git ls-remote`` so parity can tell
-    a daemon that is behind the live remote master apart from one whose local
-    checkout simply hasn't been pulled. ``None`` means the live head could not
-    be resolved (offline, no such remote, git unavailable, error) -- callers
-    must treat unknown live state as *not mutation-safe* while never blocking
-    read-only diagnostics. A ``GITEA_TEST_LIVE_REMOTE_HEAD`` override takes
-    precedence so the wiring can be exercised deterministically and offline.
-
-    The result is cached for *ttl* seconds per (root, remote, branch) so the
-    gate does not run a network probe on every mutation/read (``ttl=0`` forces
-    a live probe). Both hits and ``None`` misses are cached to bound offline
-    latency; the env override bypasses the cache and the subprocess entirely.
-
-    Under suite hermetic mode (``set_hermetic_test_mode(True)``, set by
-    conftest) a missing override returns ``None`` without network I/O so
-    feature-worktree test runs cannot observe live_stale against real master
-    (PR #788 F1) and unit tests stay offline (F2). Opt out with an explicit
-    ``GITEA_TEST_LIVE_REMOTE_HEAD`` pin or ``GITEA_TEST_ALLOW_LIVE_REMOTE_PROBE``.
-    """
-    forced = os.environ.get(ENV_TEST_LIVE_REMOTE_HEAD)
-    if forced is not None:
-        return forced.strip() or None
-    if _HERMETIC_TEST_MODE and not (
-        os.environ.get(ENV_TEST_ALLOW_LIVE_REMOTE_PROBE) or ""
-    ).strip():
-        # Hermetic default: live head unknown. live_stale stays False;
-        # mutation_safe is False when live is unknown (documented #610 note).
-        return None
-    # Defense in depth: even without the module flag, never probe while pytest
-    # is running unless the test opted into a real probe or set an override.
-    if (os.environ.get("PYTEST_CURRENT_TEST") or "").strip() and not (
-        os.environ.get(ENV_TEST_ALLOW_LIVE_REMOTE_PROBE) or ""
-    ).strip():
-        return None
-    if not root:
-        return None
-    key = (root, remote, branch)
-    now = time.monotonic()
-    if ttl > 0:
-        cached = _REMOTE_HEAD_CACHE.get(key)
-        if cached is not None and (now - cached[0]) < ttl:
-            return cached[1]
-    sha: str | None = None
-    try:
-        res = subprocess.run(
-            ["git", "-C", root, "ls-remote", remote, f"refs/heads/{branch}"],
-            capture_output=True,
-            text=True,
-            check=False,
-            timeout=5,
-        )
-        if res.returncode == 0:
-            lines = (res.stdout or "").strip().splitlines()
-            if lines:
-                sha = lines[0].split("\t", 1)[0].split()[0].strip() or None
-    except Exception:
-        sha = None
-    _REMOTE_HEAD_CACHE[key] = (now, sha)
-    return sha
-
-
 def capture_startup_parity(root: str, head: str | None = None) -> dict:
     """Capture the process source-tree baseline once at server startup.
 
@@ -179,38 +72,18 @@ def _short(sha: str | None) -> str:
     return sha[:12] if sha else "unknown"
 
 
-def assess_master_parity(
-    startup: dict | None,
-    current_head: str | None,
-    live_remote_head: str | None = None,
-) -> dict:
+def assess_master_parity(startup: dict | None, current_head: str | None) -> dict:
     """Compare the startup baseline against the current on-disk ``HEAD``.
 
-    Pure: all HEADs are supplied by the caller. Returns a structured result:
+    Pure: both HEADs are supplied by the caller. Returns a structured result:
 
     - ``in_parity``    -- server code matches the on-disk master (or parity
       could not be determined, which is not treated as stale).
     - ``stale``        -- the on-disk master has definitively advanced past the
       running process.
-    - ``restart_required`` -- ``stale`` or ``live_stale``; the recovery action.
-    - ``determinable`` -- whether both local HEADs were known well enough to
-      compare.
+    - ``restart_required`` -- alias of ``stale``; the recovery action.
+    - ``determinable`` -- whether both HEADs were known well enough to compare.
     - ``startup_head`` / ``current_head`` / ``reasons``.
-
-    #610 adds live-remote awareness so a daemon that is stale relative to the
-    *live* remote master cannot report a mutation-safe result even when the
-    local checkout HEAD still matches the daemon's startup commit:
-
-    - ``daemon_start_head`` -- the commit the running process started at
-      (alias of ``startup_head``, named for clarity in reports).
-    - ``local_head`` -- the on-disk checkout HEAD (alias of ``current_head``).
-    - ``live_remote_head`` -- the live remote target commit, or ``None`` when it
-      could not be fetched.
-    - ``live_known`` -- whether the live remote target was resolved.
-    - ``live_stale`` -- the live remote master has advanced past the running
-      process (daemon is behind live master) even if local parity is green.
-    - ``mutation_safe`` -- the daemon code, local checkout, and live remote
-      target all agree; the only state in which a mutation may rely on parity.
     """
     startup_head = (startup or {}).get("startup_head")
     reasons: list[str] = []
@@ -218,56 +91,32 @@ def assess_master_parity(
     if startup_head is None:
         reasons.append(
             "startup commit was not captured; code parity cannot be enforced")
-        return _result(True, False, False, startup_head, current_head,
-                       live_remote_head, False, reasons)
+        return _result(True, False, False, startup_head, current_head, reasons)
 
     if current_head is None:
         reasons.append(
             "current workspace HEAD could not be read; code parity cannot be "
             "enforced")
-        return _result(True, False, False, startup_head, current_head,
-                       live_remote_head, False, reasons)
+        return _result(True, False, False, startup_head, current_head, reasons)
 
-    local_in_parity = startup_head == current_head
-    local_stale = not local_in_parity
-    if local_stale:
-        reasons.append(
-            f"MCP server started at commit {_short(startup_head)} but the "
-            f"workspace master is now {_short(current_head)}; restart the "
-            f"server to load the current capability gates")
+    if startup_head == current_head:
+        return _result(True, False, True, startup_head, current_head, reasons)
 
-    live_known = live_remote_head is not None
-    live_stale = live_known and live_remote_head != startup_head
-    if live_stale:
-        reasons.append(
-            f"live remote master is {_short(live_remote_head)} but the MCP "
-            f"server started at {_short(startup_head)}; the daemon is stale "
-            f"relative to live master -- restart/reconnect before mutating")
-
-    return _result(
-        local_in_parity, local_stale, True, startup_head, current_head,
-        live_remote_head, live_stale, reasons)
+    reasons.append(
+        f"MCP server started at commit {_short(startup_head)} but the workspace "
+        f"master is now {_short(current_head)}; restart the server to load the "
+        f"current capability gates")
+    return _result(False, True, True, startup_head, current_head, reasons)
 
 
-def _result(in_parity, stale, determinable, startup_head, current_head,
-            live_remote_head, live_stale, reasons):
-    live_known = live_remote_head is not None
-    mutation_safe = (
-        determinable and in_parity and live_known and not live_stale)
+def _result(in_parity, stale, determinable, startup_head, current_head, reasons):
     return {
         "in_parity": in_parity,
         "stale": stale,
-        "restart_required": stale or live_stale,
+        "restart_required": stale,
         "determinable": determinable,
         "startup_head": startup_head,
         "current_head": current_head,
-        # #610 distinguished signals:
-        "daemon_start_head": startup_head,
-        "local_head": current_head,
-        "live_remote_head": live_remote_head,
-        "live_known": live_known,
-        "live_stale": live_stale,
-        "mutation_safe": mutation_safe,
         "reasons": list(reasons),
     }
 
@@ -281,13 +130,11 @@ def parity_block_reasons(assessment: dict) -> list[str]:
     """Block reasons for a mutation gate (empty when the mutation may proceed).
 
     A disabled gate or an in-parity / non-determinable assessment yields no
-    reasons. A definitively stale server blocks, and (#610) a daemon that is
-    stale relative to the *live* remote master blocks even when the local
-    checkout HEAD still matches the daemon's startup commit.
+    reasons; only a definitively stale server blocks.
     """
     if gate_disabled():
         return []
-    if assessment.get("stale") or assessment.get("live_stale"):
+    if assessment.get("stale"):
         return list(assessment.get("reasons") or
                     ["server code is stale relative to master (fail closed)"])
     return []
@@ -300,10 +147,6 @@ def parity_report(assessment: dict) -> dict:
         "restart_required": True,
         "startup_head": assessment.get("startup_head"),
         "current_head": assessment.get("current_head"),
-        # #610: name the live remote target so the report distinguishes a
-        # local-code stale from a daemon-behind-live-master stale.
-        "live_remote_head": assessment.get("live_remote_head"),
-        "live_stale": bool(assessment.get("live_stale")),
         "reasons": list(assessment.get("reasons") or []),
         "recovery": [
             "The running MCP server is executing code older than the current "
@@ -314,45 +157,6 @@ def parity_report(assessment: dict) -> dict:
     }
 
 
-def parity_resolver_disagreement(
-    assessment: dict,
-    resolver_restart_required: bool,
-) -> dict | None:
-    """Typed blocker when the resolver requires restart but parity looks green.
-
-    The capability resolver (``gitea_resolve_task_capability``) detects stale
-    runtime authoritatively for mutation safety (#610). When it requires a
-    restart, local-only parity must never override it: this returns a typed,
-    fail-closed blocker that names the resolver as authoritative. Returns
-    ``None`` when the resolver does not require a restart.
-    """
-    if not resolver_restart_required:
-        return None
-    parity_optimistic = bool(assessment.get("in_parity")) and not (
-        assessment.get("stale") or assessment.get("live_stale"))
-    return {
-        "kind": "parity_resolver_disagreement",
-        "restart_required": True,
-        "resolver_authoritative": True,
-        "parity_optimistic": parity_optimistic,
-        "daemon_start_head": assessment.get("daemon_start_head"),
-        "local_head": assessment.get("local_head"),
-        "live_remote_head": assessment.get("live_remote_head"),
-        "reasons": [
-            "The capability resolver requires a restart/reconnect (stale "
-            "runtime) but master-parity reported local code as in-parity. "
-            "The resolver is authoritative for mutation safety; do not mutate "
-            "on local parity alone. Restart/reconnect the Gitea MCP server "
-            "and re-verify before mutating.",
-        ],
-        "recovery": [
-            "Trust the resolver: treat this session as stale.",
-            "Restart or /mcp reconnect the Gitea MCP namespace so it reloads "
-            "current master and live target state, then re-run preflight.",
-        ],
-    }
-
-
 def format_parity(assessment: dict) -> str:
     """One-line human summary for logs / runtime context."""
     if assessment.get("stale"):

task_capability_map.py

@@ -32,6 +32,15 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = {
         "permission": "gitea.issue.comment",
         "role": "author",
     },
+    # #790 Slice A: prove an owned author lease is still active. Strictly
+    # narrower than lock_issue — it can only slide a lease this exact session
+    # already owns, never acquire, take over, or revive one — so it gates on the
+    # same authority rather than introducing an operation name that every
+    # already-configured author profile would be missing.
+    "heartbeat_issue_lock": {
+        "permission": "gitea.issue.comment",
+        "role": "author",
+    },
     "set_issue_labels": {
         "permission": "gitea.issue.comment",
         "role": "author",

tests/conftest.py

@@ -167,35 +167,6 @@ def _reset_mutation_authority(monkeypatch):
 import pytest
 
 
-@pytest.fixture(autouse=True)
-def _hermetic_live_remote_master_head():
-    """#610 / PR #788 F1/F2: keep live-remote parity reads offline in tests.
-
-    ``read_remote_master_head`` would otherwise ``git ls-remote`` whenever
-    ``GITEA_TEST_LIVE_REMOTE_HEAD`` is unset. Feature worktrees under
-    ``branches/`` always differ from live master, so legacy suites that assert
-    runtime-context ``safe_next_action`` flip to live_stale. Module-level
-    hermetic mode survives ``patch.dict(os.environ, …, clear=True)``.
-    Tests that exercise the real probe path call
-    ``master_parity_gate.set_hermetic_test_mode(False)`` and/or set
-    ``GITEA_TEST_ALLOW_LIVE_REMOTE_PROBE``.
-    """
-    try:
-        import master_parity_gate as _mpg
-
-        _mpg.set_hermetic_test_mode(True)
-    except Exception:
-        _mpg = None
-    try:
-        yield
-    finally:
-        if _mpg is not None:
-            try:
-                _mpg.set_hermetic_test_mode(False)
-            except Exception:
-                pass
-
-
 @pytest.fixture(autouse=True)
 def _deterministic_workspace_remotes():
     try:

tests/test_issue_790_heartbeat_mcp_path.py

@@ -0,0 +1,444 @@
+"""Task heartbeat through the native MCP author path (#790 Slice A, AC-N6).
+
+Assessor-level coverage is not sufficient here, and this project has already
+paid for learning that: in review #499 on PR #791 the #760 renewal waiver was
+computed correctly and then *discarded* at two later gates, so every real
+renewal still failed while the unit suite stayed green. AC-N6 exists because of
+that, and requires driving the real tools against a real git repository and a
+real durable lock file, composing the gates in production order.
+
+These tests therefore call ``gitea_lock_issue`` and
+``gitea_heartbeat_issue_lock`` themselves and assert on what lands on disk,
+never on an assessor's return value alone.
+"""
+
+from __future__ import annotations
+
+import os
+import subprocess
+import sys
+import tempfile
+import unittest
+from datetime import datetime, timedelta, timezone
+from unittest.mock import patch
+
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+
+from mutation_profile_fixture import shared_mutation_env  # noqa: E402
+
+import issue_lock_provenance  # noqa: E402
+import issue_lock_store  # noqa: E402
+import lease_policy  # noqa: E402
+import mcp_server  # noqa: E402
+
+ISSUE = 9791
+BRANCH = f"fix/issue-{ISSUE}-heartbeat-mcp"
+IDENTITY = "example-user"
+PROFILE = "test-author-prgs"
+ORG = "Scaled-Tech-Consulting"
+REPO = "Gitea-Tools"
+
+
+def _ts(moment: datetime) -> str:
+    return (
+        moment.astimezone(timezone.utc)
+        .replace(microsecond=0)
+        .isoformat()
+        .replace("+00:00", "Z")
+    )
+
+
+class _HeartbeatMcpBase(unittest.TestCase):
+    """Real git repo plus a real durable lock, driven through the real tools."""
+
+    def setUp(self):
+        self.lock_dir = tempfile.TemporaryDirectory()
+        self.addCleanup(self.lock_dir.cleanup)
+        self.repo = tempfile.mkdtemp(prefix="issue790-mcp-")
+        self.addCleanup(lambda: subprocess.run(["rm", "-rf", self.repo], check=False))
+        self._init_worktree()
+        self.remotes = patch.dict(
+            mcp_server.REMOTES,
+            {"prgs": {"host": "gitea.prgs.cc", "org": ORG, "repo": REPO}},
+        )
+        self.remotes.start()
+        self.addCleanup(patch.stopall)
+        mcp_server._IDENTITY_CACHE.clear()
+
+    def _git(self, *args):
+        return subprocess.run(
+            ["git", "-C", self.repo, *args], capture_output=True, text=True, check=True
+        )
+
+    def _init_worktree(self):
+        self._git("init", "-q", "-b", "master")
+        self._git("config", "user.email", "[email protected]")
+        self._git("config", "user.name", "Test")
+        with open(os.path.join(self.repo, "seed.txt"), "w") as fh:
+            fh.write("seed\n")
+        self._git("add", "seed.txt")
+        self._git("commit", "-q", "-m", "seed")
+        self.base_sha = self._git("rev-parse", "HEAD").stdout.strip()
+        # A fresh claim starts base-equivalent, which is the ordinary first-lock
+        # shape and exercises assess_issue_lock_worktree on its normal path.
+        self._git("checkout", "-q", "-b", BRANCH)
+        self.head_sha = self.base_sha
+        self.worktree = os.path.realpath(self.repo)
+
+    def _lock_path(self):
+        return issue_lock_store.lock_file_path(
+            remote="prgs",
+            org=ORG,
+            repo=REPO,
+            issue_number=ISSUE,
+            lock_dir=self.lock_dir.name,
+        )
+
+    def _tool_env(self):
+        env = shared_mutation_env(
+            PROFILE, include_example_repo=True, GITEA_ISSUE_LOCK_DIR=self.lock_dir.name
+        )
+        env["GITEA_ISSUE_LOCK_DIR"] = self.lock_dir.name
+        return env
+
+    def _git_state(self, *, porcelain="", base_equivalent=True):
+        return {
+            "current_branch": BRANCH,
+            "porcelain_status": porcelain,
+            "base_equivalent": base_equivalent,
+            "head_sha": self.head_sha,
+            "inspected_git_root": self.worktree,
+            "base_branch": "master",
+        }
+
+    def run_lock_issue(
+        self,
+        *,
+        branch_entries=None,
+        open_prs=None,
+        git_state=None,
+        identity=IDENTITY,
+        profile=PROFILE,
+    ):
+        branch_entries = branch_entries if branch_entries is not None else []
+        open_prs = open_prs if open_prs is not None else []
+        git_state = git_state or self._git_state()
+        env = self._tool_env()
+        with patch(
+            "mcp_server.api_get_all", return_value=list(branch_entries)
+        ), patch(
+            "mcp_server._list_open_pulls", return_value=list(open_prs)
+        ), patch(
+            "mcp_server.get_auth_header", return_value="token x"
+        ), patch(
+            "mcp_server._work_lease_claimant",
+            return_value={"username": identity, "profile": profile},
+        ), patch(
+            "mcp_server.issue_lock_worktree.read_worktree_git_state",
+            return_value=git_state,
+        ), patch(
+            "mcp_server.issue_duplicate_context_fetcher",
+            side_effect=lambda h, o, r, auth, issue_number: (
+                list(open_prs),
+                [b.get("name") for b in branch_entries if isinstance(b, dict)],
+                {"status": "not_claimed"},
+            ),
+        ), patch.dict(os.environ, env, clear=True):
+            os.environ["GITEA_ISSUE_LOCK_DIR"] = self.lock_dir.name
+            return mcp_server.gitea_lock_issue(
+                issue_number=ISSUE,
+                branch_name=BRANCH,
+                remote="prgs",
+                worktree_path=self.worktree,
+            )
+
+    def run_heartbeat(
+        self, *, task_session_id, identity=IDENTITY, profile=PROFILE, **kwargs
+    ):
+        env = self._tool_env()
+        with patch(
+            "mcp_server._work_lease_claimant",
+            return_value={"username": identity, "profile": profile},
+        ), patch("mcp_server.get_auth_header", return_value="token x"), patch.dict(
+            os.environ, env, clear=True
+        ):
+            os.environ["GITEA_ISSUE_LOCK_DIR"] = self.lock_dir.name
+            return mcp_server.gitea_heartbeat_issue_lock(
+                issue_number=ISSUE,
+                branch_name=kwargs.pop("branch_name", BRANCH),
+                task_session_id=task_session_id,
+                remote="prgs",
+                worktree_path=kwargs.pop("worktree_path", self.worktree),
+                **kwargs,
+            )
+
+    def write_legacy_lock(self, *, hours_old: float = 3.0, ttl_hours: float = 4.0):
+        """A durable lock in the shape the store wrote before this slice."""
+        now = datetime.now(timezone.utc)
+        claimant = {"username": IDENTITY, "profile": PROFILE}
+        created = now - timedelta(hours=hours_old)
+        record = {
+            "issue_number": ISSUE,
+            "branch_name": BRANCH,
+            "remote": "prgs",
+            "org": ORG,
+            "repo": REPO,
+            "worktree_path": self.worktree,
+            "session_pid": os.getpid(),
+            "pid": os.getpid(),
+            "lock_generation": 1,
+            "work_lease": {
+                "operation_type": issue_lock_store.AUTHOR_ISSUE_WORK_LEASE,
+                "issue_number": ISSUE,
+                "pr_number": None,
+                "branch": BRANCH,
+                "worktree_path": self.worktree,
+                "claimant": claimant,
+                "created_at": _ts(created),
+                # The legacy signature: never advanced past creation.
+                "last_heartbeat_at": _ts(created),
+                "expires_at": _ts(created + timedelta(hours=ttl_hours)),
+            },
+            "lock_provenance": issue_lock_provenance.build_sanctioned_lock_provenance(
+                tool="gitea_lock_issue", claimant=claimant
+            ),
+        }
+        path = self._lock_path()
+        record["lock_file_path"] = path
+        issue_lock_store.save_lock_file(path, record)
+        return record
+
+
+class TestLockIssueMintsTheLifecycle(_HeartbeatMcpBase):
+    """Durable lock creation and read-back through the real tool."""
+
+    def test_native_lock_writes_the_marker_and_a_task_session_id(self):
+        result = self.run_lock_issue()
+        self.assertTrue(result["success"], result)
+
+        written = issue_lock_store.read_lock_file(result["lock_file_path"])
+        lease = written["work_lease"]
+        self.assertEqual(
+            lease["lifecycle_version"], lease_policy.LIFECYCLE_HEARTBEAT_V1
+        )
+        self.assertTrue(lease["task_session_id"])
+        self.assertFalse(issue_lock_store.is_legacy_lease(written))
+        # AC-N1: the ownership key is not the daemon pid, which is recorded
+        # separately as evidence.
+        self.assertNotIn(str(written["session_pid"]), lease["task_session_id"])
+        self.assertEqual(written["session_pid"], os.getpid())
+
+    def test_native_lease_uses_the_policy_window_not_four_hours(self):
+        result = self.run_lock_issue()
+        lease = result["work_lease"]
+        created = datetime.fromisoformat(lease["created_at"].replace("Z", "+00:00"))
+        expires = datetime.fromisoformat(lease["expires_at"].replace("Z", "+00:00"))
+        policy = lease_policy.policy_for(lease_policy.TASK_CLASS_AUTHOR_ISSUE_WORK)
+        self.assertEqual(
+            (expires - created).total_seconds() / 60.0, policy.initial_ttl_minutes
+        )
+
+    def test_freshness_of_a_new_native_lock_is_live(self):
+        result = self.run_lock_issue()
+        self.assertEqual(
+            result["lock_freshness"]["status"], issue_lock_store.STATUS_LIVE
+        )
+        self.assertTrue(result["lock_freshness"]["live"])
+
+
+class TestHeartbeatThroughTheTool(_HeartbeatMcpBase):
+    def _lock_and_session(self):
+        result = self.run_lock_issue()
+        self.assertTrue(result["success"], result)
+        return result, result["work_lease"]["task_session_id"]
+
+    def test_heartbeat_slides_the_lease_and_advances_the_generation(self):
+        locked, session = self._lock_and_session()
+        before = issue_lock_store.read_lock_file(locked["lock_file_path"])
+
+        beat = self.run_heartbeat(task_session_id=session)
+
+        self.assertTrue(beat["success"], beat)
+        self.assertEqual(beat["operation"], "heartbeat")
+        after = issue_lock_store.read_lock_file(locked["lock_file_path"])
+        self.assertGreater(
+            issue_lock_store.lock_generation(after),
+            issue_lock_store.lock_generation(before),
+        )
+        self.assertGreaterEqual(
+            after["work_lease"]["expires_at"], before["work_lease"]["expires_at"]
+        )
+        self.assertEqual(after["work_lease"]["heartbeat_count"], 2)
+
+    def test_heartbeat_evidence_survives_the_downstream_mutation_gate(self):
+        """The #499 F2 lesson, applied.
+
+        A sanction that is computed and then discarded downstream is worthless.
+        After a heartbeat the lock must still satisfy the gate every author
+        mutation runs through.
+        """
+        locked, session = self._lock_and_session()
+        self.run_heartbeat(task_session_id=session)
+
+        written = issue_lock_store.read_lock_file(locked["lock_file_path"])
+        verdict = issue_lock_store.verify_lock_for_mutation(
+            written,
+            issue_number=ISSUE,
+            branch_name=BRANCH,
+            worktree_path=self.worktree,
+        )
+        self.assertTrue(verdict["proven"], verdict)
+        self.assertFalse(verdict["block"])
+
+    def _duplicate_gate(self, *, open_prs, branches):
+        env = self._tool_env()
+        with patch("mcp_server.get_auth_header", return_value="token x"), patch(
+            "mcp_server.issue_duplicate_context_fetcher",
+            side_effect=lambda h, o, r, auth, issue_number: (
+                list(open_prs),
+                list(branches),
+                {"status": "not_claimed"},
+            ),
+        ), patch.dict(os.environ, env, clear=True):
+            os.environ["GITEA_ISSUE_LOCK_DIR"] = self.lock_dir.name
+            return mcp_server.gitea_assess_work_issue_duplicate(
+                issue_number=ISSUE, branch_name=BRANCH, remote="prgs"
+            )
+
+    def test_heartbeat_does_not_change_the_duplicate_gate_verdict(self):
+        """The gate must be invariant under heartbeating.
+
+        The point is not that the gate passes — with a linked open PR at the
+        lock phase it correctly blocks (#400), heartbeat or not. The property
+        that matters is that sliding a lease neither loosens the gate nor
+        corrupts the lock state it reads: the verdict before and after a
+        heartbeat must be identical, for both the clear and the blocking shape.
+        """
+        _, session = self._lock_and_session()
+        linked = [{"number": 4242, "head": {"ref": BRANCH, "sha": self.head_sha}}]
+
+        clear_before = self._duplicate_gate(open_prs=[], branches=[])
+        blocked_before = self._duplicate_gate(open_prs=linked, branches=[BRANCH])
+
+        self.assertTrue(self.run_heartbeat(task_session_id=session)["success"])
+
+        clear_after = self._duplicate_gate(open_prs=[], branches=[])
+        blocked_after = self._duplicate_gate(open_prs=linked, branches=[BRANCH])
+
+        self.assertEqual(clear_before["outcome"], clear_after["outcome"])
+        self.assertFalse(clear_after["block"])
+        self.assertEqual(blocked_before["outcome"], blocked_after["outcome"])
+        self.assertTrue(blocked_after["block"])
+        self.assertEqual(blocked_after["linked_open_pr"], 4242)
+
+    def test_foreign_session_id_is_refused_through_the_tool(self):
+        self._lock_and_session()
+        beat = self.run_heartbeat(task_session_id="author_issue_work-ffffffffffffffff")
+        self.assertFalse(beat["success"])
+        self.assertIn("task_session_id does not match", " ".join(beat["reasons"]))
+
+    def test_stale_generation_is_refused_through_the_tool(self):
+        locked, session = self._lock_and_session()
+        current = issue_lock_store.lock_generation(
+            issue_lock_store.read_lock_file(locked["lock_file_path"])
+        )
+        beat = self.run_heartbeat(
+            task_session_id=session, expected_generation=current + 5
+        )
+        self.assertFalse(beat["success"])
+        self.assertIn("generation changed", beat["reasons"][0])
+
+    def test_foreign_claimant_is_refused_through_the_tool(self):
+        _, session = self._lock_and_session()
+        beat = self.run_heartbeat(task_session_id=session, identity="someone-else")
+        self.assertFalse(beat["success"])
+
+    def test_heartbeat_cannot_acquire_a_missing_lock(self):
+        beat = self.run_heartbeat(task_session_id="author_issue_work-000000000000")
+        self.assertFalse(beat["success"])
+        self.assertIn("no durable lock", beat["reasons"][0])
+
+    def test_alive_pid_alone_does_not_keep_a_lease_live_through_the_tool(self):
+        """PID-only refusal, end to end.
+
+        The recorded pid is this live process. The lock is aged past its grace
+        with no heartbeat, so the tool must refuse to slide it and the durable
+        record must classify as a missed heartbeat rather than as live.
+        """
+        locked, session = self._lock_and_session()
+        record = issue_lock_store.read_lock_file(locked["lock_file_path"])
+        record["work_lease"]["last_heartbeat_at"] = _ts(
+            datetime.now(timezone.utc) - timedelta(minutes=30)
+        )
+        record["work_lease"]["expires_at"] = _ts(
+            datetime.now(timezone.utc) + timedelta(hours=2)
+        )
+        issue_lock_store.save_lock_file(locked["lock_file_path"], record)
+
+        self.assertTrue(issue_lock_store.is_process_alive(record["session_pid"]))
+        fresh = issue_lock_store.assess_lock_freshness(record)
+        self.assertEqual(
+            fresh["status"], issue_lock_store.STATUS_STALE_MISSED_HEARTBEAT
+        )
+        self.assertTrue(fresh["pid_alive"])
+
+        beat = self.run_heartbeat(task_session_id=session)
+        self.assertFalse(beat["success"])
+        self.assertIn("reclaimed", " ".join(beat["reasons"]))
+
+
+class TestLegacyLocksThroughTheTool(_HeartbeatMcpBase):
+    """AC-N8 end to end: protected on deployment, and rebindable."""
+
+    def test_legacy_lock_stays_protected_after_deployment(self):
+        record = self.write_legacy_lock(hours_old=3.0, ttl_hours=4.0)
+        fresh = issue_lock_store.assess_lock_freshness(record)
+        self.assertEqual(fresh["status"], issue_lock_store.STATUS_LIVE)
+        self.assertTrue(fresh["legacy_lease"])
+        self.assertTrue(fresh["legacy_expiry_preserved"])
+        # It had never heartbeated, so under the new grace alone it would be
+        # long gone; the preserved absolute expiry is what protects it.
+        self.assertEqual(
+            record["work_lease"]["created_at"],
+            record["work_lease"]["last_heartbeat_at"],
+        )
+
+    def test_tool_rebinds_a_legacy_lock_and_mints_a_first_heartbeat(self):
+        self.write_legacy_lock(hours_old=3.0, ttl_hours=4.0)
+
+        result = self.run_heartbeat(task_session_id=None)
+
+        self.assertTrue(result["success"], result)
+        self.assertEqual(result["operation"], "legacy_rebind")
+        self.assertTrue(result["task_session_id"])
+
+        written = issue_lock_store.read_lock_file(self._lock_path())
+        lease = written["work_lease"]
+        self.assertEqual(
+            lease["lifecycle_version"], lease_policy.LIFECYCLE_HEARTBEAT_V1
+        )
+        self.assertEqual(lease["heartbeat_count"], 1)
+        self.assertNotEqual(
+            lease["created_at"],
+            written["legacy_rebind"]["legacy_origin"]["created_at"],
+        )
+        self.assertFalse(issue_lock_store.is_legacy_lease(written))
+
+    def test_rebound_lock_then_heartbeats_through_the_tool(self):
+        self.write_legacy_lock(hours_old=3.0, ttl_hours=4.0)
+        rebound = self.run_heartbeat(task_session_id=None)
+        beat = self.run_heartbeat(task_session_id=rebound["task_session_id"])
+        self.assertTrue(beat["success"], beat)
+        self.assertEqual(beat["operation"], "heartbeat")
+        self.assertEqual(beat["heartbeat_count"], 2)
+
+    def test_rebind_refuses_a_foreign_owner_through_the_tool(self):
+        self.write_legacy_lock(hours_old=3.0, ttl_hours=4.0)
+        result = self.run_heartbeat(task_session_id=None, identity="someone-else")
+        self.assertFalse(result["success"])
+        self.assertEqual(result["operation"], "legacy_rebind")
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_issue_790_lease_policy.py

@@ -0,0 +1,594 @@
+"""Central lease policy and load-bearing heartbeat freshness (#790 Slice A).
+
+Before this slice, ``issue_lock_store.assess_lock_freshness`` parsed
+``last_heartbeat_at`` and then never consulted it: liveness was decided by an
+absolute four-hour ``expires_at`` and by PID liveness. Because the recorded PID
+is the long-lived MCP daemon rather than the authoring task, an abandoned claim
+stayed "live" for the full four hours, and a claim whose work had already landed
+blocked reconciliation for just as long (Issue #787 / PR #789, and again Issue
+#760 / PR #791).
+
+These tests pin the corrected semantics, including the two asymmetries that are
+easy to lose in a refactor:
+
+* an **alive** PID must never make anything live (AC-N2), while
+* a **dead** PID must still mark a lease stale, because #753 dead-session
+  recovery keys on exactly that classification.
+
+Durable-state helpers here write real lock files through the real flock path;
+they are not mocks of the store.
+"""
+
+from __future__ import annotations
+
+import os
+import sys
+import tempfile
+import unittest
+from datetime import datetime, timedelta, timezone
+from unittest.mock import patch
+
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+import issue_lock_store  # noqa: E402
+import lease_policy  # noqa: E402
+import pr_work_lease  # noqa: E402
+import reviewer_pr_lease  # noqa: E402
+
+ISSUE = 9790
+BRANCH = f"fix/issue-{ISSUE}-heartbeat"
+IDENTITY = "example-user"
+PROFILE = "test-author-prgs"
+ORG = "Example-Org"
+REPO = "Example-Repo"
+REMOTE = "prgs"
+DEAD_PID = 2**22  # far above any live pid on a test host
+
+
+def _ts(moment: datetime) -> str:
+    return (
+        moment.astimezone(timezone.utc)
+        .replace(microsecond=0)
+        .isoformat()
+        .replace("+00:00", "Z")
+    )
+
+
+class _LockFixture(unittest.TestCase):
+    def setUp(self):
+        self.lock_dir = tempfile.TemporaryDirectory()
+        self.addCleanup(self.lock_dir.cleanup)
+        self.now = datetime.now(timezone.utc)
+        self.worktree = os.path.realpath(tempfile.mkdtemp(prefix="issue790-"))
+        self.addCleanup(patch.stopall)
+
+    def _path(self):
+        return issue_lock_store.lock_file_path(
+            remote=REMOTE,
+            org=ORG,
+            repo=REPO,
+            issue_number=ISSUE,
+            lock_dir=self.lock_dir.name,
+        )
+
+    def write_lock(
+        self,
+        *,
+        lifecycle: str | None = lease_policy.LIFECYCLE_HEARTBEAT_V1,
+        created_delta: timedelta = timedelta(minutes=1),
+        heartbeat_delta: timedelta = timedelta(minutes=1),
+        expires_delta: timedelta = timedelta(minutes=9),
+        pid: int | None = None,
+        task_session_id: str | None = "author_issue_work-aaaabbbbccccdddd",
+        generation: int = 1,
+        identity: str = IDENTITY,
+        profile: str = PROFILE,
+        branch: str = BRANCH,
+        worktree: str | None = None,
+    ) -> dict:
+        """Write a real durable lock and return the record.
+
+        Deltas are relative to ``self.now``; ``expires_delta`` is added, the
+        others subtracted, so "in the past" reads naturally at each call site.
+        """
+        lease: dict = {
+            "operation_type": issue_lock_store.AUTHOR_ISSUE_WORK_LEASE,
+            "issue_number": ISSUE,
+            "pr_number": None,
+            "branch": branch,
+            "worktree_path": worktree or self.worktree,
+            "claimant": {"username": identity, "profile": profile},
+            "created_at": _ts(self.now - created_delta),
+            "last_heartbeat_at": _ts(self.now - heartbeat_delta),
+            "expires_at": _ts(self.now + expires_delta),
+        }
+        if lifecycle is not None:
+            lease["lifecycle_version"] = lifecycle
+            if task_session_id is not None:
+                lease["task_session_id"] = task_session_id
+        pid_value = os.getpid() if pid is None else pid
+        record = {
+            "issue_number": ISSUE,
+            "branch_name": branch,
+            "remote": REMOTE,
+            "org": ORG,
+            "repo": REPO,
+            "worktree_path": worktree or self.worktree,
+            "session_pid": pid_value,
+            "pid": pid_value,
+            "lock_generation": generation,
+            "work_lease": lease,
+        }
+        path = self._path()
+        record["lock_file_path"] = path
+        issue_lock_store.save_lock_file(path, record)
+        return record
+
+
+class TestPolicyIsTheSingleSource(unittest.TestCase):
+    """AC-N7: one authoritative configuration source for every duration."""
+
+    def test_author_policy_carries_the_agreed_values(self):
+        policy = lease_policy.policy_for(lease_policy.TASK_CLASS_AUTHOR_ISSUE_WORK)
+        self.assertEqual(policy.initial_ttl_minutes, 10.0)
+        self.assertEqual(policy.heartbeat_cadence_minutes, 2.0)
+        self.assertEqual(policy.stale_warning_minutes, 5.0)
+        self.assertEqual(policy.missed_heartbeat_grace_minutes, 10.0)
+        self.assertEqual(policy.absolute_cap_hours, 8.0)
+        self.assertEqual(policy.recovery_grace_minutes, 10.0)
+        self.assertEqual(policy.terminal_race_drain_minutes, 2.0)
+        self.assertTrue(policy.terminal_retirement_eligible)
+        self.assertTrue(policy.heartbeat_lifecycle_active)
+
+    def test_the_four_hour_author_ttl_literal_is_gone(self):
+        """The duplicated literal AC-N7 exists to remove."""
+        self.assertFalse(hasattr(issue_lock_store, "WORK_LEASE_TTL_HOURS"))
+        import gitea_mcp_server
+
+        self.assertFalse(hasattr(gitea_mcp_server, "WORK_LEASE_TTL_HOURS"))
+
+    def test_declared_reviewer_values_match_the_module_still_using_them(self):
+        """Slice A declares reviewer/merger numbers without rewiring them.
+
+        Recording a value in two places is only safe if drift is detectable, so
+        this asserts the declaration still equals the constants #747 owns. When
+        Slice C migrates those call sites, this test becomes the proof the
+        migration changed nothing.
+        """
+        policy = lease_policy.policy_for(lease_policy.TASK_CLASS_REVIEWER_PR)
+        self.assertEqual(
+            policy.initial_ttl_minutes, float(reviewer_pr_lease.LEASE_TTL_MINUTES)
+        )
+        self.assertEqual(
+            policy.stale_warning_minutes,
+            float(reviewer_pr_lease.STALE_WARNING_MINUTES),
+        )
+        self.assertFalse(policy.heartbeat_lifecycle_active)
+
+    def test_declared_conflict_fix_value_matches_its_module(self):
+        policy = lease_policy.policy_for(lease_policy.TASK_CLASS_CONFLICT_FIX)
+        self.assertEqual(
+            policy.initial_ttl_minutes,
+            float(pr_work_lease.DEFAULT_CONFLICT_FIX_TTL_MINUTES),
+        )
+        self.assertFalse(policy.heartbeat_lifecycle_active)
+
+    def test_environment_override_applies(self):
+        var = lease_policy.env_var_name(
+            lease_policy.TASK_CLASS_AUTHOR_ISSUE_WORK, "initial_ttl_minutes"
+        )
+        with patch.dict(os.environ, {var: "7"}):
+            self.assertEqual(
+                lease_policy.policy_for(
+                    lease_policy.TASK_CLASS_AUTHOR_ISSUE_WORK
+                ).initial_ttl_minutes,
+                7.0,
+            )
+
+    def test_unusable_override_falls_back_instead_of_minting_a_zero_lease(self):
+        """A typo must not make every claim instantly reclaimable."""
+        var = lease_policy.env_var_name(
+            lease_policy.TASK_CLASS_AUTHOR_ISSUE_WORK, "initial_ttl_minutes"
+        )
+        for bad in ("0", "-5", "not-a-number", "   "):
+            with self.subTest(value=bad), patch.dict(os.environ, {var: bad}):
+                self.assertEqual(
+                    lease_policy.policy_for(
+                        lease_policy.TASK_CLASS_AUTHOR_ISSUE_WORK
+                    ).initial_ttl_minutes,
+                    10.0,
+                )
+
+    def test_unknown_task_class_does_not_raise(self):
+        policy = lease_policy.policy_for("something-new")
+        self.assertEqual(policy.task_class, lease_policy.TASK_CLASS_AUTHOR_ISSUE_WORK)
+
+
+class TestLifecycleDiscrimination(_LockFixture):
+    """AC-N8: the marker, never a timestamp, decides legacy vs heartbeat."""
+
+    def test_missing_marker_reads_as_legacy(self):
+        record = self.write_lock(lifecycle=None)
+        self.assertTrue(issue_lock_store.is_legacy_lease(record))
+        self.assertEqual(
+            issue_lock_store.lease_lifecycle_version(record),
+            lease_policy.LIFECYCLE_LEGACY,
+        )
+
+    def test_marker_present_reads_as_heartbeat_lifecycle(self):
+        record = self.write_lock()
+        self.assertFalse(issue_lock_store.is_legacy_lease(record))
+
+    def test_equal_created_and_heartbeat_never_implies_a_fresh_heartbeat(self):
+        """The exact inversion AC-N8 forbids.
+
+        A legacy lock has ``last_heartbeat_at == created_at`` forever because
+        nothing ever advanced it. Reading that equality as "recently
+        heartbeated" would classify every never-heartbeated lock as fresh.
+        """
+        legacy = self.write_lock(
+            lifecycle=None,
+            created_delta=timedelta(hours=3),
+            heartbeat_delta=timedelta(hours=3),
+        )
+        lease = legacy["work_lease"]
+        self.assertEqual(lease["created_at"], lease["last_heartbeat_at"])
+        self.assertTrue(issue_lock_store.is_legacy_lease(legacy))
+
+        # A brand-new heartbeat lease has them equal too, so the equality
+        # carries no information in either direction.
+        fresh = self.write_lock(
+            created_delta=timedelta(seconds=0), heartbeat_delta=timedelta(seconds=0)
+        )
+        self.assertEqual(
+            fresh["work_lease"]["created_at"],
+            fresh["work_lease"]["last_heartbeat_at"],
+        )
+        self.assertFalse(issue_lock_store.is_legacy_lease(fresh))
+
+    def test_minted_session_id_contains_no_pid(self):
+        """AC-N1: the ownership key must not be derived from the daemon pid."""
+        minted = issue_lock_store.mint_task_session_id()
+        self.assertNotIn(str(os.getpid()), minted)
+        self.assertNotEqual(minted, issue_lock_store.mint_task_session_id())
+
+
+class TestFreshnessIsHeartbeatDriven(_LockFixture):
+    """AC-N2 and the new bands."""
+
+    def test_fresh_heartbeat_is_live(self):
+        record = self.write_lock(heartbeat_delta=timedelta(minutes=1))
+        fresh = issue_lock_store.assess_lock_freshness(record, now=self.now)
+        self.assertEqual(fresh["status"], issue_lock_store.STATUS_LIVE)
+        self.assertTrue(fresh["live"])
+        self.assertFalse(fresh["heartbeat_warning"])
+
+    def test_heartbeat_past_warning_is_still_live_but_flagged(self):
+        record = self.write_lock(heartbeat_delta=timedelta(minutes=6))
+        fresh = issue_lock_store.assess_lock_freshness(record, now=self.now)
+        self.assertEqual(fresh["status"], issue_lock_store.STATUS_LIVE)
+        self.assertTrue(fresh["heartbeat_warning"])
+
+    def test_missed_heartbeat_past_grace_is_classified_explicitly(self):
+        record = self.write_lock(
+            heartbeat_delta=timedelta(minutes=11),
+            expires_delta=timedelta(minutes=30),
+        )
+        fresh = issue_lock_store.assess_lock_freshness(record, now=self.now)
+        self.assertEqual(
+            fresh["status"], issue_lock_store.STATUS_STALE_MISSED_HEARTBEAT
+        )
+        self.assertFalse(fresh["live"])
+        self.assertTrue(fresh["stale"])
+
+    def test_alive_pid_never_establishes_freshness(self):
+        """The defect in one assertion.
+
+        The recorded PID is this very process, so it is unambiguously alive —
+        and the lease is still not live, because the task stopped heartbeating.
+        """
+        record = self.write_lock(
+            pid=os.getpid(),
+            heartbeat_delta=timedelta(hours=4),
+            expires_delta=timedelta(hours=4),
+        )
+        fresh = issue_lock_store.assess_lock_freshness(record, now=self.now)
+        self.assertTrue(fresh["pid_alive"])
+        self.assertFalse(fresh["live"])
+        self.assertEqual(
+            fresh["status"], issue_lock_store.STATUS_STALE_MISSED_HEARTBEAT
+        )
+
+    def test_dead_pid_still_marks_stale_for_issue_753(self):
+        """The opposite asymmetry: dead-PID corroboration is preserved."""
+        record = self.write_lock(pid=DEAD_PID, heartbeat_delta=timedelta(minutes=1))
+        fresh = issue_lock_store.assess_lock_freshness(record, now=self.now)
+        self.assertEqual(fresh["status"], issue_lock_store.STATUS_STALE)
+        self.assertFalse(fresh["live"])
+        self.assertIn("not alive", fresh["reason"])
+
+    def test_absolute_cap_requires_readoption(self):
+        record = self.write_lock(
+            created_delta=timedelta(hours=9), heartbeat_delta=timedelta(minutes=1)
+        )
+        fresh = issue_lock_store.assess_lock_freshness(record, now=self.now)
+        self.assertEqual(fresh["status"], issue_lock_store.STATUS_STALE_ABSOLUTE_CAP)
+        self.assertIn("re-adoption", fresh["reason"])
+
+    def test_heartbeat_lifecycle_without_a_heartbeat_fails_closed(self):
+        record = self.write_lock()
+        del record["work_lease"]["last_heartbeat_at"]
+        issue_lock_store.save_lock_file(self._path(), record)
+        fresh = issue_lock_store.assess_lock_freshness(record, now=self.now)
+        self.assertEqual(
+            fresh["status"], issue_lock_store.STATUS_STALE_MISSED_HEARTBEAT
+        )
+        self.assertIn("fail closed", fresh["reason"])
+
+    def test_absent_lock(self):
+        fresh = issue_lock_store.assess_lock_freshness(None)
+        self.assertEqual(fresh["status"], issue_lock_store.STATUS_ABSENT)
+        self.assertFalse(fresh["stale"])
+
+
+class TestLegacyLocksStayProtected(_LockFixture):
+    """AC-N8: deployment must not retroactively shorten an existing claim."""
+
+    def test_legacy_lock_with_a_stale_heartbeat_remains_live(self):
+        """The deployment-safety case.
+
+        A four-hour legacy lease minted three hours ago has not heartbeated
+        once. Under the new grace it would be long gone; under its preserved
+        absolute expiry it is still live, and must stay that way.
+        """
+        record = self.write_lock(
+            lifecycle=None,
+            created_delta=timedelta(hours=3),
+            heartbeat_delta=timedelta(hours=3),
+            expires_delta=timedelta(hours=1),
+        )
+        fresh = issue_lock_store.assess_lock_freshness(record, now=self.now)
+        self.assertEqual(fresh["status"], issue_lock_store.STATUS_LIVE)
+        self.assertTrue(fresh["live"])
+        self.assertTrue(fresh["legacy_lease"])
+        self.assertTrue(fresh["legacy_expiry_preserved"])
+
+    def test_legacy_lock_past_its_absolute_expiry_is_expired_as_before(self):
+        record = self.write_lock(
+            lifecycle=None,
+            created_delta=timedelta(hours=5),
+            heartbeat_delta=timedelta(hours=5),
+            expires_delta=timedelta(hours=-1),
+        )
+        fresh = issue_lock_store.assess_lock_freshness(record, now=self.now)
+        self.assertEqual(fresh["status"], issue_lock_store.STATUS_EXPIRED)
+
+    def test_legacy_lock_is_never_reclaimed_by_the_heartbeat_band(self):
+        record = self.write_lock(
+            lifecycle=None,
+            created_delta=timedelta(hours=3),
+            heartbeat_delta=timedelta(hours=3),
+            expires_delta=timedelta(hours=1),
+        )
+        reclaim = issue_lock_store.assess_expired_lock_reclaim(record, now=self.now)
+        self.assertFalse(reclaim["reclaim_allowed"])
+
+
+class TestReclaimAfterMissedHeartbeat(_LockFixture):
+    def test_missed_heartbeat_makes_ownership_reclaimable(self):
+        record = self.write_lock(
+            pid=os.getpid(),
+            heartbeat_delta=timedelta(minutes=15),
+            expires_delta=timedelta(hours=3),
+        )
+        reclaim = issue_lock_store.assess_expired_lock_reclaim(record, now=self.now)
+        self.assertTrue(reclaim["reclaim_allowed"])
+        self.assertIn("stale_missed_heartbeat", reclaim["reasons"][0])
+
+    def test_live_lease_is_never_reclaimable(self):
+        record = self.write_lock(heartbeat_delta=timedelta(minutes=1))
+        reclaim = issue_lock_store.assess_expired_lock_reclaim(record, now=self.now)
+        self.assertFalse(reclaim["reclaim_allowed"])
+
+    def test_dead_pid_reclaim_path_is_unchanged(self):
+        """#753 must keep working through its original conditions."""
+        record = self.write_lock(pid=DEAD_PID, heartbeat_delta=timedelta(minutes=1))
+        reclaim = issue_lock_store.assess_expired_lock_reclaim(record, now=self.now)
+        self.assertTrue(reclaim["reclaim_allowed"])
+        self.assertTrue(reclaim["owner_pid_dead"])
+
+
+class TestHeartbeatWriter(_LockFixture):
+    """A4: flock + CAS + exact verification, and no revival path."""
+
+    def _heartbeat(self, **kwargs):
+        params = {
+            "remote": REMOTE,
+            "org": ORG,
+            "repo": REPO,
+            "issue_number": ISSUE,
+            "branch_name": BRANCH,
+            "worktree_path": self.worktree,
+            "identity": IDENTITY,
+            "profile": PROFILE,
+            "task_session_id": "author_issue_work-aaaabbbbccccdddd",
+            "lock_dir": self.lock_dir.name,
+            "now": self.now,
+        }
+        params.update(kwargs)
+        return issue_lock_store.heartbeat_session_lock(**params)
+
+    def test_heartbeat_slides_expiry_and_advances_generation(self):
+        self.write_lock(heartbeat_delta=timedelta(minutes=4), generation=5)
+        result = self._heartbeat()
+        self.assertTrue(result["success"], result)
+        self.assertEqual(result["prior_generation"], 5)
+        self.assertEqual(result["lock_generation"], 6)
+        self.assertEqual(result["heartbeat_count"], 1)
+        self.assertEqual(result["last_heartbeat_at"], _ts(self.now))
+        self.assertEqual(result["expires_at"], _ts(self.now + timedelta(minutes=10)))
+        self.assertTrue(result["freshness"]["live"])
+
+    def test_heartbeat_is_durable_and_repeatable(self):
+        self.write_lock(heartbeat_delta=timedelta(minutes=4))
+        self._heartbeat()
+        second = self._heartbeat(now=self.now + timedelta(minutes=1))
+        self.assertTrue(second["success"], second)
+        self.assertEqual(second["heartbeat_count"], 2)
+        written = issue_lock_store.read_lock_file(self._path())
+        self.assertEqual(written["work_lease"]["heartbeat_count"], 2)
+
+    def test_stale_generation_is_refused(self):
+        self.write_lock(generation=5)
+        result = self._heartbeat(expected_generation=4)
+        self.assertFalse(result["success"])
+        self.assertIn("generation changed", result["reasons"][0])
+
+    def test_foreign_session_is_refused(self):
+        self.write_lock()
+        result = self._heartbeat(task_session_id="author_issue_work-ffffffffffffffff")
+        self.assertFalse(result["success"])
+        self.assertIn("task_session_id does not match", " ".join(result["reasons"]))
+
+    def test_missing_session_id_is_refused(self):
+        self.write_lock()
+        result = self._heartbeat(task_session_id="")
+        self.assertFalse(result["success"])
+
+    def test_foreign_claimant_is_refused(self):
+        self.write_lock()
+        for field, value in (
+            ("identity", "someone-else"),
+            ("profile", "other-profile"),
+        ):
+            with self.subTest(field=field):
+                result = self._heartbeat(**{field: value})
+                self.assertFalse(result["success"])
+
+    def test_branch_and_worktree_mismatch_are_refused(self):
+        self.write_lock()
+        wrong_branch = self._heartbeat(branch_name=f"fix/issue-{ISSUE}-other")
+        self.assertFalse(wrong_branch["success"])
+        wrong_worktree = self._heartbeat(worktree_path="/tmp/not-the-worktree")
+        self.assertFalse(wrong_worktree["success"])
+
+    def test_lapsed_lease_cannot_be_heartbeated_back_to_life(self):
+        """No revival path (A4).
+
+        A session that stopped proving liveness must reclaim under a fresh
+        generation, not restore ownership retroactively.
+        """
+        self.write_lock(
+            heartbeat_delta=timedelta(minutes=30), expires_delta=timedelta(hours=1)
+        )
+        result = self._heartbeat()
+        self.assertFalse(result["success"])
+        self.assertIn("reclaimed", " ".join(result["reasons"]))
+
+    def test_absent_lock_cannot_be_created_by_heartbeat(self):
+        result = self._heartbeat()
+        self.assertFalse(result["success"])
+        self.assertIn("no durable lock", result["reasons"][0])
+
+    def test_legacy_lock_is_refused_until_rebound(self):
+        self.write_lock(lifecycle=None)
+        result = self._heartbeat()
+        self.assertFalse(result["success"])
+        self.assertTrue(result["legacy_lease"])
+        self.assertIn("rebound", " ".join(result["reasons"]))
+
+
+class TestLegacyRebind(_LockFixture):
+    """AC-N8 exit route: canonical exact-owner rebinding."""
+
+    def _rebind(self, **kwargs):
+        params = {
+            "remote": REMOTE,
+            "org": ORG,
+            "repo": REPO,
+            "issue_number": ISSUE,
+            "branch_name": BRANCH,
+            "worktree_path": self.worktree,
+            "identity": IDENTITY,
+            "profile": PROFILE,
+            "lock_dir": self.lock_dir.name,
+            "now": self.now,
+        }
+        params.update(kwargs)
+        return issue_lock_store.rebind_legacy_lock(**params)
+
+    def test_rebind_mints_a_session_and_a_genuine_first_heartbeat(self):
+        self.write_lock(
+            lifecycle=None,
+            created_delta=timedelta(hours=3),
+            heartbeat_delta=timedelta(hours=3),
+            expires_delta=timedelta(hours=1),
+            generation=2,
+        )
+        result = self._rebind()
+        self.assertTrue(result["success"], result)
+        self.assertTrue(result["task_session_id"])
+        self.assertEqual(result["lock_generation"], 3)
+
+        written = issue_lock_store.read_lock_file(self._path())
+        lease = written["work_lease"]
+        self.assertEqual(
+            lease["lifecycle_version"], lease_policy.LIFECYCLE_HEARTBEAT_V1
+        )
+        self.assertEqual(lease["last_heartbeat_at"], _ts(self.now))
+        self.assertEqual(lease["expires_at"], _ts(self.now + timedelta(minutes=10)))
+        self.assertFalse(issue_lock_store.is_legacy_lease(written))
+        # The original claim is preserved for audit rather than overwritten.
+        origin = written["legacy_rebind"]["legacy_origin"]
+        self.assertTrue(origin["created_at"])
+        self.assertEqual(origin["lifecycle"], lease_policy.LIFECYCLE_LEGACY)
+
+    def test_rebound_lock_can_then_heartbeat(self):
+        self.write_lock(
+            lifecycle=None,
+            created_delta=timedelta(hours=3),
+            heartbeat_delta=timedelta(hours=3),
+            expires_delta=timedelta(hours=1),
+        )
+        rebound = self._rebind()
+        beat = issue_lock_store.heartbeat_session_lock(
+            remote=REMOTE,
+            org=ORG,
+            repo=REPO,
+            issue_number=ISSUE,
+            branch_name=BRANCH,
+            worktree_path=self.worktree,
+            identity=IDENTITY,
+            profile=PROFILE,
+            task_session_id=rebound["task_session_id"],
+            lock_dir=self.lock_dir.name,
+            now=self.now + timedelta(minutes=1),
+        )
+        self.assertTrue(beat["success"], beat)
+
+    def test_rebind_refuses_a_foreign_owner(self):
+        self.write_lock(lifecycle=None, expires_delta=timedelta(hours=1))
+        result = self._rebind(identity="someone-else")
+        self.assertFalse(result["success"])
+
+    def test_rebind_refuses_a_lock_already_on_the_lifecycle(self):
+        self.write_lock()
+        result = self._rebind()
+        self.assertFalse(result["success"])
+        self.assertFalse(result["legacy_lease"])
+
+    def test_rebind_is_not_a_recovery_path_for_a_lapsed_legacy_lease(self):
+        """An expired legacy lease belongs to #760 renewal or #601 reclaim."""
+        self.write_lock(
+            lifecycle=None,
+            created_delta=timedelta(hours=5),
+            heartbeat_delta=timedelta(hours=5),
+            expires_delta=timedelta(hours=-1),
+        )
+        result = self._rebind()
+        self.assertFalse(result["success"])
+        self.assertIn("not a recovery path", " ".join(result["reasons"]))
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_master_parity_gate.py

@@ -78,95 +78,6 @@ class TestBlockReasonsAndReport(unittest.TestCase):
         self.assertTrue(report["recovery"])
 
 
-class TestLiveRemoteParity(unittest.TestCase):
-    """#610: parity must account for the live remote master, not just local.
-
-    The daemon can be stale relative to the live remote target while the local
-    checkout HEAD still matches the daemon's startup commit, so local parity
-    reports green even though a mutation would run against outdated code.
-    """
-
-    SHA_C = "c" * 40
-
-    def test_distinguishes_three_shas(self):
-        res = mp.assess_master_parity(
-            {"startup_head": SHA_A}, SHA_A, live_remote_head=SHA_B)
-        self.assertEqual(res["daemon_start_head"], SHA_A)
-        self.assertEqual(res["local_head"], SHA_A)
-        self.assertEqual(res["live_remote_head"], SHA_B)
-
-    def test_mutation_safe_only_when_all_three_match(self):
-        res = mp.assess_master_parity(
-            {"startup_head": SHA_A}, SHA_A, live_remote_head=SHA_A)
-        self.assertTrue(res["mutation_safe"])
-        self.assertTrue(res["live_known"])
-        self.assertFalse(res["live_stale"])
-
-    def test_live_stale_when_remote_advanced_past_daemon(self):
-        # Local checkout still matches the daemon start (local parity green),
-        # but the live remote master has advanced -> daemon is live-stale.
-        res = mp.assess_master_parity(
-            {"startup_head": SHA_A}, SHA_A, live_remote_head=SHA_B)
-        self.assertTrue(res["in_parity"])  # local parity still green
-        self.assertTrue(res["live_stale"])
-        self.assertFalse(res["mutation_safe"])
-        self.assertTrue(any("live" in r.lower() for r in res["reasons"]))
-
-    def test_live_unknown_is_not_mutation_safe_but_not_stale(self):
-        # Non-goal: unfetchable live remote must not be treated as stale for
-        # read-only, but a mutation-safe claim fails closed.
-        res = mp.assess_master_parity(
-            {"startup_head": SHA_A}, SHA_A, live_remote_head=None)
-        self.assertFalse(res["live_known"])
-        self.assertFalse(res["mutation_safe"])
-        self.assertFalse(res["live_stale"])
-        self.assertTrue(res["in_parity"])
-
-    def test_default_live_remote_preserves_legacy_shape(self):
-        # Callers that do not supply a live head keep the pre-#610 behavior:
-        # in-parity, not live-stale, no live-derived block.
-        res = mp.assess_master_parity({"startup_head": SHA_A}, SHA_A)
-        self.assertFalse(res["live_stale"])
-        self.assertEqual(mp.parity_block_reasons(res), [])
-
-
-class TestLiveStaleBlockAndReport(unittest.TestCase):
-    """#610: live-staleness must block mutations and surface a typed blocker."""
-
-    def test_live_stale_produces_block_reasons(self):
-        res = mp.assess_master_parity(
-            {"startup_head": SHA_A}, SHA_A, live_remote_head=SHA_B)
-        self.assertTrue(mp.parity_block_reasons(res))
-
-    def test_disable_env_suppresses_live_stale_block(self):
-        res = mp.assess_master_parity(
-            {"startup_head": SHA_A}, SHA_A, live_remote_head=SHA_B)
-        with patch.dict(os.environ, {mp.ENV_DISABLE: "1"}):
-            self.assertEqual(mp.parity_block_reasons(res), [])
-
-    def test_resolver_disagreement_returns_typed_blocker(self):
-        # Parity says local-green, resolver says restart required -> disagreement
-        # is a typed, fail-closed blocker naming the resolver as authoritative.
-        res = mp.assess_master_parity({"startup_head": SHA_A}, SHA_A)
-        blocker = mp.parity_resolver_disagreement(res, resolver_restart_required=True)
-        self.assertIsNotNone(blocker)
-        self.assertEqual(blocker["kind"], "parity_resolver_disagreement")
-        self.assertTrue(blocker["restart_required"])
-        self.assertTrue(blocker["resolver_authoritative"])
-
-    def test_no_disagreement_when_resolver_agrees(self):
-        res = mp.assess_master_parity({"startup_head": SHA_A}, SHA_A)
-        self.assertIsNone(
-            mp.parity_resolver_disagreement(res, resolver_restart_required=False))
-
-    def test_live_stale_report_names_live_remote(self):
-        res = mp.assess_master_parity(
-            {"startup_head": SHA_A}, SHA_A, live_remote_head=SHA_B)
-        report = mp.parity_report(res)
-        self.assertEqual(report["live_remote_head"], SHA_B)
-        self.assertTrue(report["restart_required"])
-
-
 class TestReadGitHead(unittest.TestCase):
     def test_test_override_takes_precedence(self):
         with patch.dict(os.environ, {mp.ENV_TEST_CURRENT_HEAD: SHA_B}):
@@ -184,149 +95,6 @@ class TestReadGitHead(unittest.TestCase):
             self.assertIsNone(mp.read_git_head(""))
 
 
-class TestReadRemoteMasterHead(unittest.TestCase):
-    """#610: live remote master head reader (env-overridable, fails to None)."""
-
-    def test_test_override_takes_precedence(self):
-        with patch.dict(os.environ, {mp.ENV_TEST_LIVE_REMOTE_HEAD: SHA_B}):
-            self.assertEqual(mp.read_remote_master_head("/nonexistent"), SHA_B)
-
-    def test_blank_override_is_none(self):
-        with patch.dict(os.environ, {mp.ENV_TEST_LIVE_REMOTE_HEAD: "   "}):
-            self.assertIsNone(mp.read_remote_master_head("/nonexistent"))
-
-    def test_unfetchable_remote_is_none(self):
-        # No override; a bogus root/remote must fail closed to None, never raise.
-        env = {k: v for k, v in os.environ.items()
-               if k != mp.ENV_TEST_LIVE_REMOTE_HEAD}
-        with patch.dict(os.environ, env, clear=True):
-            self.assertIsNone(
-                mp.read_remote_master_head("/nonexistent", remote="nope"))
-
-
-class TestRemoteHeadCache(unittest.TestCase):
-    """#610: live remote reads are cached with a TTL to stay off the network.
-
-    The parity gate runs on every mutation and every runtime-context read, so an
-    unbounded ``git ls-remote`` per call would be a latency/flakiness regression.
-    """
-
-    def setUp(self):
-        # These cases intentionally exercise the subprocess/cache path, so they
-        # opt out of suite-wide hermetic mode (PR #788 F1).
-        self._saved_hermetic = mp.hermetic_test_mode()
-        mp.set_hermetic_test_mode(False)
-        mp._clear_remote_head_cache()
-        env = {
-            k: v for k, v in os.environ.items()
-            if k not in (mp.ENV_TEST_LIVE_REMOTE_HEAD,
-                         mp.ENV_TEST_ALLOW_LIVE_REMOTE_PROBE,
-                         "PYTEST_CURRENT_TEST")
-        }
-        # Allow the probe path under hermetic defenses while still mocking
-        # subprocess so no real network call runs.
-        env[mp.ENV_TEST_ALLOW_LIVE_REMOTE_PROBE] = "1"
-        self._env = patch.dict(os.environ, env, clear=True)
-        self._env.start()
-        self.addCleanup(self._env.stop)
-        self.addCleanup(mp._clear_remote_head_cache)
-        self.addCleanup(
-            lambda: mp.set_hermetic_test_mode(self._saved_hermetic)
-        )
-
-    def _fake_run(self, sha):
-        class _R:
-            returncode = 0
-            stdout = f"{sha}\trefs/heads/master\n"
-        calls = {"n": 0}
-
-        def run(*args, **kwargs):
-            calls["n"] += 1
-            return _R()
-        return run, calls
-
-    def test_second_call_within_ttl_uses_cache(self):
-        run, calls = self._fake_run(SHA_B)
-        with patch.object(mp.subprocess, "run", run):
-            a = mp.read_remote_master_head("/repo", remote="prgs", ttl=100)
-            b = mp.read_remote_master_head("/repo", remote="prgs", ttl=100)
-        self.assertEqual(a, SHA_B)
-        self.assertEqual(b, SHA_B)
-        self.assertEqual(calls["n"], 1)
-
-    def test_zero_ttl_bypasses_cache(self):
-        run, calls = self._fake_run(SHA_B)
-        with patch.object(mp.subprocess, "run", run):
-            mp.read_remote_master_head("/repo", remote="prgs", ttl=0)
-            mp.read_remote_master_head("/repo", remote="prgs", ttl=0)
-        self.assertEqual(calls["n"], 2)
-
-    def test_env_override_never_touches_subprocess(self):
-        run, calls = self._fake_run(SHA_B)
-        with patch.dict(os.environ, {mp.ENV_TEST_LIVE_REMOTE_HEAD: SHA_A}):
-            with patch.object(mp.subprocess, "run", run):
-                self.assertEqual(
-                    mp.read_remote_master_head("/repo", remote="prgs"), SHA_A)
-        self.assertEqual(calls["n"], 0)
-
-
-class TestHermeticLiveRemoteReads(unittest.TestCase):
-    """#610 / PR #788 F1/F2: suite hermetic mode never hits the network."""
-
-    def setUp(self):
-        self._saved = mp.hermetic_test_mode()
-        mp.set_hermetic_test_mode(True)
-        mp._clear_remote_head_cache()
-        self.addCleanup(lambda: mp.set_hermetic_test_mode(self._saved))
-        self.addCleanup(mp._clear_remote_head_cache)
-
-    def test_hermetic_mode_returns_none_without_subprocess(self):
-        run_calls = {"n": 0}
-
-        def boom(*args, **kwargs):
-            run_calls["n"] += 1
-            raise AssertionError("ls-remote must not run under hermetic mode")
-
-        env = {
-            k: v for k, v in os.environ.items()
-            if k not in (mp.ENV_TEST_LIVE_REMOTE_HEAD,
-                         mp.ENV_TEST_ALLOW_LIVE_REMOTE_PROBE)
-        }
-        with patch.dict(os.environ, env, clear=True):
-            with patch.object(mp.subprocess, "run", boom):
-                self.assertIsNone(
-                    mp.read_remote_master_head("/repo", remote="prgs")
-                )
-        self.assertEqual(run_calls["n"], 0)
-
-    def test_hermetic_mode_survives_clear_true_env(self):
-        """Module flag, not env pin: clear=True cannot re-enable the probe."""
-        run_calls = {"n": 0}
-
-        def boom(*args, **kwargs):
-            run_calls["n"] += 1
-            raise AssertionError("ls-remote must not run after clear=True")
-
-        with patch.dict(os.environ, {}, clear=True):
-            with patch.object(mp.subprocess, "run", boom):
-                self.assertIsNone(mp.read_remote_master_head("/repo"))
-        self.assertEqual(run_calls["n"], 0)
-
-    def test_explicit_override_still_wins_under_hermetic(self):
-        run_calls = {"n": 0}
-
-        def boom(*args, **kwargs):
-            run_calls["n"] += 1
-            raise AssertionError("override must bypass subprocess")
-
-        with patch.dict(os.environ, {mp.ENV_TEST_LIVE_REMOTE_HEAD: SHA_B}):
-            with patch.object(mp.subprocess, "run", boom):
-                self.assertEqual(
-                    mp.read_remote_master_head("/repo"), SHA_B
-                )
-        self.assertEqual(run_calls["n"], 0)
-
-
 class TestServerWiring(unittest.TestCase):
     """Integration with the gate choke point in the server namespace."""
 
@@ -337,13 +105,6 @@ class TestServerWiring(unittest.TestCase):
         self._saved = self.srv._STARTUP_PARITY
         self.srv._STARTUP_PARITY = {"root": self.srv.PROJECT_ROOT,
                                     "startup_head": SHA_A}
-        # Keep the live-remote read hermetic (no real ls-remote network call):
-        # default the live master to the daemon start so parity is fully green
-        # unless a test overrides the live head explicitly (#610).
-        self._live_patch = patch.dict(
-            os.environ, {mp.ENV_TEST_LIVE_REMOTE_HEAD: SHA_A})
-        self._live_patch.start()
-        self.addCleanup(self._live_patch.stop)
 
     def tearDown(self):
         self.srv._STARTUP_PARITY = self._saved
@@ -386,36 +147,6 @@ class TestServerWiring(unittest.TestCase):
             self.assertTrue(out["in_parity"])
             self.assertNotIn("report", out)
 
-    # --- #610: live-remote wiring -------------------------------------------
-
-    def test_live_stale_blocks_mutation_though_local_green(self):
-        # Local checkout matches the daemon start (local parity green) but the
-        # live remote master has advanced -> mutations must fail closed.
-        with patch.dict(os.environ, {mp.ENV_TEST_CURRENT_HEAD: SHA_A,
-                                     mp.ENV_TEST_LIVE_REMOTE_HEAD: SHA_B}):
-            self.assertEqual(self.srv._master_parity_block("gitea.read"), [])
-            self.assertTrue(
-                self.srv._master_parity_block("gitea.pr.create"))
-
-    def test_assess_tool_exposes_three_distinct_shas(self):
-        with patch.dict(os.environ, {mp.ENV_TEST_CURRENT_HEAD: SHA_A,
-                                     mp.ENV_TEST_LIVE_REMOTE_HEAD: SHA_B}):
-            out = self.srv.gitea_assess_master_parity(remote="prgs")
-            self.assertEqual(out["daemon_start_head"], SHA_A)
-            self.assertEqual(out["local_head"], SHA_A)
-            self.assertEqual(out["live_remote_head"], SHA_B)
-            self.assertTrue(out["live_stale"])
-            self.assertFalse(out["mutation_safe"])
-            self.assertIn("report", out)
-
-    def test_assess_tool_mutation_safe_when_all_three_match(self):
-        with patch.dict(os.environ, {mp.ENV_TEST_CURRENT_HEAD: SHA_A,
-                                     mp.ENV_TEST_LIVE_REMOTE_HEAD: SHA_A}):
-            out = self.srv.gitea_assess_master_parity(remote="prgs")
-            self.assertTrue(out["mutation_safe"])
-            self.assertFalse(out["live_stale"])
-            self.assertNotIn("report", out)
-
 
 if __name__ == "__main__":
     unittest.main()

tests/test_webui_worker_registry.py

@@ -1,458 +0,0 @@
-"""Tests for the worker registry and configuration schema (#798, epic #797)."""
-import json
-import sys
-import tempfile
-import unittest
-from pathlib import Path
-
-sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
-
-from webui.worker_registry import (
-    ALLOWED_ROLES,
-    SCHEMA_VERSION,
-    RegistryValidationError,
-    WorkerRegistry,
-    default_registry_path,
-    find_provider,
-    find_worker,
-    history_dir,
-    list_revisions,
-    load_registry,
-    registry_to_dict,
-    registry_to_document,
-    rollback_to_revision,
-    save_registry,
-    validate_payload,
-    worker_to_dict,
-    workers_for_provider,
-)
-
-_EXPECTED_PROVIDER_IDS = ("claude", "grok", "codex", "agy", "kimi-k")
-
-
-def _provider(provider_id: str = "claude", **overrides) -> dict:
-    payload = {
-        "id": provider_id,
-        "display_name": "Claude",
-        "vendor": "Anthropic",
-        "executable": "claude",
-        "available": True,
-        "models": ["claude-opus-4-8"],
-        "notes": "",
-    }
-    payload.update(overrides)
-    return payload
-
-
-def _worker(worker_id: str = "claude-author", **overrides) -> dict:
-    payload = {
-        "id": worker_id,
-        "display_name": "Claude author",
-        "provider": "claude",
-        "model": "claude-opus-4-8",
-        "project": "gitea-tools",
-        "role": "author",
-        "namespace": "gitea-author",
-        "profile": "prgs-author",
-        "workflow": "skills/llm-project-workflow/workflows/work-issue.md",
-        "schedule": {"kind": "cron", "expression": "0 * * * *"},
-        "timeout_seconds": 3600,
-        "enabled": True,
-        "scheduler": {"kind": "launchd", "label": "cc.prgs.claude.author"},
-        "notes": "",
-    }
-    payload.update(overrides)
-    return payload
-
-
-def _document(providers=None, workers=None, **overrides) -> dict:
-    payload = {
-        "version": SCHEMA_VERSION,
-        "revision": 1,
-        "updated_at": "2026-07-22T00:00:00Z",
-        "providers": providers if providers is not None else [_provider()],
-        "workers": workers if workers is not None else [_worker()],
-    }
-    payload.update(overrides)
-    return payload
-
-
-class _TempRegistryCase(unittest.TestCase):
-    """Base case giving each test an isolated registry file."""
-
-    def setUp(self):
-        self._tmp = tempfile.TemporaryDirectory()
-        self.addCleanup(self._tmp.cleanup)
-        self.path = Path(self._tmp.name) / "workers.registry.json"
-
-    def write(self, document: dict) -> Path:
-        self.path.write_text(json.dumps(document, indent=2) + "\n", encoding="utf-8")
-        return self.path
-
-    def parse(self, document: dict) -> WorkerRegistry:
-        return validate_payload(document, source_path=self.path)
-
-
-class TestPackagedRegistry(unittest.TestCase):
-    """AC: the declarative registry is the source of truth and ships with the app."""
-
-    def test_default_path_points_at_packaged_data(self):
-        path = default_registry_path()
-        self.assertEqual(path.name, "workers.registry.json")
-        self.assertEqual(path.parent.name, "data")
-
-    def test_packaged_registry_loads_and_validates(self):
-        registry = load_registry()
-        self.assertEqual(registry.version, SCHEMA_VERSION)
-        self.assertGreaterEqual(registry.revision, 1)
-
-    def test_packaged_registry_declares_all_five_providers(self):
-        registry = load_registry()
-        self.assertEqual(
-            tuple(provider.id for provider in registry.providers),
-            _EXPECTED_PROVIDER_IDS,
-        )
-
-    def test_packaged_registry_carries_no_credentials(self):
-        raw = default_registry_path().read_text(encoding="utf-8").lower()
-        for marker in ("token", "password", "secret", "api_key", "credential"):
-            self.assertNotIn(marker, raw)
-
-
-class TestSeparateEntities(_TempRegistryCase):
-    """AC: providers and configured workers are separate entities."""
-
-    def test_provider_may_exist_with_no_workers(self):
-        registry = self.parse(
-            _document(providers=[_provider("grok", display_name="Grok")], workers=[])
-        )
-        self.assertEqual(len(registry.providers), 1)
-        self.assertEqual(registry.workers, ())
-        self.assertEqual(workers_for_provider(registry, "grok"), ())
-
-    def test_many_workers_may_share_one_provider(self):
-        registry = self.parse(
-            _document(
-                workers=[
-                    _worker("claude-author"),
-                    _worker(
-                        "claude-reviewer",
-                        role="reviewer",
-                        namespace="gitea-reviewer",
-                        profile="prgs-reviewer",
-                        scheduler={"kind": "launchd", "label": "cc.prgs.claude.reviewer"},
-                    ),
-                ]
-            )
-        )
-        self.assertEqual(len(workers_for_provider(registry, "claude")), 2)
-        self.assertEqual(len(registry.providers), 1)
-
-    def test_worker_referencing_unknown_provider_is_refused(self):
-        with self.assertRaises(RegistryValidationError) as ctx:
-            self.parse(_document(workers=[_worker(provider="mystery")]))
-        self.assertIn("unknown provider", str(ctx.exception))
-
-    def test_lookup_helpers(self):
-        registry = self.parse(_document())
-        self.assertIsNotNone(find_worker(registry, "claude-author"))
-        self.assertIsNone(find_worker(registry, "absent"))
-        self.assertIsNotNone(find_provider(registry, "claude"))
-        self.assertIsNone(find_provider(registry, "absent"))
-
-
-class TestRecordedFields(_TempRegistryCase):
-    """AC: records provider, model, project, role, namespace/profile, workflow,
-    schedule, timeout, enabled state, and scheduler metadata."""
-
-    def test_every_required_field_is_recorded(self):
-        registry = self.parse(_document())
-        worker = registry.workers[0]
-        self.assertEqual(worker.provider, "claude")
-        self.assertEqual(worker.model, "claude-opus-4-8")
-        self.assertEqual(worker.project, "gitea-tools")
-        self.assertEqual(worker.role, "author")
-        self.assertEqual(worker.namespace, "gitea-author")
-        self.assertEqual(worker.profile, "prgs-author")
-        self.assertEqual(worker.workflow, "skills/llm-project-workflow/workflows/work-issue.md")
-        self.assertEqual(worker.schedule.kind, "cron")
-        self.assertEqual(worker.schedule.expression, "0 * * * *")
-        self.assertEqual(worker.timeout_seconds, 3600)
-        self.assertTrue(worker.enabled)
-        self.assertEqual(worker.scheduler.kind, "launchd")
-        self.assertEqual(worker.scheduler.label, "cc.prgs.claude.author")
-
-    def test_each_required_field_is_individually_required(self):
-        for field in (
-            "provider", "model", "project", "role", "namespace",
-            "profile", "workflow", "schedule", "timeout_seconds",
-            "enabled", "scheduler", "id", "display_name",
-        ):
-            with self.subTest(field=field):
-                worker = _worker()
-                worker.pop(field)
-                with self.assertRaises(RegistryValidationError):
-                    self.parse(_document(workers=[worker]))
-
-    def test_all_sanctioned_roles_are_accepted(self):
-        for role in ALLOWED_ROLES:
-            with self.subTest(role=role):
-                registry = self.parse(_document(workers=[_worker(role=role)]))
-                self.assertEqual(registry.workers[0].role, role)
-
-    def test_unsanctioned_role_is_refused(self):
-        with self.assertRaises(RegistryValidationError) as ctx:
-            self.parse(_document(workers=[_worker(role="admin")]))
-        self.assertIn("role must be one of", str(ctx.exception))
-
-    def test_worker_dict_round_trips_every_field(self):
-        registry = self.parse(_document())
-        encoded = worker_to_dict(registry.workers[0])
-        self.assertEqual(encoded, _worker())
-        json.dumps(encoded)  # must stay JSON-safe for the #799 API
-
-
-class TestSchemaValidation(_TempRegistryCase):
-    """AC: supports schema validation — and fails closed."""
-
-    def test_unsupported_version_is_refused(self):
-        with self.assertRaises(RegistryValidationError):
-            self.parse(_document(version=2))
-
-    def test_root_must_be_an_object(self):
-        with self.assertRaises(RegistryValidationError):
-            validate_payload([], source_path=self.path)
-
-    def test_providers_must_be_non_empty(self):
-        with self.assertRaises(RegistryValidationError):
-            self.parse(_document(providers=[]))
-
-    def test_unknown_top_level_field_is_refused(self):
-        with self.assertRaises(RegistryValidationError) as ctx:
-            self.parse(_document(fleet=[]))
-        self.assertIn("unknown fields", str(ctx.exception))
-
-    def test_unknown_worker_field_is_refused_not_ignored(self):
-        # A typo'd field must not be silently dropped: "timeout_second" would
-        # otherwise read as "no timeout declared".
-        worker = _worker()
-        worker["timeout_second"] = 30
-        with self.assertRaises(RegistryValidationError) as ctx:
-            self.parse(_document(workers=[worker]))
-        self.assertIn("timeout_second", str(ctx.exception))
-
-    def test_credentials_are_refused_anywhere_in_the_document(self):
-        for label, mutate in (
-            ("provider.api_token", lambda doc: doc["providers"][0].__setitem__("api_token", "x")),
-            ("worker.password", lambda doc: doc["workers"][0].__setitem__("password", "x")),
-            ("root.secret", lambda doc: doc.__setitem__("secret", "x")),
-        ):
-            with self.subTest(field=label):
-                document = _document()
-                mutate(document)
-                with self.assertRaises(ValueError) as ctx:
-                    self.parse(document)
-                self.assertIn("credential", str(ctx.exception).lower())
-
-    def test_duplicate_worker_id_is_refused(self):
-        workers = [_worker("dup"), _worker("dup", scheduler={"kind": "manual"})]
-        with self.assertRaises(RegistryValidationError) as ctx:
-            self.parse(_document(workers=workers))
-        self.assertIn("duplicate worker id", str(ctx.exception))
-
-    def test_duplicate_provider_id_is_refused(self):
-        with self.assertRaises(RegistryValidationError) as ctx:
-            self.parse(_document(providers=[_provider("claude"), _provider("claude")], workers=[]))
-        self.assertIn("duplicate provider id", str(ctx.exception))
-
-    def test_duplicate_launchagent_label_is_refused(self):
-        # Two workers sharing a label would silently overwrite each other's agent.
-        workers = [
-            _worker("a", scheduler={"kind": "launchd", "label": "cc.prgs.same"}),
-            _worker("b", scheduler={"kind": "launchd", "label": "cc.prgs.same"}),
-        ]
-        with self.assertRaises(RegistryValidationError) as ctx:
-            self.parse(_document(workers=workers))
-        self.assertIn("duplicate scheduler label", str(ctx.exception))
-
-    def test_manual_scheduler_needs_no_label_and_many_may_coexist(self):
-        workers = [
-            _worker("a", scheduler={"kind": "manual"}),
-            _worker("b", scheduler={"kind": "manual"}),
-        ]
-        registry = self.parse(_document(workers=workers))
-        self.assertEqual([w.scheduler.label for w in registry.workers], [None, None])
-
-    def test_launchd_scheduler_requires_a_label(self):
-        with self.assertRaises(RegistryValidationError) as ctx:
-            self.parse(_document(workers=[_worker(scheduler={"kind": "launchd"})]))
-        self.assertIn("label is required", str(ctx.exception))
-
-    def test_unknown_scheduler_kind_is_refused(self):
-        with self.assertRaises(RegistryValidationError):
-            self.parse(_document(workers=[_worker(scheduler={"kind": "systemd", "label": "x"})]))
-
-    def test_timeout_must_be_a_positive_bounded_integer(self):
-        for bad in (0, -1, "3600", 1.5, True, 86_401):
-            with self.subTest(timeout=bad):
-                with self.assertRaises(RegistryValidationError):
-                    self.parse(_document(workers=[_worker(timeout_seconds=bad)]))
-
-    def test_enabled_must_be_a_real_boolean(self):
-        for bad in ("true", 1, None):
-            with self.subTest(enabled=bad):
-                with self.assertRaises(RegistryValidationError):
-                    self.parse(_document(workers=[_worker(enabled=bad)]))
-
-    def test_identifier_shape_is_enforced(self):
-        for bad in ("Claude Author", "-leading", "UPPER", ""):
-            with self.subTest(worker_id=bad):
-                with self.assertRaises(RegistryValidationError):
-                    self.parse(_document(workers=[_worker(bad)]))
-
-
-class TestScheduleValidation(_TempRegistryCase):
-    """Schedules are declarations; next-run computation belongs to #803."""
-
-    def test_interval_schedule_requires_positive_seconds(self):
-        registry = self.parse(
-            _document(workers=[_worker(schedule={"kind": "interval", "seconds": 900})])
-        )
-        self.assertEqual(registry.workers[0].schedule.seconds, 900)
-        with self.assertRaises(RegistryValidationError):
-            self.parse(_document(workers=[_worker(schedule={"kind": "interval"})]))
-        with self.assertRaises(RegistryValidationError):
-            self.parse(_document(workers=[_worker(schedule={"kind": "interval", "seconds": 0})]))
-
-    def test_cron_schedule_requires_five_fields(self):
-        with self.assertRaises(RegistryValidationError) as ctx:
-            self.parse(_document(workers=[_worker(schedule={"kind": "cron", "expression": "0 *"})]))
-        self.assertIn("five crontab fields", str(ctx.exception))
-
-    def test_manual_schedule_needs_no_timing(self):
-        registry = self.parse(_document(workers=[_worker(schedule={"kind": "manual"})]))
-        schedule = registry.workers[0].schedule
-        self.assertEqual(schedule.kind, "manual")
-        self.assertIsNone(schedule.seconds)
-        self.assertIsNone(schedule.expression)
-
-    def test_fields_from_the_wrong_kind_are_refused(self):
-        with self.assertRaises(RegistryValidationError) as ctx:
-            self.parse(_document(workers=[_worker(schedule={"kind": "manual", "seconds": 60})]))
-        self.assertIn("not valid for kind", str(ctx.exception))
-
-    def test_unknown_schedule_kind_is_refused(self):
-        with self.assertRaises(RegistryValidationError):
-            self.parse(_document(workers=[_worker(schedule={"kind": "hourly"})]))
-
-
-class TestAtomicPersistence(_TempRegistryCase):
-    """AC: atomic persistence."""
-
-    def test_save_then_load_round_trips(self):
-        registry = self.parse(_document())
-        save_registry(registry, self.path)
-        reloaded = load_registry(self.path)
-        self.assertEqual(
-            [worker_to_dict(w) for w in reloaded.workers],
-            [worker_to_dict(w) for w in registry.workers],
-        )
-
-    def test_save_leaves_no_temp_files_behind(self):
-        registry = self.parse(_document())
-        save_registry(registry, self.path)
-        save_registry(registry, self.path)
-        leftovers = [p.name for p in self.path.parent.iterdir() if p.name.startswith(".")]
-        self.assertEqual(leftovers, [])
-
-    def test_save_refuses_to_persist_an_invalid_document(self):
-        registry = self.parse(_document())
-        broken = WorkerRegistry(
-            version=registry.version,
-            revision=registry.revision,
-            updated_at=registry.updated_at,
-            providers=registry.providers,
-            # A worker whose provider is not declared in the registry.
-            workers=tuple(
-                type(worker)(**{**worker.__dict__, "provider": "vanished"})
-                for worker in registry.workers
-            ),
-            source_path=self.path,
-        )
-        with self.assertRaises(RegistryValidationError):
-            save_registry(broken, self.path)
-        self.assertFalse(self.path.exists(), "invalid save must not create the file")
-
-    def test_document_shape_excludes_local_paths_but_api_shape_includes_it(self):
-        registry = self.parse(_document())
-        self.assertNotIn("source_path", registry_to_document(registry))
-        self.assertEqual(registry_to_dict(registry)["source_path"], str(self.path))
-
-
-class TestVersioningAndRollback(_TempRegistryCase):
-    """AC: versioning and rollback."""
-
-    def _seed(self) -> WorkerRegistry:
-        self.write(_document())
-        return load_registry(self.path)
-
-    def test_revision_increments_on_each_save(self):
-        registry = self._seed()
-        self.assertEqual(registry.revision, 1)
-        second = save_registry(registry, self.path)
-        self.assertEqual(second.revision, 2)
-        third = save_registry(second, self.path)
-        self.assertEqual(third.revision, 3)
-
-    def test_updated_at_is_refreshed_and_utc(self):
-        registry = self._seed()
-        saved = save_registry(registry, self.path)
-        self.assertRegex(saved.updated_at, r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}Z$")
-
-    def test_superseded_revisions_are_retained(self):
-        registry = self._seed()
-        second = save_registry(registry, self.path)
-        save_registry(second, self.path)
-        self.assertEqual(list_revisions(self.path), (1, 2))
-        self.assertTrue(history_dir(self.path).is_dir())
-
-    def test_rollback_restores_prior_content_as_a_new_revision(self):
-        self.write(_document(workers=[_worker("original")]))
-        registry = load_registry(self.path)
-
-        changed = WorkerRegistry(
-            version=registry.version,
-            revision=registry.revision,
-            updated_at=registry.updated_at,
-            providers=registry.providers,
-            workers=(),  # operator deletes every worker
-            source_path=self.path,
-        )
-        save_registry(changed, self.path)
-        self.assertEqual(load_registry(self.path).workers, ())
-
-        restored = rollback_to_revision(1, self.path)
-        self.assertEqual([w.id for w in restored.workers], ["original"])
-        # Append-only: the rollback publishes a new head rather than rewinding.
-        self.assertGreater(restored.revision, 2)
-        self.assertEqual([w.id for w in load_registry(self.path).workers], ["original"])
-
-    def test_rollback_to_unknown_revision_fails_closed(self):
-        self._seed()
-        with self.assertRaises(RegistryValidationError) as ctx:
-            rollback_to_revision(99, self.path)
-        self.assertIn("not retained", str(ctx.exception))
-
-    def test_revision_must_be_a_positive_integer(self):
-        for bad in (0, -1, "1", None):
-            with self.subTest(revision=bad):
-                with self.assertRaises(RegistryValidationError):
-                    self.parse(_document(revision=bad))
-
-    def test_history_is_empty_before_any_save(self):
-        self.write(_document())
-        self.assertEqual(list_revisions(self.path), ())
-
-
-if __name__ == "__main__":
-    unittest.main()

webui/data/workers.registry.json

@@ -1,57 +0,0 @@
-{
-  "version": 1,
-  "revision": 1,
-  "updated_at": "2026-07-22T00:00:00Z",
-  "providers": [
-    {
-      "id": "claude",
-      "display_name": "Claude",
-      "vendor": "Anthropic",
-      "executable": "claude",
-      "available": true,
-      "models": [
-        "claude-opus-4-8",
-        "claude-sonnet-5",
-        "claude-haiku-4-5-20251001"
-      ],
-      "notes": "Model list is a declaration. Live enumeration and version inspection belong to the provider adapter framework (#800)."
-    },
-    {
-      "id": "grok",
-      "display_name": "Grok",
-      "vendor": "xAI",
-      "executable": "grok",
-      "available": true,
-      "models": [],
-      "notes": "Models enumerated by the provider adapter (#800); not declared here."
-    },
-    {
-      "id": "codex",
-      "display_name": "Codex",
-      "vendor": "OpenAI",
-      "executable": "codex",
-      "available": true,
-      "models": [],
-      "notes": "Models enumerated by the provider adapter (#800); not declared here."
-    },
-    {
-      "id": "agy",
-      "display_name": "AGY",
-      "vendor": "Antigravity",
-      "executable": "agy",
-      "available": true,
-      "models": [],
-      "notes": "MCP allowlist gating applies to this provider; confirm server-side allowlist before configuring a worker."
-    },
-    {
-      "id": "kimi-k",
-      "display_name": "Kimi K",
-      "vendor": "Moonshot AI",
-      "executable": "kimi",
-      "available": true,
-      "models": [],
-      "notes": "Provider id is kimi-k; the executable on PATH is kimi. Models enumerated by the provider adapter (#800)."
-    }
-  ],
-  "workers": []
-}

webui/project_registry.py

@@ -8,7 +8,27 @@ from dataclasses import dataclass
 from pathlib import Path
 from typing import Any
 
-from webui.registry_safety import reject_credential_keys as _reject_credential_keys
+_FORBIDDEN_EXACT_KEYS = frozenset({
+    "token",
+    "password",
+    "secret",
+    "credential",
+    "auth",
+    "api_key",
+    "api-key",
+})
+_FORBIDDEN_KEY_PREFIXES = ("auth_", "api_key_", "api-key_")
+_FORBIDDEN_KEY_SUFFIXES = ("_token", "_secret", "_password", "_credential", "_auth")
+
+
+def _is_forbidden_key(key: str) -> bool:
+    lowered = key.lower()
+    if lowered in _FORBIDDEN_EXACT_KEYS:
+        return True
+    return (
+        lowered.startswith(_FORBIDDEN_KEY_PREFIXES)
+        or lowered.endswith(_FORBIDDEN_KEY_SUFFIXES)
+    )
 
 _REQUIRED_PROJECT_FIELDS = (
     "id",
@@ -59,6 +79,18 @@ def default_registry_path() -> Path:
     return (Path(__file__).resolve().parent / "data" / "projects.registry.json").resolve()
 
 
+def _reject_credential_keys(obj: Any, *, path: str = "") -> None:
+    if isinstance(obj, dict):
+        for key, value in obj.items():
+            key_path = f"{path}.{key}" if path else key
+            if _is_forbidden_key(key):
+                raise ValueError(f"registry must not store credentials ({key_path})")
+            _reject_credential_keys(value, path=key_path)
+    elif isinstance(obj, list):
+        for index, item in enumerate(obj):
+            _reject_credential_keys(item, path=f"{path}[{index}]")
+
+
 def _parse_onboarding(raw: list[dict[str, Any]] | None) -> tuple[OnboardingStep, ...]:
     if not raw:
         return ()

webui/registry_safety.py

@@ -1,52 +0,0 @@
-"""Shared credential-rejection guard for web UI registries (#427, #798).
-
-Registries are operator-editable declarative files that the web UI loads and,
-for the worker registry, writes back. None of them may ever carry a secret:
-credentials belong in the keychain and reach worker processes through
-environment injection, never through a file the browser layer can read.
-
-The check is structural rather than value-based on purpose. A value scanner has
-to guess what a secret looks like; a key scanner refuses the *shape* of a
-credential field, so an operator cannot introduce one by accident and a later
-loader cannot silently pass one through.
-"""
-
-from __future__ import annotations
-
-from typing import Any
-
-_FORBIDDEN_EXACT_KEYS = frozenset({
-    "token",
-    "password",
-    "secret",
-    "credential",
-    "auth",
-    "api_key",
-    "api-key",
-})
-_FORBIDDEN_KEY_PREFIXES = ("auth_", "api_key_", "api-key_")
-_FORBIDDEN_KEY_SUFFIXES = ("_token", "_secret", "_password", "_credential", "_auth")
-
-
-def is_forbidden_key(key: str) -> bool:
-    """Return True when *key* names a credential field."""
-    lowered = key.lower()
-    if lowered in _FORBIDDEN_EXACT_KEYS:
-        return True
-    return (
-        lowered.startswith(_FORBIDDEN_KEY_PREFIXES)
-        or lowered.endswith(_FORBIDDEN_KEY_SUFFIXES)
-    )
-
-
-def reject_credential_keys(obj: Any, *, path: str = "", subject: str = "registry") -> None:
-    """Raise ValueError when *obj* carries a credential-shaped key at any depth."""
-    if isinstance(obj, dict):
-        for key, value in obj.items():
-            key_path = f"{path}.{key}" if path else key
-            if is_forbidden_key(key):
-                raise ValueError(f"{subject} must not store credentials ({key_path})")
-            reject_credential_keys(value, path=key_path, subject=subject)
-    elif isinstance(obj, list):
-        for index, item in enumerate(obj):
-            reject_credential_keys(item, path=f"{path}[{index}]", subject=subject)

webui/worker_registry.py

@@ -1,647 +0,0 @@
-"""Declarative worker registry and configuration schema (#798, epic #797).
-
-The registry is the single source of truth for the scheduled multi-LLM worker
-fleet. It is a versioned JSON document holding two *separate* entity kinds:
-
-* **Providers** — the LLM runtimes a worker can be built on (Claude, Grok,
-  Codex, AGY, Kimi K). A provider describes the runtime itself: vendor,
-  executable name, models it can serve, and whether it is available on this
-  machine. Providers exist whether or not any worker uses them.
-* **Workers** — a configured *instance*: one provider, one model, one project,
-  one role, one MCP namespace/profile, one workflow, one schedule. Several
-  workers may share a provider; a worker naming an undeclared provider is
-  refused.
-
-Keeping them separate is what lets #799 list all five providers even when a
-provider currently has no configured worker, and it stops provider facts from
-being copied into (and drifting across) every worker record.
-
-Scope boundary. This module owns the data model, its validation, and its
-persistence. It does **not** schedule anything, launch anything, probe provider
-executables, or serve HTTP. Loading a registry never touches a process; the
-live fields a dashboard wants (PID, elapsed time, next run) are derived
-elsewhere (#799, #801, #803, #804) from these declarations.
-
-Safety invariants:
-
-* No credential may be stored (:mod:`webui.registry_safety`), so the registry
-  stays safe to render and to hand to a browser layer.
-* Validation fails closed. Unknown fields are refused rather than ignored, so a
-  typo cannot silently disable a timeout or a role binding.
-* Writes are atomic and every superseded document is retained as a numbered
-  revision, so a bad edit is recoverable by rollback rather than hand-repair.
-"""
-
-from __future__ import annotations
-
-import json
-import os
-import re
-import tempfile
-from dataclasses import dataclass
-from datetime import datetime, timezone
-from pathlib import Path
-from typing import Any
-
-from webui.registry_safety import reject_credential_keys
-
-SCHEMA_VERSION = 1
-
-#: Roles a worker may hold. These mirror the sanctioned MCP role kinds; a
-#: worker may not invent one, because the role selects the namespace/profile
-#: whose capability gates constrain it.
-ALLOWED_ROLES = ("author", "reviewer", "merger", "reconciler", "cleanup")
-
-#: Scheduler backends the registry can describe. ``manual`` means the worker is
-#: only ever started on request and has no recurring trigger.
-ALLOWED_SCHEDULER_KINDS = ("launchd", "manual")
-
-#: Schedule kinds. Next-run computation belongs to #803; this module only
-#: guarantees the declaration is well formed.
-ALLOWED_SCHEDULE_KINDS = ("interval", "cron", "manual")
-
-_REQUIRED_PROVIDER_FIELDS = ("id", "display_name", "vendor", "executable", "available")
-_OPTIONAL_PROVIDER_FIELDS = ("models", "notes")
-
-_REQUIRED_WORKER_FIELDS = (
-    "id",
-    "display_name",
-    "provider",
-    "model",
-    "project",
-    "role",
-    "namespace",
-    "profile",
-    "workflow",
-    "schedule",
-    "timeout_seconds",
-    "enabled",
-    "scheduler",
-)
-_OPTIONAL_WORKER_FIELDS = ("notes",)
-
-_ID_RE = re.compile(r"^[a-z0-9][a-z0-9._-]*$")
-
-#: Guards against an operator writing a timeout that would let a worker hold a
-#: lease effectively forever. 24h is far above any sanctioned cycle.
-_MAX_TIMEOUT_SECONDS = 86_400
-
-#: How many superseded revisions to retain beside the live file.
-_HISTORY_LIMIT = 20
-
-_TOP_LEVEL_FIELDS = frozenset({"version", "revision", "updated_at", "providers", "workers"})
-
-
-class RegistryValidationError(ValueError):
-    """Raised when a registry document violates the schema."""
-
-
-@dataclass(frozen=True)
-class ProviderRecord:
-    id: str
-    display_name: str
-    vendor: str
-    executable: str
-    available: bool
-    models: tuple[str, ...]
-    notes: str
-
-
-@dataclass(frozen=True)
-class ScheduleSpec:
-    kind: str
-    #: Set for ``interval`` schedules.
-    seconds: int | None
-    #: Set for ``cron`` schedules — a five-field crontab expression.
-    expression: str | None
-
-
-@dataclass(frozen=True)
-class SchedulerSpec:
-    kind: str
-    #: LaunchAgent label; required for ``launchd``, absent for ``manual``.
-    label: str | None
-
-
-@dataclass(frozen=True)
-class WorkerRecord:
-    id: str
-    display_name: str
-    provider: str
-    model: str
-    project: str
-    role: str
-    namespace: str
-    profile: str
-    workflow: str
-    schedule: ScheduleSpec
-    timeout_seconds: int
-    enabled: bool
-    scheduler: SchedulerSpec
-    notes: str
-
-
-@dataclass(frozen=True)
-class WorkerRegistry:
-    version: int
-    revision: int
-    updated_at: str
-    providers: tuple[ProviderRecord, ...]
-    workers: tuple[WorkerRecord, ...]
-    source_path: Path
-
-
-# ── paths ────────────────────────────────────────────────────────────────────
-
-
-def default_registry_path() -> Path:
-    """Location of the packaged worker registry, overridable for tests/deploys."""
-    override = os.environ.get("WEBUI_WORKER_REGISTRY", "").strip()
-    if override:
-        return Path(override).expanduser().resolve()
-    return (Path(__file__).resolve().parent / "data" / "workers.registry.json").resolve()
-
-
-def history_dir(path: Path | None = None) -> Path:
-    """Directory holding superseded revisions of *path*."""
-    source = (path or default_registry_path()).resolve()
-    return source.parent / f"{source.name}.history"
-
-
-# ── field helpers ────────────────────────────────────────────────────────────
-
-
-def _require_exact_fields(
-    raw: Any,
-    *,
-    required: tuple[str, ...],
-    optional: tuple[str, ...],
-    subject: str,
-) -> dict[str, Any]:
-    if not isinstance(raw, dict):
-        raise RegistryValidationError(f"{subject} must be an object")
-    missing = [field for field in required if field not in raw]
-    if missing:
-        raise RegistryValidationError(
-            f"{subject} missing required fields: {', '.join(sorted(missing))}"
-        )
-    unknown = sorted(set(raw) - set(required) - set(optional))
-    if unknown:
-        # Fail closed: silently dropping an unrecognized key is how a typo'd
-        # "timeout_second" ends up meaning "no timeout".
-        raise RegistryValidationError(f"{subject} has unknown fields: {', '.join(unknown)}")
-    return raw
-
-
-def _require_identifier(value: Any, *, subject: str) -> str:
-    text = str(value).strip()
-    if not _ID_RE.match(text):
-        raise RegistryValidationError(
-            f"{subject} must be lowercase alphanumeric with '.', '_', or '-' (got {value!r})"
-        )
-    return text
-
-
-def _require_text(value: Any, *, subject: str) -> str:
-    if not isinstance(value, str):
-        raise RegistryValidationError(f"{subject} must be a string (got {value!r})")
-    text = value.strip()
-    if not text:
-        raise RegistryValidationError(f"{subject} must be a non-empty string")
-    return text
-
-
-def _require_bool(value: Any, *, subject: str) -> bool:
-    if not isinstance(value, bool):
-        raise RegistryValidationError(f"{subject} must be a boolean (got {value!r})")
-    return value
-
-
-def _require_positive_int(value: Any, *, subject: str, maximum: int | None = None) -> int:
-    if isinstance(value, bool) or not isinstance(value, int):
-        raise RegistryValidationError(f"{subject} must be an integer (got {value!r})")
-    if value <= 0:
-        raise RegistryValidationError(f"{subject} must be greater than zero (got {value})")
-    if maximum is not None and value > maximum:
-        raise RegistryValidationError(f"{subject} must not exceed {maximum} (got {value})")
-    return value
-
-
-# ── parsing ──────────────────────────────────────────────────────────────────
-
-
-def _parse_provider(raw: Any) -> ProviderRecord:
-    data = _require_exact_fields(
-        raw,
-        required=_REQUIRED_PROVIDER_FIELDS,
-        optional=_OPTIONAL_PROVIDER_FIELDS,
-        subject="provider",
-    )
-    provider_id = _require_identifier(data["id"], subject="provider.id")
-
-    models_raw = data.get("models") or []
-    if not isinstance(models_raw, list):
-        raise RegistryValidationError(f"provider[{provider_id}].models must be an array")
-    models = tuple(
-        _require_text(item, subject=f"provider[{provider_id}].models[]") for item in models_raw
-    )
-
-    return ProviderRecord(
-        id=provider_id,
-        display_name=_require_text(
-            data["display_name"], subject=f"provider[{provider_id}].display_name"
-        ),
-        vendor=_require_text(data["vendor"], subject=f"provider[{provider_id}].vendor"),
-        executable=_require_text(data["executable"], subject=f"provider[{provider_id}].executable"),
-        available=_require_bool(data["available"], subject=f"provider[{provider_id}].available"),
-        models=models,
-        notes=str(data.get("notes") or "").strip(),
-    )
-
-
-def _parse_schedule(raw: Any, *, subject: str) -> ScheduleSpec:
-    if not isinstance(raw, dict):
-        raise RegistryValidationError(f"{subject} must be an object")
-    kind = _require_text(raw.get("kind"), subject=f"{subject}.kind")
-    if kind not in ALLOWED_SCHEDULE_KINDS:
-        raise RegistryValidationError(
-            f"{subject}.kind must be one of {', '.join(ALLOWED_SCHEDULE_KINDS)} (got {kind!r})"
-        )
-
-    seconds: int | None = None
-    expression: str | None = None
-
-    if kind == "interval":
-        if "seconds" not in raw:
-            raise RegistryValidationError(f"{subject}.seconds is required for interval schedules")
-        seconds = _require_positive_int(raw["seconds"], subject=f"{subject}.seconds")
-    elif kind == "cron":
-        if "expression" not in raw:
-            raise RegistryValidationError(f"{subject}.expression is required for cron schedules")
-        expression = _require_text(raw["expression"], subject=f"{subject}.expression")
-        if len(expression.split()) != 5:
-            raise RegistryValidationError(
-                f"{subject}.expression must have five crontab fields (got {expression!r})"
-            )
-
-    allowed = {"kind"}
-    if kind == "interval":
-        allowed.add("seconds")
-    elif kind == "cron":
-        allowed.add("expression")
-    unknown = sorted(set(raw) - allowed)
-    if unknown:
-        raise RegistryValidationError(
-            f"{subject} has fields not valid for kind {kind!r}: {', '.join(unknown)}"
-        )
-
-    return ScheduleSpec(kind=kind, seconds=seconds, expression=expression)
-
-
-def _parse_scheduler(raw: Any, *, subject: str) -> SchedulerSpec:
-    if not isinstance(raw, dict):
-        raise RegistryValidationError(f"{subject} must be an object")
-    kind = _require_text(raw.get("kind"), subject=f"{subject}.kind")
-    if kind not in ALLOWED_SCHEDULER_KINDS:
-        raise RegistryValidationError(
-            f"{subject}.kind must be one of {', '.join(ALLOWED_SCHEDULER_KINDS)} (got {kind!r})"
-        )
-
-    label: str | None = None
-    if kind == "launchd":
-        if "label" not in raw:
-            raise RegistryValidationError(f"{subject}.label is required for launchd schedulers")
-        label = _require_text(raw["label"], subject=f"{subject}.label")
-
-    allowed = {"kind"}
-    if kind == "launchd":
-        allowed.add("label")
-    unknown = sorted(set(raw) - allowed)
-    if unknown:
-        raise RegistryValidationError(
-            f"{subject} has fields not valid for kind {kind!r}: {', '.join(unknown)}"
-        )
-
-    return SchedulerSpec(kind=kind, label=label)
-
-
-def _parse_worker(raw: Any) -> WorkerRecord:
-    data = _require_exact_fields(
-        raw,
-        required=_REQUIRED_WORKER_FIELDS,
-        optional=_OPTIONAL_WORKER_FIELDS,
-        subject="worker",
-    )
-    worker_id = _require_identifier(data["id"], subject="worker.id")
-
-    role = _require_text(data["role"], subject=f"worker[{worker_id}].role")
-    if role not in ALLOWED_ROLES:
-        raise RegistryValidationError(
-            f"worker[{worker_id}].role must be one of {', '.join(ALLOWED_ROLES)} (got {role!r})"
-        )
-
-    return WorkerRecord(
-        id=worker_id,
-        display_name=_require_text(
-            data["display_name"], subject=f"worker[{worker_id}].display_name"
-        ),
-        provider=_require_identifier(data["provider"], subject=f"worker[{worker_id}].provider"),
-        model=_require_text(data["model"], subject=f"worker[{worker_id}].model"),
-        project=_require_text(data["project"], subject=f"worker[{worker_id}].project"),
-        role=role,
-        namespace=_require_text(data["namespace"], subject=f"worker[{worker_id}].namespace"),
-        profile=_require_text(data["profile"], subject=f"worker[{worker_id}].profile"),
-        workflow=_require_text(data["workflow"], subject=f"worker[{worker_id}].workflow"),
-        schedule=_parse_schedule(data["schedule"], subject=f"worker[{worker_id}].schedule"),
-        timeout_seconds=_require_positive_int(
-            data["timeout_seconds"],
-            subject=f"worker[{worker_id}].timeout_seconds",
-            maximum=_MAX_TIMEOUT_SECONDS,
-        ),
-        enabled=_require_bool(data["enabled"], subject=f"worker[{worker_id}].enabled"),
-        scheduler=_parse_scheduler(data["scheduler"], subject=f"worker[{worker_id}].scheduler"),
-        notes=str(data.get("notes") or "").strip(),
-    )
-
-
-def _require_unique(values: list[str], *, subject: str) -> None:
-    seen: set[str] = set()
-    for value in values:
-        if value in seen:
-            raise RegistryValidationError(f"duplicate {subject}: {value}")
-        seen.add(value)
-
-
-def validate_payload(payload: Any, *, source_path: Path) -> WorkerRegistry:
-    """Validate a decoded registry document and return the typed registry.
-
-    Raises :class:`RegistryValidationError` on any violation; never partially
-    accepts a document.
-    """
-    if not isinstance(payload, dict):
-        raise RegistryValidationError("registry root must be an object")
-
-    version = payload.get("version")
-    if version != SCHEMA_VERSION:
-        raise RegistryValidationError(f"unsupported registry version: {version!r}")
-
-    reject_credential_keys(payload, subject="worker registry")
-
-    unknown = sorted(set(payload) - _TOP_LEVEL_FIELDS)
-    if unknown:
-        raise RegistryValidationError(f"registry has unknown fields: {', '.join(unknown)}")
-
-    revision = _require_positive_int(payload.get("revision"), subject="revision")
-    updated_at = _require_text(payload.get("updated_at"), subject="updated_at")
-
-    providers_raw = payload.get("providers")
-    if not isinstance(providers_raw, list) or not providers_raw:
-        raise RegistryValidationError("providers must be a non-empty array")
-    providers = tuple(_parse_provider(item) for item in providers_raw)
-    _require_unique([provider.id for provider in providers], subject="provider id")
-
-    workers_raw = payload.get("workers")
-    if not isinstance(workers_raw, list):
-        raise RegistryValidationError("workers must be an array")
-    workers = tuple(_parse_worker(item) for item in workers_raw)
-    _require_unique([worker.id for worker in workers], subject="worker id")
-
-    # Referential integrity: a worker naming an undeclared provider would look
-    # configured while being unrunnable, which is exactly the ambiguous
-    # ownership the epic requires to fail closed.
-    known_providers = {provider.id for provider in providers}
-    for worker in workers:
-        if worker.provider not in known_providers:
-            raise RegistryValidationError(
-                f"worker[{worker.id}].provider references unknown provider {worker.provider!r}"
-            )
-
-    # A LaunchAgent label identifies a job to launchd; two workers sharing one
-    # would silently overwrite each other's agent.
-    _require_unique(
-        [worker.scheduler.label for worker in workers if worker.scheduler.label],
-        subject="scheduler label",
-    )
-
-    return WorkerRegistry(
-        version=version,
-        revision=revision,
-        updated_at=updated_at,
-        providers=providers,
-        workers=workers,
-        source_path=source_path,
-    )
-
-
-def load_registry(path: Path | None = None) -> WorkerRegistry:
-    """Load and validate the worker registry from disk."""
-    source = (path or default_registry_path()).resolve()
-    payload = json.loads(source.read_text(encoding="utf-8"))
-    return validate_payload(payload, source_path=source)
-
-
-# ── serialization ────────────────────────────────────────────────────────────
-
-
-def provider_to_dict(provider: ProviderRecord) -> dict[str, Any]:
-    return {
-        "id": provider.id,
-        "display_name": provider.display_name,
-        "vendor": provider.vendor,
-        "executable": provider.executable,
-        "available": provider.available,
-        "models": list(provider.models),
-        "notes": provider.notes,
-    }
-
-
-def _schedule_to_dict(schedule: ScheduleSpec) -> dict[str, Any]:
-    payload: dict[str, Any] = {"kind": schedule.kind}
-    if schedule.kind == "interval":
-        payload["seconds"] = schedule.seconds
-    elif schedule.kind == "cron":
-        payload["expression"] = schedule.expression
-    return payload
-
-
-def _scheduler_to_dict(scheduler: SchedulerSpec) -> dict[str, Any]:
-    payload: dict[str, Any] = {"kind": scheduler.kind}
-    if scheduler.kind == "launchd":
-        payload["label"] = scheduler.label
-    return payload
-
-
-def worker_to_dict(worker: WorkerRecord) -> dict[str, Any]:
-    return {
-        "id": worker.id,
-        "display_name": worker.display_name,
-        "provider": worker.provider,
-        "model": worker.model,
-        "project": worker.project,
-        "role": worker.role,
-        "namespace": worker.namespace,
-        "profile": worker.profile,
-        "workflow": worker.workflow,
-        "schedule": _schedule_to_dict(worker.schedule),
-        "timeout_seconds": worker.timeout_seconds,
-        "enabled": worker.enabled,
-        "scheduler": _scheduler_to_dict(worker.scheduler),
-        "notes": worker.notes,
-    }
-
-
-def registry_to_document(registry: WorkerRegistry) -> dict[str, Any]:
-    """Serialize to the on-disk document shape (no local paths embedded)."""
-    return {
-        "version": registry.version,
-        "revision": registry.revision,
-        "updated_at": registry.updated_at,
-        "providers": [provider_to_dict(provider) for provider in registry.providers],
-        "workers": [worker_to_dict(worker) for worker in registry.workers],
-    }
-
-
-def registry_to_dict(registry: WorkerRegistry) -> dict[str, Any]:
-    """Serialize for JSON API responses (adds the resolved source path)."""
-    document = registry_to_document(registry)
-    document["source_path"] = str(registry.source_path)
-    return document
-
-
-def find_worker(registry: WorkerRegistry, worker_id: str) -> WorkerRecord | None:
-    for worker in registry.workers:
-        if worker.id == worker_id:
-            return worker
-    return None
-
-
-def find_provider(registry: WorkerRegistry, provider_id: str) -> ProviderRecord | None:
-    for provider in registry.providers:
-        if provider.id == provider_id:
-            return provider
-    return None
-
-
-def workers_for_provider(registry: WorkerRegistry, provider_id: str) -> tuple[WorkerRecord, ...]:
-    return tuple(worker for worker in registry.workers if worker.provider == provider_id)
-
-
-# ── persistence ──────────────────────────────────────────────────────────────
-
-
-def _utc_now() -> str:
-    return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
-
-
-def _atomic_write(path: Path, payload: str) -> None:
-    """Write *payload* to *path* atomically: temp file in the same dir, fsync, replace."""
-    parent = path.parent
-    parent.mkdir(parents=True, exist_ok=True)
-    fd, temp_path = tempfile.mkstemp(prefix=f".{path.name}-", suffix=".tmp", 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, path)
-    finally:
-        if os.path.exists(temp_path):
-            try:
-                os.remove(temp_path)
-            except OSError:
-                pass
-
-
-def _revision_path(directory: Path, revision: int) -> Path:
-    return directory / f"rev-{revision:06d}.json"
-
-
-def _prune_history(path: Path) -> None:
-    directory = history_dir(path)
-    revisions = list_revisions(path)
-    excess = len(revisions) - _HISTORY_LIMIT
-    for revision in revisions[: max(0, excess)]:
-        _revision_path(directory, revision).unlink(missing_ok=True)
-
-
-def _archive_current(path: Path) -> int | None:
-    """Copy the live document into the history dir under its own revision number."""
-    if not path.exists():
-        return None
-    try:
-        existing = json.loads(path.read_text(encoding="utf-8"))
-        revision = int(existing.get("revision", 0))
-    except (json.JSONDecodeError, TypeError, ValueError, AttributeError):
-        # An unreadable live file has no trustworthy revision number to file it
-        # under, so it cannot join the history chain.
-        return None
-    if revision <= 0:
-        return None
-    _atomic_write(
-        _revision_path(history_dir(path), revision),
-        json.dumps(existing, indent=2, sort_keys=True) + "\n",
-    )
-    _prune_history(path)
-    return revision
-
-
-def list_revisions(path: Path | None = None) -> tuple[int, ...]:
-    """Revision numbers retained in history for *path*, oldest first."""
-    directory = history_dir(path)
-    if not directory.is_dir():
-        return ()
-    revisions: list[int] = []
-    for entry in directory.glob("rev-*.json"):
-        try:
-            revisions.append(int(entry.stem.split("-", 1)[1]))
-        except (IndexError, ValueError):
-            continue
-    return tuple(sorted(revisions))
-
-
-def save_registry(
-    registry: WorkerRegistry,
-    path: Path | None = None,
-    *,
-    updated_at: str | None = None,
-) -> WorkerRegistry:
-    """Validate, archive the superseded revision, then atomically persist a new one.
-
-    The stored revision is always the previous revision plus one, so a reader
-    can tell two documents apart even when their content is otherwise equal.
-    Returns the registry exactly as persisted.
-    """
-    target = (path or registry.source_path or default_registry_path()).resolve()
-
-    document = registry_to_document(registry)
-    # Re-validate before writing: a registry assembled in memory has not
-    # necessarily been through the loader.
-    validate_payload(document, source_path=target)
-
-    archived = _archive_current(target)
-    document["revision"] = (archived + 1) if archived is not None else registry.revision
-    document["updated_at"] = updated_at or _utc_now()
-
-    persisted = validate_payload(document, source_path=target)
-    _atomic_write(target, json.dumps(document, indent=2, sort_keys=True) + "\n")
-    return persisted
-
-
-def rollback_to_revision(revision: int, path: Path | None = None) -> WorkerRegistry:
-    """Restore a retained *revision* as a new head revision.
-
-    History is append-only: rolling back does not delete the revisions in
-    between, it republishes the chosen one under the next revision number, so a
-    rollback is itself reversible.
-    """
-    target = (path or default_registry_path()).resolve()
-    snapshot_path = _revision_path(history_dir(target), revision)
-    if not snapshot_path.exists():
-        available = ", ".join(str(item) for item in list_revisions(target)) or "(none)"
-        raise RegistryValidationError(
-            f"revision {revision} is not retained for {target.name}; available: {available}"
-        )
-
-    payload = json.loads(snapshot_path.read_text(encoding="utf-8"))
-    restored = validate_payload(payload, source_path=target)
-    return save_registry(restored, target)