feat(webui): worker registry and configuration schema (Closes #798)

Add the declarative worker registry that epic #797 makes the source of
truth for the scheduled multi-LLM worker fleet.

Providers and configured workers are modelled as separate entities so a
provider can be listed with no worker configured, and so provider facts
are not copied into every worker record. A worker records provider,
model, project, role, namespace, profile, workflow, schedule, timeout,
enabled state, and scheduler metadata.

Validation fails closed: unknown fields are refused rather than ignored,
so a typo cannot silently disable a timeout; a worker naming an
undeclared provider is rejected; worker ids, provider ids, and
LaunchAgent labels must be unique.

Persistence is atomic (temp file in the same directory, fsync, replace).
Every superseded document is retained as a numbered revision, and
rollback republishes a chosen revision as a new head, so history stays
append-only and a rollback is itself reversible.

The credential-rejection guard is extracted to webui/registry_safety.py
so both registries share one implementation instead of two copies of a
security check; project_registry.py keeps identical behaviour.

Scope: data model, validation, persistence only. No routes, scheduler,
process control, or provider probing - those are #799/#800/#804/#805.
The workers array ships empty because populating it is #808.

Tests: tests/test_webui_worker_registry.py, 44 cases.

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>

gitea_mcp_server.py

@@ -2017,6 +2017,7 @@ import issue_lock_provenance  # noqa: E402
 import issue_lock_store  # noqa: E402
 import issue_lock_adoption  # noqa: E402
 import issue_lock_recovery  # noqa: E402
+import issue_lock_renewal  # noqa: E402
 import stacked_pr_support  # noqa: E402
 import merge_approval_gate  # noqa: E402
 import review_quarantine  # noqa: E402  # #695 contaminated formal-review quarantine
@@ -2316,7 +2317,12 @@ def _resolve_issue_lock_for_pr(
     return lock_data
 
 
-def _save_issue_lock(data: dict, *, expected_generation: int | None = None) -> str:
+def _save_issue_lock(
+    data: dict,
+    *,
+    expected_generation: int | None = None,
+    renewal_sanctioned: bool = False,
+) -> str:
     existing = issue_lock_store.load_issue_lock(
         remote=str(data.get("remote") or ""),
         org=str(data.get("org") or ""),
@@ -2328,7 +2334,9 @@ def _save_issue_lock(data: dict, *, expected_generation: int | None = None) -> s
         raise RuntimeError(overwrite_block)
     try:
         return issue_lock_store.bind_session_lock(
-            data, expected_generation=expected_generation
+            data,
+            expected_generation=expected_generation,
+            renewal_sanctioned=renewal_sanctioned,
         )
     except Exception as e:
         raise RuntimeError(f"Could not write issue lock file: {e}") from e
@@ -2448,6 +2456,79 @@ def _evaluate_issue_lock_recovery(
     )
 
 
+def _evaluate_issue_lock_renewal(
+    existing_lock: dict,
+    *,
+    issue_number: int,
+    branch_name: str,
+    worktree_path: str,
+    remote: str,
+    h: str | None,
+    o: str,
+    r: str,
+    git_state: dict,
+) -> dict:
+    """Gather evidence and decide exact-owner renewal of an expired lease (#760).
+
+    Mirrors ``_evaluate_issue_lock_recovery``: every input is durable lock state
+    or a live server-side observation (Gitea branch/PR inventory, git in the
+    declared worktree, the local lock store). Nothing is reachable from an MCP
+    caller's parameters, so no caller can assert its way into a renewal
+    (#760 AC14).
+    """
+    renewal_auth = _auth(h)
+    try:
+        renewal_branches = api_get_all(
+            f"{repo_api_url(h, o, r)}/branches", renewal_auth
+        )
+    except Exception as exc:
+        raise RuntimeError(
+            f"Could not list branches to verify exact-owner lease renewal: {exc}"
+        )
+
+    remote_head: str | None = None
+    candidates: list[str] = []
+    for entry in renewal_branches:
+        entry_name = _branch_entry_name(entry)
+        if entry_name == branch_name:
+            remote_head = _branch_entry_commit_sha(entry)
+        if issue_lock_adoption.branch_carries_issue_marker(entry_name, issue_number):
+            candidates.append(entry_name)
+
+    pr_head: str | None = None
+    pr_number: int | None = None
+    for pull in _list_open_pulls(h, o, r, renewal_auth):
+        pull_head = pull.get("head") or {}
+        if str(pull_head.get("ref") or "") == branch_name:
+            pr_head = pull_head.get("sha")
+            pr_number = pull.get("number")
+            break
+
+    claimant = _work_lease_claimant(h)
+    return issue_lock_renewal.assess_exact_owner_lease_renewal(
+        existing_lock,
+        issue_number=issue_number,
+        branch_name=branch_name,
+        worktree_path=worktree_path,
+        remote=remote,
+        org=o,
+        repo=r,
+        identity=claimant.get("username"),
+        profile=claimant.get("profile"),
+        operation_type=AUTHOR_ISSUE_WORK_LEASE,
+        current_branch=git_state.get("current_branch"),
+        porcelain_status=git_state.get("porcelain_status") or "",
+        worktree_exists=os.path.isdir(os.path.realpath(worktree_path)),
+        head_sha=git_state.get("head_sha"),
+        remote_head_sha=remote_head,
+        pr_head_sha=pr_head,
+        pr_number=pr_number,
+        competing_live_locks=issue_lock_store.list_live_locks(),
+        candidate_branches=candidates,
+        current_pid=os.getpid(),
+    )
+
+
 def _work_lease_claimant(host: str | None) -> dict:
     profile = get_profile()
     username = _IDENTITY_CACHE.get(host) if host else None
@@ -3893,15 +3974,12 @@ def gitea_lock_issue(
     existing_issue_lock = _load_existing_issue_lock(
         remote=remote, org=o, repo=r, issue_number=issue_number
     )
-    active_lease_block = issue_lock_store.assess_same_issue_lease_conflict(
-        existing_issue_lock,
-        issue_number=issue_number,
-        branch_name=branch_name,
-        worktree_path=resolved_worktree,
-        operation_type=AUTHOR_ISSUE_WORK_LEASE,
-    )
-    if active_lease_block:
-        raise RuntimeError(active_lease_block)
+    # #760: the competing-lease disposition is decided below, once the worktree
+    # and Gitea evidence an exact-owner renewal depends on has actually been
+    # observed. Deciding it here — before any of that exists — is what made the
+    # same-owner allowance unreachable for an expired lease. The authoritative
+    # check still runs inside bind_session_lock under the per-issue flock, so
+    # moving this one later cannot widen the window for a competing writer.
 
     # ── Stacked-PR base declaration (opt-in, #484) ──
     # Normal work leaves stacked_base_branch None → master-equivalent path.
@@ -3967,6 +4045,53 @@ def gitea_lock_issue(
     recovery_sanctioned = bool(
         recovery_assessment and recovery_assessment.get("recovery_sanctioned")
     )
+
+    # ── Exact-owner renewal of an expired lease (#760) ──
+    # The opposite trigger from #753 above: there the lease is unexpired and the
+    # PID is dead; here the lease has expired while the recording daemon — which
+    # is the long-lived MCP server, not the authoring task — may well still be
+    # up. Only an expired lease is assessed, so a live foreign lease is never a
+    # candidate (AC12) and dead-PID takeover keeps its existing conditions
+    # (AC11). A refusal never raises: it withholds the waiver and leaves the
+    # conflict check below to fail closed exactly as before.
+    renewal_assessment: dict | None = None
+    if (
+        existing_issue_lock
+        and existing_issue_lock.get("issue_number") == issue_number
+        and issue_lock_store.is_lease_expired(existing_issue_lock)
+    ):
+        renewal_assessment = _evaluate_issue_lock_renewal(
+            existing_issue_lock,
+            issue_number=issue_number,
+            branch_name=branch_name,
+            worktree_path=resolved_worktree,
+            remote=remote,
+            h=h,
+            o=o,
+            r=r,
+            git_state=git_state,
+        )
+    renewal_sanctioned = bool(
+        renewal_assessment and renewal_assessment.get("renewal_sanctioned")
+    )
+
+    active_lease_block = issue_lock_store.assess_same_issue_lease_conflict(
+        existing_issue_lock,
+        issue_number=issue_number,
+        branch_name=branch_name,
+        worktree_path=resolved_worktree,
+        operation_type=AUTHOR_ISSUE_WORK_LEASE,
+        renewal_sanctioned=renewal_sanctioned,
+    )
+    if active_lease_block:
+        reasons = [active_lease_block]
+        # Name the exact missing evidence when this looked like a renewal, so a
+        # blocked owner sees why rather than only the generic takeover text.
+        if renewal_assessment and renewal_assessment.get("is_candidate"):
+            reasons.append(
+                issue_lock_renewal.format_renewal_refusal(renewal_assessment)
+            )
+        raise RuntimeError("; ".join(reasons))
     # #755: a sanctioned dead-session recovery always has an owning open PR —
     # that is what makes it a recovery rather than a fresh claim. Carry the
     # server-derived owning-PR evidence into the duplicate-work gate below so
@@ -3978,6 +4103,14 @@ def gitea_lock_issue(
         if recovery_sanctioned
         else None
     )
+    # #760: a sanctioned renewal owns its open PR for the same reason, so it
+    # needs the same exemption. Without it the duplicate-work gate rejects every
+    # renewal with "open PR already covers issue", which is the PR the lock
+    # being renewed already owns. Withheld unless renewal was granted.
+    if recovered_owning_pr is None and renewal_sanctioned:
+        recovered_owning_pr = issue_lock_renewal.owning_pr_renewal_evidence(
+            renewal_assessment
+        )
     lock_assessment = issue_lock_worktree.assess_issue_lock_worktree(
         worktree_path=resolved_worktree,
         current_branch=git_state.get("current_branch"),
@@ -3986,6 +4119,10 @@ def gitea_lock_issue(
         inspected_git_root=git_state.get("inspected_git_root"),
         base_branch=git_state.get("base_branch"),
         recovery_sanctioned=recovery_sanctioned,
+        # #760: without this the renewal waiver was computed and then discarded
+        # here — the exact-owner branch always carries commits, so it can never
+        # be base-equivalent, and every real renewal failed at this gate.
+        renewal_sanctioned=renewal_sanctioned,
     )
     if lock_assessment["block"]:
         reasons = list(lock_assessment.get("reasons") or [])
@@ -3995,6 +4132,12 @@ def gitea_lock_issue(
             reasons.append(
                 issue_lock_recovery.format_recovery_refusal(recovery_assessment)
             )
+        # #760: same courtesy for a refused renewal, so an exact owner blocked
+        # at this gate sees which piece of ownership evidence was missing.
+        if renewal_assessment and renewal_assessment.get("is_candidate"):
+            reasons.append(
+                issue_lock_renewal.format_renewal_refusal(renewal_assessment)
+            )
         raise RuntimeError(
             issue_lock_worktree.format_issue_lock_worktree_error(
                 {**lock_assessment, "reasons": reasons}
@@ -4070,18 +4213,35 @@ def gitea_lock_issue(
             recovery_assessment,
             recovered_at=_work_lease_timestamp(_work_lease_now()),
         )
+    if renewal_sanctioned and renewal_assessment:
+        # #760 AC9: record both sides of the transition — prior PID and expiry,
+        # replacement PID and new expiry — so a renewed lock is auditable and
+        # never reads as an original claim.
+        data["lease_renewal"] = issue_lock_renewal.build_renewal_record(
+            renewal_assessment,
+            renewed_at=_work_lease_timestamp(_work_lease_now()),
+            new_expires_at=str(work_lease.get("expires_at") or ""),
+        )
 
     # #772 AC5: a recovery replaces a claim another session already owned, so
     # its write is a compare-and-swap against the generation the assessment was
     # made on. Two replacement sessions that both observed the same dead owner
     # cannot both succeed — the second finds a moved generation and fails
     # closed. Ordinary first-time claims keep the unconditional write.
+    # #760 uses the same compare-and-swap: a renewal also replaces a claim that
+    # already existed on disk, so two sessions that both observed the same
+    # expired lease cannot both win — the second finds a moved generation and
+    # fails closed.
     expected_generation = (
         issue_lock_store.lock_generation(existing_issue_lock)
-        if recovery_sanctioned
+        if (recovery_sanctioned or renewal_sanctioned)
         else None
     )
-    lock_file_path = _save_issue_lock(data, expected_generation=expected_generation)
+    lock_file_path = _save_issue_lock(
+        data,
+        expected_generation=expected_generation,
+        renewal_sanctioned=renewal_sanctioned,
+    )
     lock_record = issue_lock_store.read_lock_file(lock_file_path) or data
     freshness = issue_lock_store.assess_lock_freshness(lock_record)
     competing = [
@@ -4150,6 +4310,16 @@ def gitea_lock_issue(
             issue_number=issue_number,
             branch_name=branch_name,
         )
+    if renewal_sanctioned:
+        # #760 AC13: the renewal is visible in the native tool result, so an
+        # owner never has to inspect the lock file to confirm what happened.
+        result["lease_renewal"] = lock_record.get("lease_renewal")
+        result["message"] = (
+            f"Renewed the expired {AUTHOR_ISSUE_WORK_LEASE} lease on issue "
+            f"#{issue_number} for its exact recorded owner, on branch "
+            f"'{branch_name}' from worktree '{resolved_worktree}' "
+            "(fail-closed check complete)."
+        )
     if agent_artifacts:
         result["warnings"] = [
             "Agent temp artifacts at repo root (delete before implementation): "
@@ -12366,10 +12536,39 @@ 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 the on-disk master HEAD (#420)."""
+    """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.
+    """
     current_head = master_parity_gate.read_git_head(PROJECT_ROOT)
-    return master_parity_gate.assess_master_parity(_STARTUP_PARITY, current_head)
+    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)
 
 
 def _current_runtime_mode_report(refresh: bool = False) -> dict:
@@ -16310,10 +16509,29 @@ 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["stale"] and not master_parity_gate.gate_disabled():
+    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. "
@@ -16367,6 +16585,13 @@ 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"],
@@ -16383,7 +16608,7 @@ def gitea_assess_master_parity(
             source=canonical_source,
         ),
     }
-    if parity["stale"] and enforced:
+    if parity["restart_required"] and enforced:
         out["report"] = master_parity_gate.parity_report(parity)
     return out
 

issue_lock_renewal.py

@@ -0,0 +1,481 @@
+"""Exact-owner renewal of an expired author issue lease (#760).
+
+An author issue lease carries an absolute wall-clock expiry stamped once at
+lock time. The PID recorded alongside it is the long-lived MCP daemon, not the
+authoring task, so a lease that expires while its daemon is still up is the
+ordinary case for any author task that outlives the TTL — not an anomaly.
+
+Before this module, that case was unreachable.
+``issue_lock_store.assess_same_issue_lease_conflict`` computed same-owner
+evidence and then returned on the expired branch before consulting it, and
+``assess_expired_lock_reclaim`` only permits takeover on a dead PID or a
+missing worktree. An exact owner whose daemon is alive and whose worktree is
+present satisfied neither, so its own lock became permanently unmodifiable
+through sanctioned tools.
+
+This module is the pure evidence assessor for that one narrow case. It answers
+a single question: may *this* session renew a lease it can prove it already
+owns? It performs no mutation and no network I/O, and it never trusts a caller
+assertion — every field is compared against durable lock state or a live
+observation supplied by the caller and gathered server-side.
+
+Deliberate boundaries:
+
+* **Renewal is not takeover.** A refusal here never widens what
+  ``assess_expired_lock_reclaim`` already allows; foreign expired locks keep
+  requiring a dead PID or missing worktree (#760 AC11), and a *live* foreign
+  lease stays non-recoverable by construction because only an expired lease is
+  ever a candidate (AC12).
+* **PID liveness is never authorization.** A live recorded PID proves the
+  daemon is up, nothing more. It is recorded as evidence and is neither
+  necessary nor sufficient for renewal (AC16).
+* **Absolute expiry is preserved.** Renewal issues a new absolute expiry from
+  the moment of the write. It does not introduce sliding heartbeat renewal,
+  lease generations as fencing tokens, or a shared cross-role lifecycle — that
+  is #790's scope and is deliberately not implemented here.
+"""
+
+from __future__ import annotations
+
+import os
+from typing import Any, Iterable, Mapping, Sequence
+
+from issue_lock_store import AUTHOR_ISSUE_WORK_LEASE, is_lease_expired, is_process_alive
+from reviewer_worktree import parse_dirty_tracked_files
+
+# Outcome values.
+RENEWAL_SANCTIONED = "RENEWAL_SANCTIONED"
+NO_CANDIDATE = "NO_CANDIDATE"
+REFUSED = "REFUSED"
+
+# Durable fields a lock must carry before it can be considered at all.
+REQUIRED_LOCK_FIELDS = ("issue_number", "branch_name", "worktree_path")
+
+
+def _text(value: Any) -> str:
+    return str(value or "").strip()
+
+
+def _same_realpath(left: str | None, right: str | None) -> bool:
+    if not left or not right:
+        return False
+    try:
+        return os.path.realpath(left) == os.path.realpath(right)
+    except OSError:
+        return left == right
+
+
+def _lock_claimant(lock: Mapping[str, Any]) -> dict[str, Any]:
+    claimant = lock.get("claimant")
+    if not isinstance(claimant, Mapping):
+        lease = lock.get("work_lease")
+        claimant = lease.get("claimant") if isinstance(lease, Mapping) else None
+    return dict(claimant) if isinstance(claimant, Mapping) else {}
+
+
+def _lock_lease(lock: Mapping[str, Any]) -> dict[str, Any]:
+    lease = lock.get("work_lease")
+    return dict(lease) if isinstance(lease, Mapping) else {}
+
+
+def _lock_operation_type(lock: Mapping[str, Any]) -> str:
+    lease = _lock_lease(lock)
+    return _text(lease.get("operation_type")) or AUTHOR_ISSUE_WORK_LEASE
+
+
+def _recorded_pid(lock: Mapping[str, Any]) -> Any:
+    pid = lock.get("session_pid")
+    if pid is None:
+        pid = lock.get("pid")
+    return pid
+
+
+def _malformed_reasons(lock: Mapping[str, Any]) -> list[str]:
+    """Names of durable fields that are missing or unusable."""
+    missing: list[str] = []
+    for field in REQUIRED_LOCK_FIELDS:
+        if not _text(lock.get(field)):
+            missing.append(field)
+    pid = _recorded_pid(lock)
+    if pid is None or _text(pid) == "":
+        missing.append("session_pid/pid")
+    else:
+        try:
+            if int(pid) <= 0:
+                missing.append("session_pid/pid")
+        except (TypeError, ValueError):
+            missing.append("session_pid/pid")
+    return missing
+
+
+def _competing_lock_reasons(
+    competing_live_locks: Iterable[Mapping[str, Any]] | None,
+    *,
+    issue_number: int,
+    branch_name: str,
+    worktree_path: str,
+) -> list[str]:
+    """Live locks that would contend with this renewal (#760 AC7).
+
+    A live lock on the *same* issue cannot coexist with this expired lease, so
+    any live entry naming this issue, branch, or worktree belongs to somebody
+    else and refuses the renewal.
+    """
+    reasons: list[str] = []
+    for entry in competing_live_locks or ():
+        if not isinstance(entry, Mapping):
+            continue
+        entry_issue = entry.get("issue_number")
+        entry_branch = _text(entry.get("branch_name"))
+        entry_worktree = _text(entry.get("worktree_path"))
+        if entry_issue == issue_number:
+            reasons.append(
+                f"a live lock already exists for issue #{issue_number} "
+                f"(pid {entry.get('pid')}); renewal would contend with it"
+            )
+            continue
+        if entry_branch and entry_branch == _text(branch_name):
+            reasons.append(
+                f"live lock for issue #{entry_issue} already holds branch "
+                f"'{branch_name}'"
+            )
+        if entry_worktree and _same_realpath(entry_worktree, worktree_path):
+            reasons.append(
+                f"live lock for issue #{entry_issue} already holds worktree "
+                f"'{worktree_path}'"
+            )
+    return reasons
+
+
+def assess_exact_owner_lease_renewal(
+    existing_lock: Mapping[str, Any] | None,
+    *,
+    issue_number: int,
+    branch_name: str,
+    worktree_path: str,
+    remote: str,
+    org: str,
+    repo: str,
+    identity: str | None,
+    profile: str | None,
+    operation_type: str = AUTHOR_ISSUE_WORK_LEASE,
+    current_branch: str | None = None,
+    porcelain_status: str = "",
+    worktree_exists: bool = False,
+    head_sha: str | None = None,
+    remote_head_sha: str | None = None,
+    pr_head_sha: str | None = None,
+    pr_number: int | None = None,
+    competing_live_locks: Sequence[Mapping[str, Any]] | None = None,
+    candidate_branches: Sequence[str] | None = None,
+    current_pid: int | None = None,
+    now: Any = None,
+) -> dict[str, Any]:
+    """Decide whether an expired lease may be renewed by its exact owner.
+
+    Returns a disposition dict; it never raises and never mutates. A refusal
+    withholds permission, leaving every pre-existing guard to fail closed
+    exactly as before — this assessment can only ever *add* permission.
+
+    ``NO_CANDIDATE`` means the situation is not an exact-owner renewal at all
+    (no lock, different issue, different operation, or an unexpired lease) and
+    the caller should carry on with its normal path. ``REFUSED`` means it looked
+    like one but the evidence did not hold, and ``reasons`` names exactly what
+    was missing.
+    """
+    evidence: dict[str, Any] = {
+        "issue_number": issue_number,
+        "branch_name": branch_name,
+        "worktree_path": worktree_path,
+        "remote": remote,
+        "org": org,
+        "repo": repo,
+        "operation_type": operation_type,
+        "identity": identity,
+        "profile": profile,
+    }
+
+    def _result(outcome: str, reasons: list[str], **extra: Any) -> dict[str, Any]:
+        return {
+            "outcome": outcome,
+            "renewal_sanctioned": outcome == RENEWAL_SANCTIONED,
+            "is_candidate": outcome in (RENEWAL_SANCTIONED, REFUSED),
+            "reasons": reasons,
+            "evidence": {**evidence, **extra},
+        }
+
+    if not isinstance(existing_lock, Mapping) or not existing_lock:
+        return _result(NO_CANDIDATE, ["no existing lock to renew"])
+
+    if existing_lock.get("issue_number") != issue_number:
+        return _result(
+            NO_CANDIDATE,
+            [
+                f"existing lock is for issue #{existing_lock.get('issue_number')}, "
+                f"not #{issue_number}"
+            ],
+        )
+
+    existing_operation = _lock_operation_type(existing_lock)
+    if existing_operation != operation_type:
+        return _result(
+            NO_CANDIDATE,
+            [
+                f"existing lease operation '{existing_operation}' is not "
+                f"'{operation_type}'"
+            ],
+        )
+
+    # Only an *expired* lease is ever a renewal candidate. An unexpired lease —
+    # live, or stale by dead PID — is somebody else's problem: the first needs no
+    # renewal, and the second is #753's dead-session recovery. This is also what
+    # makes a live foreign lease non-recoverable here (#760 AC12).
+    if not is_lease_expired(existing_lock, now=now):
+        return _result(
+            NO_CANDIDATE,
+            ["lease has not expired; renewal does not apply"],
+        )
+
+    malformed = _malformed_reasons(existing_lock)
+    if malformed:
+        return _result(
+            REFUSED,
+            ["durable lock is missing or has unusable fields: " + ", ".join(malformed)],
+        )
+
+    lease = _lock_lease(existing_lock)
+    claimant = _lock_claimant(existing_lock)
+    recorded_pid = _recorded_pid(existing_lock)
+    prior_expires_at = _text(lease.get("expires_at"))
+
+    # #760 AC16: recorded purely as evidence. A live daemon PID is neither
+    # necessary nor sufficient for renewal, and nothing below branches on it.
+    recorded_pid_alive = is_process_alive(recorded_pid)
+
+    extra: dict[str, Any] = {
+        "prior_pid": recorded_pid,
+        "prior_pid_alive": recorded_pid_alive,
+        "prior_expires_at": prior_expires_at,
+        "replacement_pid": current_pid,
+        "recorded_claimant": claimant,
+        "head_sha": head_sha,
+        "remote_head_sha": remote_head_sha,
+        "pr_head_sha": pr_head_sha,
+        "pr_number": pr_number,
+    }
+
+    reasons: list[str] = []
+
+    # ── AC3: exact ownership identity ──
+    if _text(existing_lock.get("remote")) != _text(remote):
+        reasons.append(
+            f"recorded remote '{existing_lock.get('remote')}' does not match "
+            f"'{remote}'"
+        )
+    if _text(existing_lock.get("org")) != _text(org):
+        reasons.append(
+            f"recorded org '{existing_lock.get('org')}' does not match '{org}'"
+        )
+    if _text(existing_lock.get("repo")) != _text(repo):
+        reasons.append(
+            f"recorded repo '{existing_lock.get('repo')}' does not match '{repo}'"
+        )
+    if _text(existing_lock.get("branch_name")) != _text(branch_name):
+        reasons.append(
+            f"recorded branch '{existing_lock.get('branch_name')}' does not match "
+            f"'{branch_name}'"
+        )
+    if not _same_realpath(_text(existing_lock.get("worktree_path")), worktree_path):
+        reasons.append(
+            f"recorded worktree '{existing_lock.get('worktree_path')}' does not "
+            f"match '{worktree_path}'"
+        )
+
+    recorded_identity = _text(claimant.get("username"))
+    recorded_profile = _text(claimant.get("profile"))
+    if not recorded_identity or not recorded_profile:
+        reasons.append(
+            "durable lock does not record both a claimant username and profile"
+        )
+    if recorded_identity and recorded_identity != _text(identity):
+        reasons.append(
+            f"recorded claimant '{recorded_identity}' does not match active "
+            f"identity '{_text(identity) or 'unknown'}'"
+        )
+    if recorded_profile and recorded_profile != _text(profile):
+        reasons.append(
+            f"recorded profile '{recorded_profile}' does not match active profile "
+            f"'{_text(profile) or 'unknown'}'"
+        )
+
+    # ── AC4: the registered worktree still exists, is on the branch, and is clean ──
+    if not worktree_exists:
+        reasons.append(f"declared worktree '{worktree_path}' does not exist")
+    if _text(current_branch) != _text(branch_name):
+        reasons.append(
+            f"worktree is on branch '{_text(current_branch) or 'unknown'}', not "
+            f"'{branch_name}'"
+        )
+    dirty = parse_dirty_tracked_files(porcelain_status or "")
+    if dirty:
+        reasons.append(
+            "worktree has uncommitted tracked changes: " + ", ".join(sorted(dirty))
+        )
+
+    # ── AC5/AC6: published heads must agree ──
+    if not _text(head_sha):
+        reasons.append("local head could not be observed")
+    if not _text(remote_head_sha):
+        reasons.append(
+            "remote branch head could not be observed; an unpublished branch "
+            "cannot prove exact-owner renewal"
+        )
+    if _text(head_sha) and _text(remote_head_sha) and head_sha != remote_head_sha:
+        reasons.append(
+            f"local head {head_sha} does not equal remote head {remote_head_sha}"
+        )
+    if pr_number is not None:
+        if not _text(pr_head_sha):
+            reasons.append(f"owning PR #{pr_number} head could not be observed")
+        elif _text(head_sha) and pr_head_sha != head_sha:
+            reasons.append(
+                f"owning PR #{pr_number} head {pr_head_sha} does not equal local "
+                f"head {head_sha}"
+            )
+
+    # ── AC7: nothing else claims this work ──
+    reasons.extend(
+        _competing_lock_reasons(
+            competing_live_locks,
+            issue_number=issue_number,
+            branch_name=branch_name,
+            worktree_path=worktree_path,
+        )
+    )
+    other_branches = [
+        name
+        for name in (candidate_branches or ())
+        if _text(name) and _text(name) != _text(branch_name)
+    ]
+    if other_branches:
+        reasons.append(
+            "other branches already carry this issue marker: "
+            + ", ".join(sorted(other_branches))
+        )
+
+    if reasons:
+        return _result(REFUSED, reasons, **extra)
+
+    return _result(
+        RENEWAL_SANCTIONED,
+        [
+            f"exact owner '{recorded_identity}' ({recorded_profile}) proved "
+            f"ownership of issue #{issue_number} on branch '{branch_name}' from "
+            f"worktree '{worktree_path}'; local, remote"
+            + (f", and PR #{pr_number}" if pr_number is not None else "")
+            + f" heads all equal {head_sha}; lease expired at "
+            f"{prior_expires_at or 'unknown'}"
+        ],
+        **extra,
+    )
+
+
+def owning_pr_renewal_evidence(
+    assessment: Mapping[str, Any] | None,
+) -> dict[str, Any] | None:
+    """Server-derived proof of the open PR a sanctioned renewal already owns.
+
+    The mirror of ``issue_lock_recovery.owning_pr_recovery_evidence`` (#755) for
+    the renewal disposition. An exact-owner renewal of a published branch is, by
+    construction, renewal of work that already has an open PR — so the
+    duplicate-work gate's linked-open-PR blocker would otherwise discard every
+    sanctioned renewal, exactly as it once discarded every sanctioned recovery.
+
+    Returns ``None`` unless renewal was actually granted and the evidence names
+    one owning PR whose head agrees with both the local and remote heads the
+    assessor accepted. Nothing is caller-supplied: every field is copied from
+    evidence built out of durable lock state plus live git/Gitea observation.
+
+    Renewal has no descendant case — it requires the local, remote, and PR heads
+    to be equal — so there is only one head to report.
+    """
+    if not isinstance(assessment, Mapping):
+        return None
+    if assessment.get("outcome") != RENEWAL_SANCTIONED:
+        return None
+    if not assessment.get("renewal_sanctioned"):
+        return None
+
+    evidence = assessment.get("evidence") or {}
+    branch_name = _text(evidence.get("branch_name"))
+    pr_head = _text(evidence.get("pr_head_sha"))
+    local_head = _text(evidence.get("head_sha"))
+    remote_head = _text(evidence.get("remote_head_sha"))
+    raw_pr_number = evidence.get("pr_number")
+
+    if raw_pr_number is None or not branch_name or not pr_head:
+        return None
+    # The assessor already required these to agree. Re-check, so a truncated or
+    # hand-built evidence map can never authorize an exemption.
+    if pr_head != local_head or pr_head != remote_head:
+        return None
+    try:
+        pr_number = int(raw_pr_number)
+        issue_number = int(evidence.get("issue_number"))
+    except (TypeError, ValueError):
+        return None
+
+    return {
+        "issue_number": issue_number,
+        "pr_number": pr_number,
+        "branch_name": branch_name,
+        "head_sha": pr_head,
+        "recorded_head": pr_head,
+        "accepted_head": pr_head,
+        "head_relation": "equal",
+    }
+
+
+def build_renewal_record(
+    assessment: Mapping[str, Any] | None,
+    *,
+    renewed_at: str,
+    new_expires_at: str,
+) -> dict[str, Any]:
+    """Durable audit record for a sanctioned renewal (#760 AC9).
+
+    Records both sides of the transition — prior PID and expiry, replacement PID
+    and new expiry — so a renewed lock is never mistakable for an original
+    claim, and so the evidence the waiver was granted on stays inspectable.
+    """
+    data = dict(assessment or {})
+    evidence = dict(data.get("evidence") or {})
+    recorded_claimant = dict(evidence.get("recorded_claimant") or {})
+    return {
+        "renewed": bool(data.get("renewal_sanctioned")),
+        "renewed_at": renewed_at,
+        "prior_pid": evidence.get("prior_pid"),
+        "prior_pid_alive": evidence.get("prior_pid_alive"),
+        "prior_expires_at": evidence.get("prior_expires_at"),
+        "replacement_pid": evidence.get("replacement_pid"),
+        "new_expires_at": new_expires_at,
+        "identity": recorded_claimant.get("username"),
+        "profile": recorded_claimant.get("profile"),
+        "branch_name": evidence.get("branch_name"),
+        "worktree_path": evidence.get("worktree_path"),
+        "head_sha": evidence.get("head_sha"),
+        "remote_head_sha": evidence.get("remote_head_sha"),
+        "pr_head_sha": evidence.get("pr_head_sha"),
+        "pr_number": evidence.get("pr_number"),
+        "reason": "expired lease renewed by its exact recorded owner",
+        "proof": list(data.get("reasons") or []),
+    }
+
+
+def format_renewal_refusal(assessment: Mapping[str, Any] | None) -> str:
+    """One-line refusal summary for a blocked caller."""
+    data = dict(assessment or {})
+    reasons = list(data.get("reasons") or [])
+    if not reasons:
+        return "exact-owner lease renewal was not available (no evidence recorded)"
+    return "exact-owner lease renewal refused: " + "; ".join(reasons)

issue_lock_store.py

@@ -168,6 +168,7 @@ def bind_session_lock(
     lock_dir: str | None = None,
     *,
     expected_generation: int | None = None,
+    renewal_sanctioned: bool = False,
 ) -> str:
     """Persist a keyed lock and bind it to the current process session.
 
@@ -220,6 +221,7 @@ def bind_session_lock(
                 issue_number=issue_number,
                 branch_name=str(record.get("branch_name") or ""),
                 worktree_path=str(record.get("worktree_path") or ""),
+                renewal_sanctioned=renewal_sanctioned,
             )
             if lease_block:
                 raise RuntimeError(lease_block)
@@ -483,9 +485,19 @@ def assess_same_issue_lease_conflict(
     branch_name: str,
     worktree_path: str,
     operation_type: str = AUTHOR_ISSUE_WORK_LEASE,
+    renewal_sanctioned: bool = False,
     now: datetime | None = None,
 ) -> str | None:
-    """Return a fail-closed error when a competing live lease blocks acquisition."""
+    """Return a fail-closed error when a competing live lease blocks acquisition.
+
+    ``renewal_sanctioned`` is set only when
+    ``issue_lock_renewal.assess_exact_owner_lease_renewal`` has already proven,
+    from the durable lock plus live server-side observation, that this session
+    is the exact recorded owner of an *expired* lease (#760). It is never a
+    caller-supplied parameter of any MCP tool (#760 AC14): the server computes
+    it and passes it down. Left False, every pre-existing disposition is
+    unchanged.
+    """
     if not existing_lock:
         return None
 
@@ -506,6 +518,16 @@ def assess_same_issue_lease_conflict(
         and _same_realpath(str(existing_worktree or ""), worktree_path)
     )
     if is_lease_expired(existing_lock, now=now):
+        # #760 AC1/AC2: exact-owner renewal is a different disposition from
+        # foreign takeover and is evaluated first. Before this, both branches
+        # below returned unconditionally, so the same_owner allowance further
+        # down was unreachable for every expired lease — an owner could never
+        # renew its own lock once the wall clock passed, no matter how complete
+        # its ownership evidence. Requires BOTH the locally recomputed
+        # same_owner match and the server-proven renewal waiver; either alone is
+        # insufficient.
+        if same_owner and renewal_sanctioned:
+            return None
         reclaim = assess_expired_lock_reclaim(existing_lock, now=now)
         if reclaim.get("reclaim_allowed"):
             # #601: expired + dead pid / missing worktree may be reclaimed

issue_lock_worktree.py

@@ -285,6 +285,7 @@ def assess_issue_lock_worktree(
     base_branch: str | None = None,
     base_branches: frozenset[str] | None = None,
     recovery_sanctioned: bool = False,
+    renewal_sanctioned: bool = False,
 ) -> dict:
     """Fail closed when lock preconditions are not met on the declared worktree.
 
@@ -296,6 +297,19 @@ def assess_issue_lock_worktree(
     by construction and could never satisfy it. Every other precondition —
     notably worktree cleanliness — still applies unchanged, and brand-new issue
     claims keep the full base-equivalence requirement.
+
+    ``renewal_sanctioned`` waives base-equivalence on exactly the same grounds
+    for the other proven-ownership case (#760): ``issue_lock_renewal`` has shown
+    that an *expired* lease is being renewed by its exact recorded owner — same
+    remote, org, repo, issue, operation, branch, realpath-normalized worktree,
+    claimant username and profile — with the local head matching the remote head
+    and any owning PR head. Such a branch carries committed work for the same
+    reason a recovered one does, so it can never be base-equivalent either.
+
+    Both waivers relax this one requirement and nothing else. Neither is
+    caller-supplied: each is computed server-side from durable lock state plus
+    live observation. With both False every precondition applies exactly as
+    before.
     """
     bases = base_branches or BASE_BRANCHES
     reasons: list[str] = []
@@ -314,9 +328,12 @@ def assess_issue_lock_worktree(
             f"(dirty files: {', '.join(dirty_files)})"
         )
 
-    if recovery_sanctioned:
+    if recovery_sanctioned or renewal_sanctioned:
         # Base-equivalence intentionally not evaluated: ownership was proven
-        # against the durable lock record instead (#753).
+        # against the durable lock record instead — by dead-session recovery
+        # (#753) or by exact-owner renewal of an expired lease (#760). Every
+        # other precondition above and below still applies; cleanliness in
+        # particular is checked before this branch and is never waived.
         pass
     elif base_equivalent is False:
         reasons.append(
@@ -347,6 +364,7 @@ def assess_issue_lock_worktree(
         base_branch=base_branch,
         base_equivalent=base_equivalent,
         recovery_sanctioned=recovery_sanctioned,
+        renewal_sanctioned=renewal_sanctioned,
     )
 
 
@@ -406,6 +424,7 @@ def _assessment(
     base_branch: str | None = None,
     base_equivalent: bool | None = None,
     recovery_sanctioned: bool = False,
+    renewal_sanctioned: bool = False,
 ) -> dict:
     return {
         "proven": proven,
@@ -418,7 +437,10 @@ def _assessment(
         "base_branch": base_branch,
         "base_equivalent": base_equivalent,
         "recovery_sanctioned": recovery_sanctioned,
-        "base_equivalence_waived": bool(recovery_sanctioned),
+        "renewal_sanctioned": renewal_sanctioned,
+        # Either proven-ownership waiver relaxes base-equivalence; the two are
+        # reported separately so an audit can tell which one applied.
+        "base_equivalence_waived": bool(recovery_sanctioned or renewal_sanctioned),
     }
 
 

master_parity_gate.py

@@ -24,12 +24,50 @@ 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:
@@ -58,6 +96,75 @@ 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.
 
@@ -72,18 +179,38 @@ def _short(sha: str | None) -> str:
     return sha[:12] if sha else "unknown"
 
 
-def assess_master_parity(startup: dict | None, current_head: str | None) -> dict:
+def assess_master_parity(
+    startup: dict | None,
+    current_head: str | None,
+    live_remote_head: str | None = None,
+) -> dict:
     """Compare the startup baseline against the current on-disk ``HEAD``.
 
-    Pure: both HEADs are supplied by the caller. Returns a structured result:
+    Pure: all 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`` -- alias of ``stale``; the recovery action.
-    - ``determinable`` -- whether both HEADs were known well enough to compare.
+    - ``restart_required`` -- ``stale`` or ``live_stale``; the recovery action.
+    - ``determinable`` -- whether both local 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] = []
@@ -91,32 +218,56 @@ def assess_master_parity(startup: dict | None, current_head: str | None) -> dict
     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, reasons)
+        return _result(True, False, False, startup_head, current_head,
+                       live_remote_head, False, 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, reasons)
-
-    if startup_head == current_head:
-        return _result(True, False, True, startup_head, current_head, reasons)
+        return _result(True, False, False, startup_head, current_head,
+                       live_remote_head, False, 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 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)
+            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")
+
+    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)
 
 
-def _result(in_parity, stale, determinable, 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)
     return {
         "in_parity": in_parity,
         "stale": stale,
-        "restart_required": stale,
+        "restart_required": stale or live_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),
     }
 
@@ -130,11 +281,13 @@ 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; only a definitively stale server blocks.
+    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.
     """
     if gate_disabled():
         return []
-    if assessment.get("stale"):
+    if assessment.get("stale") or assessment.get("live_stale"):
         return list(assessment.get("reasons") or
                     ["server code is stale relative to master (fail closed)"])
     return []
@@ -147,6 +300,10 @@ 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 "
@@ -157,6 +314,45 @@ 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"):

tests/conftest.py

@@ -167,6 +167,35 @@ 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_760_exact_owner_lease_renewal.py

@@ -0,0 +1,447 @@
+"""Exact-owner renewal of an expired author issue lease (#760).
+
+Covers the renewal disposition that lets the exact recorded owner re-acquire
+its own lock after the wall-clock lease expires — including while the recording
+MCP daemon PID is still alive — plus every rejection condition that must keep
+failing closed, and the pre-existing dead-PID and live-foreign dispositions
+that must remain untouched.
+"""
+
+import inspect
+import os
+import subprocess
+import sys
+import tempfile
+import unittest
+from datetime import datetime, timedelta, timezone
+
+sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
+
+import issue_lock_renewal  # noqa: E402
+import issue_lock_store  # noqa: E402
+
+ISSUE = 5150
+BRANCH = f"fix/issue-{ISSUE}-demo"
+WORKTREE = "/scratch/wt-5150"
+HEAD = "c" * 40
+OTHER_SHA = "d" * 40
+IDENTITY = "example-user"
+PROFILE = "example-author"
+REMOTE = "prgs"
+ORG = "ExampleOrg"
+REPO = "ExampleRepo"
+
+
+def dead_pid() -> int:
+    """A PID that has certainly exited (spawned, then reaped)."""
+    proc = subprocess.Popen([sys.executable, "-c", "pass"])
+    proc.wait()
+    return proc.pid
+
+
+def past_ts(hours: int = 1) -> str:
+    return (
+        (datetime.now(timezone.utc) - timedelta(hours=hours))
+        .isoformat()
+        .replace("+00:00", "Z")
+    )
+
+
+def future_ts(hours: int = 4) -> str:
+    return (
+        (datetime.now(timezone.utc) + timedelta(hours=hours))
+        .isoformat()
+        .replace("+00:00", "Z")
+    )
+
+
+def make_lock(*, expires_at: str | None = None, pid: int | None = None, **overrides):
+    """An expired lock owned by a still-alive daemon PID — the #760 condition."""
+    lock = {
+        "issue_number": ISSUE,
+        "branch_name": BRANCH,
+        "worktree_path": WORKTREE,
+        "remote": REMOTE,
+        "org": ORG,
+        "repo": REPO,
+        # os.getpid() is unambiguously alive: the whole point of #760 is that
+        # daemon liveness is not evidence of an active author task.
+        "session_pid": os.getpid() if pid is None else pid,
+        "lock_generation": 3,
+        "work_lease": {
+            "operation_type": issue_lock_store.AUTHOR_ISSUE_WORK_LEASE,
+            "issue_number": ISSUE,
+            "branch": BRANCH,
+            "worktree_path": WORKTREE,
+            "claimant": {"username": IDENTITY, "profile": PROFILE},
+            "created_at": past_ts(5),
+            "expires_at": expires_at or past_ts(),
+        },
+    }
+    lease_overrides = overrides.pop("work_lease", None)
+    if lease_overrides:
+        lock["work_lease"].update(lease_overrides)
+    lock.update(overrides)
+    return lock
+
+
+def assess(lock=None, **overrides):
+    """Run the assessor with all-passing evidence unless overridden."""
+    kwargs = {
+        "issue_number": ISSUE,
+        "branch_name": BRANCH,
+        "worktree_path": WORKTREE,
+        "remote": REMOTE,
+        "org": ORG,
+        "repo": REPO,
+        "identity": IDENTITY,
+        "profile": PROFILE,
+        "current_branch": BRANCH,
+        "porcelain_status": "",
+        "worktree_exists": True,
+        "head_sha": HEAD,
+        "remote_head_sha": HEAD,
+        "pr_head_sha": None,
+        "pr_number": None,
+        "competing_live_locks": [],
+        "candidate_branches": [BRANCH],
+        "current_pid": 4242,
+    }
+    kwargs.update(overrides)
+    return issue_lock_renewal.assess_exact_owner_lease_renewal(
+        make_lock() if lock is None else lock, **kwargs
+    )
+
+
+class ExactOwnerRenewalGranted(unittest.TestCase):
+    """AC1/AC3-AC7: the positive path."""
+
+    def test_expired_lease_alive_pid_exact_owner_is_renewable(self):
+        result = assess()
+        self.assertEqual(result["outcome"], issue_lock_renewal.RENEWAL_SANCTIONED)
+        self.assertTrue(result["renewal_sanctioned"])
+        self.assertTrue(result["is_candidate"])
+
+    def test_renewal_holds_when_owning_pr_head_matches(self):
+        result = assess(pr_number=999, pr_head_sha=HEAD)
+        self.assertTrue(result["renewal_sanctioned"])
+
+    def test_evidence_records_both_sides_of_the_transition(self):
+        result = assess()
+        evidence = result["evidence"]
+        self.assertEqual(evidence["prior_pid"], os.getpid())
+        self.assertTrue(evidence["prior_pid_alive"])
+        self.assertEqual(evidence["replacement_pid"], 4242)
+        self.assertTrue(evidence["prior_expires_at"])
+
+
+class ExactOwnerRenewalRefused(unittest.TestCase):
+    """AC3-AC8: every near-match must fail closed, one reason at a time."""
+
+    def _refused(self, **overrides):
+        result = assess(**overrides)
+        self.assertEqual(result["outcome"], issue_lock_renewal.REFUSED)
+        self.assertFalse(result["renewal_sanctioned"])
+        self.assertTrue(result["reasons"])
+        return result
+
+    def test_different_branch_refused(self):
+        result = self._refused(branch_name=f"fix/issue-{ISSUE}-other")
+        self.assertTrue(any("branch" in r for r in result["reasons"]))
+
+    def test_different_worktree_refused(self):
+        result = self._refused(worktree_path="/scratch/somewhere-else")
+        self.assertTrue(any("worktree" in r for r in result["reasons"]))
+
+    def test_different_claimant_refused(self):
+        result = self._refused(identity="someone-else")
+        self.assertTrue(any("claimant" in r for r in result["reasons"]))
+
+    def test_different_profile_refused(self):
+        result = self._refused(profile="other-author")
+        self.assertTrue(any("profile" in r for r in result["reasons"]))
+
+    def test_different_remote_org_or_repo_refused(self):
+        self._refused(remote="dadeschools")
+        self._refused(org="OtherOrg")
+        self._refused(repo="OtherRepo")
+
+    def test_dirty_worktree_refused(self):
+        result = self._refused(porcelain_status=" M gitea_mcp_server.py\n")
+        self.assertTrue(any("uncommitted" in r for r in result["reasons"]))
+
+    def test_missing_worktree_refused(self):
+        result = self._refused(worktree_exists=False)
+        self.assertTrue(any("does not exist" in r for r in result["reasons"]))
+
+    def test_worktree_on_wrong_branch_refused(self):
+        self._refused(current_branch="master")
+
+    def test_local_and_remote_head_mismatch_refused(self):
+        result = self._refused(remote_head_sha=OTHER_SHA)
+        self.assertTrue(
+            any("does not equal remote head" in r for r in result["reasons"])
+        )
+
+    def test_unpublished_branch_refused(self):
+        result = self._refused(remote_head_sha=None)
+        self.assertTrue(any("remote branch head" in r for r in result["reasons"]))
+
+    def test_pr_head_mismatch_refused(self):
+        result = self._refused(pr_number=999, pr_head_sha=OTHER_SHA)
+        self.assertTrue(any("does not equal local" in r for r in result["reasons"]))
+
+    def test_unobservable_pr_head_refused(self):
+        self._refused(pr_number=999, pr_head_sha=None)
+
+    def test_competing_live_lock_on_same_issue_refused(self):
+        result = self._refused(
+            competing_live_locks=[
+                {"issue_number": ISSUE, "branch_name": BRANCH, "pid": 777}
+            ]
+        )
+        self.assertTrue(any("live lock" in r for r in result["reasons"]))
+
+    def test_competing_live_lock_holding_the_branch_refused(self):
+        self._refused(
+            competing_live_locks=[
+                {"issue_number": 111, "branch_name": BRANCH, "worktree_path": ""}
+            ]
+        )
+
+    def test_competing_branch_claim_refused(self):
+        result = self._refused(candidate_branches=[BRANCH, f"feat/issue-{ISSUE}-rival"])
+        self.assertTrue(any("issue marker" in r for r in result["reasons"]))
+
+    def test_malformed_durable_lock_refused(self):
+        lock = make_lock()
+        lock["worktree_path"] = ""
+        result = assess(lock)
+        self.assertEqual(result["outcome"], issue_lock_renewal.REFUSED)
+
+    def test_lock_without_recorded_claimant_refused(self):
+        lock = make_lock()
+        lock["work_lease"]["claimant"] = {}
+        result = assess(lock)
+        self.assertEqual(result["outcome"], issue_lock_renewal.REFUSED)
+
+
+class NotARenewalCandidate(unittest.TestCase):
+    """AC12 and scope: situations renewal must decline to judge at all."""
+
+    def test_live_foreign_lease_is_never_a_candidate(self):
+        lock = make_lock(expires_at=future_ts())
+        result = assess(lock, identity="someone-else")
+        self.assertEqual(result["outcome"], issue_lock_renewal.NO_CANDIDATE)
+        self.assertFalse(result["renewal_sanctioned"])
+
+    def test_unexpired_lease_is_never_a_candidate(self):
+        lock = make_lock(expires_at=future_ts())
+        result = assess(lock)
+        self.assertEqual(result["outcome"], issue_lock_renewal.NO_CANDIDATE)
+
+    def test_dead_pid_under_unexpired_lease_stays_with_753(self):
+        """The opposite trigger; #760 must not re-own it."""
+        lock = make_lock(expires_at=future_ts(), pid=dead_pid())
+        result = assess(lock)
+        self.assertEqual(result["outcome"], issue_lock_renewal.NO_CANDIDATE)
+
+    def test_absent_lock_is_not_a_candidate(self):
+        result = assess({})
+        self.assertEqual(result["outcome"], issue_lock_renewal.NO_CANDIDATE)
+
+    def test_different_issue_is_not_a_candidate(self):
+        lock = make_lock()
+        lock["issue_number"] = ISSUE + 1
+        result = assess(lock)
+        self.assertEqual(result["outcome"], issue_lock_renewal.NO_CANDIDATE)
+
+    def test_different_operation_type_is_not_a_candidate(self):
+        lock = make_lock()
+        lock["work_lease"]["operation_type"] = "review_pr_work"
+        result = assess(lock)
+        self.assertEqual(result["outcome"], issue_lock_renewal.NO_CANDIDATE)
+
+
+class DaemonPidIsNotTaskLiveness(unittest.TestCase):
+    """AC16: a live recorded PID is never, by itself, authorization."""
+
+    def test_alive_pid_alone_does_not_authorize_renewal(self):
+        # Every ownership fact except the live PID is wrong.
+        result = assess(identity="someone-else", branch_name="fix/issue-1-nope")
+        self.assertEqual(result["outcome"], issue_lock_renewal.REFUSED)
+        self.assertTrue(result["evidence"]["prior_pid_alive"])
+
+    def test_renewal_does_not_require_a_dead_pid(self):
+        result = assess()
+        self.assertTrue(result["evidence"]["prior_pid_alive"])
+        self.assertTrue(result["renewal_sanctioned"])
+
+    def test_dead_pid_does_not_block_an_otherwise_exact_owner(self):
+        lock = make_lock(pid=dead_pid())
+        result = assess(lock)
+        self.assertTrue(result["renewal_sanctioned"])
+
+
+class ConflictGateOrdering(unittest.TestCase):
+    """AC2: the same-owner allowance is reachable on an expired lease.
+
+    These cases need a worktree that genuinely exists on disk. The #601 reclaim
+    affordance already permits takeover when the recorded worktree is missing,
+    so a fictional path would satisfy the gate for the wrong reason and never
+    exercise the ordering defect this issue is about.
+    """
+
+    @classmethod
+    def setUpClass(cls):
+        cls._tmp = tempfile.TemporaryDirectory()
+        cls.worktree = cls._tmp.name
+
+    @classmethod
+    def tearDownClass(cls):
+        cls._tmp.cleanup()
+
+    def present_lock(self, **overrides):
+        return make_lock(worktree_path=self.worktree, **overrides)
+
+    def test_expired_same_owner_is_allowed_when_renewal_is_sanctioned(self):
+        block = issue_lock_store.assess_same_issue_lease_conflict(
+            self.present_lock(),
+            issue_number=ISSUE,
+            branch_name=BRANCH,
+            worktree_path=self.worktree,
+            renewal_sanctioned=True,
+        )
+        self.assertIsNone(block)
+
+    def test_expired_same_owner_still_blocks_without_the_waiver(self):
+        """Regression for the ordering defect: no waiver, no change in behavior.
+
+        Live PID and a present worktree, so the #601 reclaim affordance refuses;
+        before #760 this was the permanent dead end for an exact owner.
+        """
+        lock = self.present_lock()
+        self.assertFalse(
+            issue_lock_store.assess_expired_lock_reclaim(lock)["reclaim_allowed"]
+        )
+        block = issue_lock_store.assess_same_issue_lease_conflict(
+            lock,
+            issue_number=ISSUE,
+            branch_name=BRANCH,
+            worktree_path=self.worktree,
+        )
+        self.assertIsNotNone(block)
+        self.assertIn("Recovery review is required", block)
+
+    def test_waiver_does_not_unlock_a_different_owner(self):
+        """AC11: the waiver is scoped by same_owner, not merely by its own flag."""
+        block = issue_lock_store.assess_same_issue_lease_conflict(
+            self.present_lock(),
+            issue_number=ISSUE,
+            branch_name=f"fix/issue-{ISSUE}-someone-else",
+            worktree_path=self.worktree,
+            renewal_sanctioned=True,
+        )
+        self.assertIsNotNone(block)
+        self.assertIn("Recovery review is required", block)
+
+    def test_live_lease_disposition_is_unchanged(self):
+        """AC12: a live foreign lease still blocks, waiver or not."""
+        block = issue_lock_store.assess_same_issue_lease_conflict(
+            self.present_lock(expires_at=future_ts()),
+            issue_number=ISSUE,
+            branch_name=f"fix/issue-{ISSUE}-someone-else",
+            worktree_path="/scratch/other",
+            renewal_sanctioned=True,
+        )
+        self.assertIsNotNone(block)
+        self.assertIn("already has an active", block)
+
+    def test_dead_pid_reclaim_path_is_unchanged(self):
+        """AC11: expired + dead PID still reclaims through the #601 affordance."""
+        lock = self.present_lock(pid=dead_pid())
+        reclaim = issue_lock_store.assess_expired_lock_reclaim(lock)
+        self.assertTrue(reclaim["reclaim_allowed"])
+        block = issue_lock_store.assess_same_issue_lease_conflict(
+            lock,
+            issue_number=ISSUE,
+            branch_name=BRANCH,
+            worktree_path=self.worktree,
+        )
+        self.assertIsNone(block)
+
+
+class RenewalRecordAndDownstream(unittest.TestCase):
+    """AC9/AC10: durable audit trail, and a renewed lock that actually works."""
+
+    def test_record_captures_prior_and_replacement_state(self):
+        assessment = assess()
+        record = issue_lock_renewal.build_renewal_record(
+            assessment,
+            renewed_at="2026-01-01T00:00:00Z",
+            new_expires_at="2026-01-01T04:00:00Z",
+        )
+        self.assertTrue(record["renewed"])
+        self.assertEqual(record["prior_pid"], os.getpid())
+        self.assertEqual(record["new_expires_at"], "2026-01-01T04:00:00Z")
+        self.assertEqual(record["renewed_at"], "2026-01-01T00:00:00Z")
+        self.assertEqual(record["identity"], IDENTITY)
+        self.assertEqual(record["profile"], PROFILE)
+        self.assertTrue(record["prior_expires_at"])
+        self.assertTrue(record["proof"])
+
+    def test_renewed_lock_satisfies_verify_lock_for_mutation(self):
+        renewed = make_lock(expires_at=future_ts())
+        renewed["session_pid"] = os.getpid()
+        renewed["lease_renewal"] = {"renewed": True}
+        verdict = issue_lock_store.verify_lock_for_mutation(
+            renewed,
+            issue_number=ISSUE,
+            branch_name=BRANCH,
+        )
+        self.assertTrue(verdict["proven"])
+        self.assertFalse(verdict["block"])
+
+    def test_refusal_message_names_the_missing_evidence(self):
+        assessment = assess(porcelain_status=" M gitea_mcp_server.py\n")
+        message = issue_lock_renewal.format_renewal_refusal(assessment)
+        self.assertIn("refused", message)
+        self.assertIn("uncommitted", message)
+
+
+class NoCallerControlledRenewalFlag(unittest.TestCase):
+    """AC14: renewal eligibility is never declarable by a caller."""
+
+    def test_lock_issue_tool_exposes_no_renewal_parameter(self):
+        import gitea_mcp_server
+
+        target = gitea_mcp_server.gitea_lock_issue
+        target = getattr(target, "fn", getattr(target, "__wrapped__", target))
+        params = set(inspect.signature(target).parameters)
+        for forbidden in ("renewal_sanctioned", "renew", "allow_renewal", "is_owner"):
+            self.assertNotIn(forbidden, params)
+
+    def test_store_defaults_to_no_waiver(self):
+        params = inspect.signature(
+            issue_lock_store.assess_same_issue_lease_conflict
+        ).parameters
+        self.assertIs(params["renewal_sanctioned"].default, False)
+        bind_params = inspect.signature(issue_lock_store.bind_session_lock).parameters
+        self.assertIs(bind_params["renewal_sanctioned"].default, False)
+
+
+class NoIssueNumberSpecialCasing(unittest.TestCase):
+    """AC17: no repository issue or PR number is special-cased."""
+
+    def test_module_contains_no_hardcoded_issue_special_cases(self):
+        source = inspect.getsource(issue_lock_renewal)
+        code = "\n".join(
+            line for line in source.splitlines() if not line.strip().startswith("#")
+        )
+        for literal in ("757", "759", "760"):
+            self.assertNotIn(f"== {literal}", code)
+            self.assertNotIn(f"issue_number == {literal}", code)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_issue_760_mcp_renewal_path.py

@@ -0,0 +1,342 @@
+"""MCP-level exact-owner lease renewal through ``gitea_lock_issue`` (#760).
+
+The unit suite in ``test_issue_760_exact_owner_lease_renewal`` proves the
+renewal *disposition*. It cannot prove the disposition survives the rest of the
+tool, and it did not: the waiver was computed and then discarded before
+``assess_issue_lock_worktree``, so every real renewal still failed on
+base-equivalence. A branch being renewed always carries committed work, so it is
+never base-equivalent by construction — exactly the argument #753 already makes
+for recovery.
+
+These tests drive the public tool end to end against a real git repository and a
+real durable lock file, composing every gate in the production order.
+"""
+
+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 mcp_server  # noqa: E402
+
+ISSUE = 9760
+BRANCH = f"fix/issue-{ISSUE}-renewal-mcp"
+IDENTITY = "example-user"
+PROFILE = "test-author-prgs"
+ORG = "Scaled-Tech-Consulting"
+REPO = "Gitea-Tools"
+
+
+def _past_ts(hours: int = 1) -> str:
+    return (
+        (datetime.now(timezone.utc) - timedelta(hours=hours))
+        .isoformat()
+        .replace("+00:00", "Z")
+    )
+
+
+class _RenewalMcpBase(unittest.TestCase):
+    """Real git repo + durable expired lock owned by a live PID.
+
+    The recorded PID is ``os.getpid()`` — unambiguously alive. That is the whole
+    point of #760: the PID belongs to the long-lived MCP daemon, so its liveness
+    says nothing about whether the authoring task still holds the work.
+    """
+
+    def setUp(self):
+        self.lock_dir = tempfile.TemporaryDirectory()
+        self.addCleanup(self.lock_dir.cleanup)
+        self.repo = tempfile.mkdtemp(prefix="issue760-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()
+        # The branch carries committed work, so it is NOT base-equivalent.
+        self._git("checkout", "-q", "-b", BRANCH)
+        with open(os.path.join(self.repo, "work.txt"), "w") as fh:
+            fh.write("author work\n")
+        self._git("add", "work.txt")
+        self._git("commit", "-q", "-m", "author work")
+        self.head_sha = self._git("rev-parse", "HEAD").stdout.strip()
+        self.worktree = os.path.realpath(self.repo)
+
+    def write_expired_lock(self, **overrides):
+        path = issue_lock_store.lock_file_path(
+            remote="prgs",
+            org=ORG,
+            repo=REPO,
+            issue_number=ISSUE,
+            lock_dir=self.lock_dir.name,
+        )
+        claimant = {"username": IDENTITY, "profile": PROFILE}
+        pid = overrides.pop("session_pid", os.getpid())
+        overrides.pop("pid", None)
+        lease_overrides = overrides.pop("work_lease", {})
+        data = {
+            "issue_number": ISSUE,
+            "branch_name": BRANCH,
+            "remote": "prgs",
+            "org": ORG,
+            "repo": REPO,
+            "worktree_path": self.worktree,
+            "session_pid": pid,
+            "pid": pid,
+            "lock_generation": 3,
+            "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": _past_ts(5),
+                "last_heartbeat_at": _past_ts(5),
+                "expires_at": _past_ts(),          # already expired
+            },
+            "lock_provenance": issue_lock_provenance.build_sanctioned_lock_provenance(
+                tool="gitea_lock_issue",
+                claimant=claimant,
+            ),
+        }
+        data["work_lease"].update(lease_overrides)
+        data.update(overrides)
+        data["session_pid"] = pid
+        data["pid"] = pid
+        data["lock_file_path"] = path
+        issue_lock_store.save_lock_file(path, data)
+        return path
+
+    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="", branch=BRANCH, head=None):
+        return {
+            "current_branch": branch,
+            "porcelain_status": porcelain,
+            # The decisive fact: a branch carrying work is never base-equivalent.
+            "base_equivalent": False,
+            "head_sha": head or 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,
+    ):
+        """Drive the public tool for the published exact-owner renewal shape."""
+        if branch_entries is None:
+            branch_entries = [{"name": BRANCH, "commit": {"id": self.head_sha}}]
+        if open_prs is None:
+            open_prs = [{"number": 4242, "head": {"ref": BRANCH, "sha": self.head_sha}}]
+        if git_state is None:
+            git_state = 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,
+            )
+
+
+class TestRenewalReachableThroughTool(_RenewalMcpBase):
+    """F1: the sanctioned renewal must survive every downstream gate."""
+
+    def test_expired_lease_live_pid_exact_owner_renews_through_the_tool(self):
+        prior = issue_lock_store.read_lock_file(self.write_expired_lock())
+        self.assertTrue(issue_lock_store.is_lease_expired(prior))
+        self.assertTrue(issue_lock_store.is_process_alive(prior["session_pid"]))
+
+        result = self.run_lock_issue()
+
+        self.assertTrue(result["success"], result)
+        self.assertEqual(result["issue_number"], ISSUE)
+        self.assertEqual(result["branch_name"], BRANCH)
+        # The renewal is reported natively, so no lock-file inspection is needed.
+        self.assertIn("lease_renewal", result)
+        self.assertTrue(result["lease_renewal"]["renewed"])
+        self.assertIn("Renewed the expired", result["message"])
+
+    def test_renewed_lock_records_prior_and_replacement_evidence(self):
+        prior = issue_lock_store.read_lock_file(self.write_expired_lock())
+        prior_expiry = prior["work_lease"]["expires_at"]
+        prior_generation = issue_lock_store.lock_generation(prior)
+
+        result = self.run_lock_issue()
+        written = issue_lock_store.read_lock_file(result["lock_file_path"])
+
+        renewal = written["lease_renewal"]
+        self.assertTrue(renewal["renewed"])
+        self.assertEqual(renewal["prior_pid"], prior["session_pid"])
+        self.assertTrue(renewal["prior_pid_alive"])
+        self.assertEqual(renewal["prior_expires_at"], prior_expiry)
+        self.assertEqual(renewal["identity"], IDENTITY)
+        self.assertEqual(renewal["profile"], PROFILE)
+        self.assertEqual(renewal["head_sha"], self.head_sha)
+        self.assertTrue(renewal["proof"])
+        # New expiry is a fresh absolute stamp, later than the one it replaced.
+        self.assertEqual(renewal["new_expires_at"], written["work_lease"]["expires_at"])
+        self.assertGreater(renewal["new_expires_at"], prior_expiry)
+        # Compare-and-swap advanced the generation exactly once.
+        self.assertEqual(
+            issue_lock_store.lock_generation(written), prior_generation + 1
+        )
+
+    def test_renewed_lock_is_live_and_satisfies_mutation_ownership(self):
+        self.write_expired_lock()
+        result = self.run_lock_issue()
+        written = issue_lock_store.read_lock_file(result["lock_file_path"])
+
+        self.assertTrue(issue_lock_store.assess_lock_freshness(written)["live"])
+        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 test_recovery_record_is_not_written_for_a_live_owner_renewal(self):
+        """#753 recovery must not be claimed when the recorded PID is alive."""
+        self.write_expired_lock()
+        result = self.run_lock_issue()
+        written = issue_lock_store.read_lock_file(result["lock_file_path"])
+        self.assertNotIn("dead_session_recovery", written)
+
+
+class TestRenewalWaiverIsNarrow(_RenewalMcpBase):
+    """The waiver relaxes base-equivalence and nothing else."""
+
+    def test_dirty_worktree_still_blocks_a_would_be_renewal(self):
+        """Cleanliness is never waived; the renewal assessor refuses first.
+
+        A dirty worktree makes the renewal refuse, so no waiver is issued and
+        the lease-conflict gate fails closed ahead of the worktree gate. The
+        refusal names the uncommitted files, so the owner still learns why.
+        """
+        self.write_expired_lock()
+        with self.assertRaises(Exception) as ctx:
+            self.run_lock_issue(
+                git_state=self._git_state(porcelain=" M gitea_mcp_server.py\n")
+            )
+        message = str(ctx.exception)
+        self.assertIn("Recovery review is required before takeover", message)
+        self.assertIn("worktree has uncommitted tracked changes", message)
+        self.assertIn("gitea_mcp_server.py", message)
+
+    def test_foreign_claimant_cannot_use_the_waiver(self):
+        """A near-match owner gets no renewal and no base-equivalence waiver."""
+        self.write_expired_lock()
+        with self.assertRaises(Exception) as ctx:
+            self.run_lock_issue(identity="someone-else")
+        message = str(ctx.exception)
+        self.assertIn("Recovery review is required before takeover", message)
+        # The refusal names the missing ownership evidence (#760 diagnostics).
+        self.assertIn("does not match active identity", message)
+
+    def test_foreign_profile_cannot_use_the_waiver(self):
+        self.write_expired_lock()
+        with self.assertRaises(Exception) as ctx:
+            self.run_lock_issue(profile="other-author")
+        self.assertIn(
+            "Recovery review is required before takeover", str(ctx.exception)
+        )
+
+    def test_unpublished_branch_cannot_use_the_waiver(self):
+        """No remote head to agree with, so exact-owner renewal is refused."""
+        self.write_expired_lock()
+        with self.assertRaises(Exception) as ctx:
+            self.run_lock_issue(branch_entries=[], open_prs=[])
+        self.assertIn(
+            "Recovery review is required before takeover", str(ctx.exception)
+        )
+
+    def test_pr_head_mismatch_cannot_use_the_waiver(self):
+        self.write_expired_lock()
+        other = "9" * 40
+        with self.assertRaises(Exception) as ctx:
+            self.run_lock_issue(
+                open_prs=[{"number": 4242, "head": {"ref": BRANCH, "sha": other}}]
+            )
+        self.assertIn(
+            "Recovery review is required before takeover", str(ctx.exception)
+        )
+
+    def test_non_base_equivalent_branch_still_blocks_without_any_waiver(self):
+        """No durable lock at all: the ordinary base-equivalence rule applies."""
+        with self.assertRaises(Exception) as ctx:
+            self.run_lock_issue()
+        self.assertIn("must be base-equivalent", str(ctx.exception))
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_issue_772_unpublished_claim_recovery.py

@@ -871,9 +871,13 @@ class TestAc6McpUnpublishedClaimRecovery(_UnpublishedMcpBase):
         save_calls: list[dict] = []
         real_save = mcp_server._save_issue_lock
 
-        def tracking_save(data, *, expected_generation=None):
+        def tracking_save(data, *, expected_generation=None, renewal_sanctioned=False):
             save_calls.append({"expected_generation": expected_generation, "data": dict(data)})
-            return real_save(data, expected_generation=expected_generation)
+            return real_save(
+                data,
+                expected_generation=expected_generation,
+                renewal_sanctioned=renewal_sanctioned,
+            )
 
         with patch("mcp_server._save_issue_lock", side_effect=tracking_save):
             result = self.run_lock_issue()
@@ -926,18 +930,33 @@ class TestAc6McpUnpublishedClaimRecovery(_UnpublishedMcpBase):
         real_bind = issue_lock_store.bind_session_lock
         bind_calls: list[int | None] = []
 
-        def racing_bind(data, lock_dir=None, expected_generation=None):
+        # #760 added the renewal waiver keyword; the double forwards it verbatim
+        # so this race still exercises the real compare-and-swap.
+        def racing_bind(
+            data, lock_dir=None, expected_generation=None, renewal_sanctioned=False
+        ):
             bind_calls.append(expected_generation)
             if expected_generation is None:
-                return real_bind(data, lock_dir=lock_dir, expected_generation=None)
+                return real_bind(
+                    data,
+                    lock_dir=lock_dir,
+                    expected_generation=None,
+                    renewal_sanctioned=renewal_sanctioned,
+                )
             # First concurrent writer wins.
             if len([c for c in bind_calls if c is not None]) == 1:
                 return real_bind(
-                    data, lock_dir=lock_dir, expected_generation=expected_generation
+                    data,
+                    lock_dir=lock_dir,
+                    expected_generation=expected_generation,
+                    renewal_sanctioned=renewal_sanctioned,
                 )
             # Second concurrent writer still holds the pre-race generation.
             return real_bind(
-                data, lock_dir=lock_dir, expected_generation=expected_generation
+                data,
+                lock_dir=lock_dir,
+                expected_generation=expected_generation,
+                renewal_sanctioned=renewal_sanctioned,
             )
 
         # First recovery succeeds and advances generation.

tests/test_master_parity_gate.py

@@ -78,6 +78,95 @@ 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}):
@@ -95,6 +184,149 @@ 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."""
 
@@ -105,6 +337,13 @@ 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
@@ -147,6 +386,36 @@ 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

@@ -0,0 +1,458 @@
+"""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

@@ -0,0 +1,57 @@
+{
+  "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,27 +8,7 @@ from dataclasses import dataclass
 from pathlib import Path
 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:
-    lowered = key.lower()
-    if lowered in _FORBIDDEN_EXACT_KEYS:
-        return True
-    return (
-        lowered.startswith(_FORBIDDEN_KEY_PREFIXES)
-        or lowered.endswith(_FORBIDDEN_KEY_SUFFIXES)
-    )
+from webui.registry_safety import reject_credential_keys as _reject_credential_keys
 
 _REQUIRED_PROJECT_FIELDS = (
     "id",
@@ -79,18 +59,6 @@ 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

@@ -0,0 +1,52 @@
+"""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

@@ -0,0 +1,647 @@
+"""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)