test(author): execute the native #953 tool functions and compensation paths (#953)

Review 632 F4. Every prior reference to `gitea_recover_incomplete_bootstrap_lock`
and `gitea_inspect_issue_lock_contract` under `tests/` was a string literal — a
`tool=` argument, an assertion on returned prose, or a docs substring check —
and the suite reconstructed the recovery sequence by hand from
`assess_bootstrap_lock_recovery`, `build_canonical_issue_lock`,
`build_recovery_record`, and `bind_session_lock`. A hand-written sequence
validates the decision layer but cannot see a divergence between itself and
the tool body, which is exactly how F1 and F2 — both call-site defects —
survived 61 passing cases.

The new cases drive the registered functions against a real `git init`
repository, a real durable lock file, and config-backed profiles:

* `NamespaceMutationWallOnRecovery` — the gate is reached with this task and
  `author_role_exclusive=True`; its return value aborts the tool rather than
  being computed and discarded; the author namespace succeeds and produces a
  canonical lock; a reviewer namespace is refused with `namespace_block` even
  when the claimant data would otherwise match, and emits the standard BLOCKED
  audit record; merger is refused; a mismatched claimant profile is refused by
  the exact-owner layer with the namespace wall explicitly clear; a mismatched
  head is refused; every refusal leaves the lock bytes, generation, branch,
  worktree, and an unrelated lock untouched. A subtest matrix asserts the
  role-kind wall admits `author` and refuses reviewer, merger, limited, and
  mixed.
* `InspectionToolExecutes` — the registered read-only tool reports the contract
  and the recovery preview while leaving lock bytes, mtime, HEAD, and porcelain
  status unchanged, and reports an absent lock without creating one.
* `Ac7PostCompensationGuidance` — drives the real bootstrap to its AC7 refusal
  with a forced partial lock and asserts the returned action against the state
  the rollback actually left: complete cleanup directs to a bootstrap retry and
  that retry is then executed and succeeds, leaving exactly one canonical lock
  and one branch; partial cleanup with a surviving lock, and with a surviving
  branch and worktree, each get their own executable action; a rollback that
  never completed is distinguished from both; no recommendation names a deleted
  artifact; unrelated locks are byte-identical afterwards. Two cases cover
  `release_session_lock` directly — that the rollback now really removes the
  lock, and that it refuses a lock owned by another session.
* `NativeEndToEndBootstrapToCreatePr` — bootstrap, inspect, heartbeat,
  legitimate divergence (commit and push), pre-mutation ownership re-check, and
  the unchanged #447 create-PR provenance guard, in one sequence against a real
  origin. `gitea_lock_issue` is patched to fail the test if anything reaches for
  it, so the bootstrap lock is proved to carry the whole cycle unrepaired.
* `DeadProvenanceConstantRemoved` — the removed constant stays removed and the
  sanctioned source set stays unwidened.

`_NativeToolBase` clears `role_session_router` route state per test: the sticky
reviewer-stop marker is process-global and, now that this task is registered in
`AUTHOR_TASKS`, an earlier suite leaving it set would make the first gate refuse
before the namespace gate under test is reached.

Also documents both new tools in `docs/mcp-tool-inventory.md`, so this branch
adds no drift to `test_documented_inventory_equals_registered_tools`; the
failure reason there is now identical to the pinned base's.

Suite: 61 -> 92 cases.

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

author_issue_bootstrap.py

@@ -23,6 +23,7 @@ import shutil
 import subprocess
 from typing import Any, Mapping
 
+import author_lock_contract
 import author_mutation_worktree
 import control_plane_db
 import issue_lock_store
@@ -196,6 +197,58 @@ def _verify_assignment_and_lease_ids(
         # Some lease rows may not yet have an assignment join; still require
         # the lease itself to exist and bind to the claimed session/issue.
         pass
+    # #943 review 622 B2: a lease that is no longer live confers no ownership.
+    # Existence alone previously satisfied this gate, so a released or expired
+    # lease could still authorize a bootstrap for a claim its session had given
+    # up. Checked before the session comparison so the reason names the real
+    # problem rather than reporting a mismatch.
+    from datetime import datetime, timezone
+
+    lease_status = str(lease.get("status") or "").strip().lower()
+    if lease_status and lease_status != "active":
+        return {
+            "success": False,
+            "reason_code": "lease_not_live",
+            "message": (
+                f"lease_id '{lid}' is '{lease_status}', not active; a lease that "
+                "is not live confers no ownership (fail closed)."
+            ),
+            "exact_next_action": (
+                "Re-allocate the work item and pass the live assignment/lease pair."
+            ),
+        }
+    expires_raw = str(lease.get("expires_at") or "").strip()
+    if expires_raw:
+        try:
+            expires_at = datetime.fromisoformat(expires_raw.replace("Z", "+00:00"))
+        except ValueError:
+            return {
+                "success": False,
+                "reason_code": "lease_not_live",
+                "message": (
+                    f"lease_id '{lid}' records an unparseable expiry "
+                    f"'{expires_raw}' (fail closed)."
+                ),
+                "exact_next_action": (
+                    "Re-allocate the work item and pass the live "
+                    "assignment/lease pair."
+                ),
+            }
+        if expires_at.tzinfo is None:
+            expires_at = expires_at.replace(tzinfo=timezone.utc)
+        if expires_at <= datetime.now(timezone.utc):
+            return {
+                "success": False,
+                "reason_code": "lease_not_live",
+                "message": (
+                    f"lease_id '{lid}' expired at {expires_raw}; an expired lease "
+                    "confers no ownership (fail closed)."
+                ),
+                "exact_next_action": (
+                    "Reclaim or re-allocate the lease, then retry with the live pair."
+                ),
+            }
+
     lease_session = str(lease.get("session_id") or "").strip()
     if lease_session and lease_session != owner_session:
         return {
@@ -226,6 +279,36 @@ def _verify_assignment_and_lease_ids(
     return None
 
 
+def _branch_exists(canonical_repo_root: str, branch_name: str) -> bool:
+    """Whether *branch_name* still resolves in the canonical checkout (#953 F2).
+
+    Used after compensating recovery to observe what survived rather than infer
+    it from the journal. Fails closed to ``True``: an unobservable branch is
+    reported as present, so the recommendation stays conservative rather than
+    telling an author to re-bootstrap over something that may still be there.
+    """
+    if not branch_name:
+        return False
+    try:
+        res = subprocess.run(
+            [
+                "git",
+                "-C",
+                canonical_repo_root,
+                "rev-parse",
+                "--verify",
+                "--quiet",
+                f"refs/heads/{branch_name}",
+            ],
+            capture_output=True,
+            text=True,
+            check=False,
+        )
+    except Exception:
+        return True
+    return res.returncode == 0
+
+
 def run_compensating_recovery(
     journal: dict[str, Any],
     canonical_repo_root: str,
@@ -262,10 +345,21 @@ def run_compensating_recovery(
                     issue_number=issue_num,
                     session=session_id,
                     lock_dir=journal_dir,
+                    remote=journal.get("remote"),
+                    # The same defaults the lock was written under, so the
+                    # rollback targets the exact file bind_session_lock keyed.
+                    org=journal.get("org") or "Scaled-Tech-Consulting",
+                    repo=journal.get("repo") or "Gitea-Tools",
                 )
                 rolled_back.append(f"lock:issue-{issue_num}")
-            except Exception:
+            except Exception as exc:
-                pass
+                # #953 F2: a swallowed failure here is what made the rollback
+                # report success while leaving an unrecoverable lock behind.
+                # Record it so the post-compensation classification can see the
+                # lock survived and recommend accordingly.
+                rolled_back.append(
+                    f"lock_release_failed:issue-{issue_num}:{type(exc).__name__}"
+                )
         artifacts["lock_created"] = False
 
     worktree_created = (
@@ -1146,26 +1240,31 @@ def bootstrap_author_issue_worktree(
         save_phase_journal(journal, journal_dir=lock_dir)
 
         # Phase 6: STATE_ESTABLISHED — Issue Lock Acquisition
+        #
+        # #953: this used to hand-build a thinner record — claimant at the top
+        # level, no work_lease, no lock_provenance, no expiry — which every
+        # downstream reader then refused. It now builds through the one shared
+        # canonical contract, so the lock bootstrap writes is the same lock
+        # gitea_lock_issue writes.
         from datetime import datetime, timezone
         try:
-            lock_data = {
+            lock_data = author_lock_contract.build_canonical_issue_lock(
-                "remote": remote,
+                issue_number=issue_number,
-                "org": org or "Scaled-Tech-Consulting",
+                branch_name=target_branch,
-                "repo": repo or "Gitea-Tools",
+                worktree_path=target_worktree,
-                "issue_number": issue_number,
+                remote=remote,
-                "branch": target_branch,
+                org=org or "Scaled-Tech-Consulting",
-                "branch_name": target_branch,
+                repo=repo or "Gitea-Tools",
-                "worktree_path": target_worktree,
+                identity=identity,
-                "owner_session": session,
+                profile=profile,
-                "claimant": {
+                tool="gitea_bootstrap_author_issue_worktree",
-                    "username": identity,
+                source=author_lock_contract.SOURCE_BOOTSTRAP,
-                    "profile": profile,
+                owner_session=session,
-                },
+                assignment_id=assignment_id,
-                "assignment_id": assignment_id,
+                lease_id=lease_id,
-                "lease_id": lease_id,
+                expected_base_sha=live_master_sha,
-                "expected_base_sha": live_master_sha,
+            )
-                "created_at": datetime.now(timezone.utc).isoformat(),
+            lock_data["created_at"] = datetime.now(timezone.utc).isoformat()
-            }
             journal.setdefault("pending_creations", {})["lock"] = True
             journal["artifacts_created"]["lock_created"] = True
             save_phase_journal(journal, journal_dir=lock_dir)
@@ -1183,9 +1282,65 @@ def bootstrap_author_issue_worktree(
                 "exact_next_action": "Verify lease/assignment state and retry.",
             }
 
+        # ── #953 AC7: verify the lock that was actually written ──
+        # Reporting "lock_created: true" and then directing the author to
+        # implement is what produced the unrecoverable state: by the time any
+        # reader refused the lock, the branch already carried commits and every
+        # sanctioned recovery path had become ineligible. The lock is therefore
+        # read back from disk and structurally verified *before* this function
+        # can report success, and a partial lock fails closed here — while the
+        # branch is still base-equivalent and recovery is still cheap.
+        written_lock = issue_lock_store.read_lock_file(lock_res)
+        contract = author_lock_contract.assess_lock_contract(written_lock)
+        if not contract["canonical"]:
+            journal["failure_reason"] = author_lock_contract.format_contract_refusal(
+                contract
+            )
+            compensation = run_compensating_recovery(
+                journal, root, journal_dir=lock_dir
+            )
+            # AC5/AC15: the recommendation must describe the state compensation
+            # actually left, not the state that provoked it.
+            # ``run_compensating_recovery`` has by now released the lock, removed
+            # the worktree, and deleted the branch, so recommending
+            # incomplete-lock recovery for those exact artifacts would refuse
+            # twice over. Observe what survived and answer for that.
+            post_state = author_lock_contract.assess_post_compensation_state(
+                compensation,
+                lock_present=bool(lock_res) and os.path.exists(lock_res),
+                worktree_present=os.path.isdir(target_worktree),
+                branch_present=_branch_exists(root, target_branch),
+            )
+            return {
+                "success": False,
+                "reason_code": "incomplete_issue_lock_contract",
+                "message": author_lock_contract.format_contract_refusal(contract),
+                "issue_number": issue_number,
+                "branch_name": target_branch,
+                "worktree_path": target_worktree,
+                "lock_state": lock_res,
+                "lock_contract": contract,
+                "missing_fields": contract["missing_fields"],
+                "implementation_allowed": False,
+                "compensating_recovery": compensation,
+                "post_compensation_state": post_state,
+                # AC15: never strand a branch or worktree without a structured
+                # recovery recommendation — and never name an artifact the
+                # rollback has already deleted.
+                "exact_next_action": author_lock_contract.post_compensation_action(
+                    post_state,
+                    issue_number=issue_number,
+                    branch_name=target_branch,
+                    worktree_path=target_worktree,
+                    missing_fields=contract["missing_fields"],
+                ),
+                "phase_journal": journal,
+            }
+
         journal["phases"][PHASE_6_STATE_ESTABLISHED] = {
             "status": "completed",
             "lock": lock_res,
+            "lock_contract": contract["contract"],
         }
         journal["phases"][PHASE_7_TRANSITION_COMPLETED] = {
             "status": "completed",
@@ -1209,9 +1364,16 @@ def bootstrap_author_issue_worktree(
             "assignment_id": assignment_id,
             "idempotency_key": key,
             "lock_state": lock_res,
+            "lock_contract": contract,
+            # #953 AC6: the canonical ownership token for this claim. Never null
+            # on a successful bootstrap — it is the fencing token every
+            # subsequent heartbeat and renewal is checked against.
+            "task_session_id": contract["task_session_id"],
+            "implementation_allowed": True,
             "phase_journal": journal,
-            "exact_next_action": (
+            # #953 AC5: executable under the state actually returned. The lock
-                "Call gitea_whoami, then gitea_resolve_task_capability(task='work_issue') "
+            # has been read back and verified canonical, so proceeding to
-                "and proceed with author implementation in the bootstrapped worktree."
+            # implementation is genuinely the correct next step here — which is
-            ),
+            # exactly what the old unconditional wording could not promise.
+            "exact_next_action": author_lock_contract.recommended_action(contract),
         }

author_lock_contract.py

@@ -0,0 +1,623 @@
+"""One canonical author issue-lock contract shared by every writer (#953).
+
+Before this module, ``gitea_lock_issue`` and
+``gitea_bootstrap_author_issue_worktree`` each wrote their own lock record.
+``gitea_lock_issue`` wrote the canonical shape — ``work_lease`` carrying the
+claimant plus a sanctioned ``lock_provenance`` — while bootstrap wrote a thinner
+record with the claimant at the lock top level, ``lease_id: null``, and no
+``work_lease``, ``lock_provenance``, or expiry at all.
+
+Every downstream reader was written against the canonical shape, so a lock that
+bootstrap reported as successfully created was simultaneously:
+
+* un-heartbeatable — the ownership check read the claimant only from
+  ``work_lease.claimant``;
+* un-renewable — expiry is read only from ``work_lease.expires_at``, so a
+  missing lease read as "never expires", and #760 exact-owner renewal only ever
+  assesses an *expired* lease;
+* un-re-lockable — the branch had by then advanced past its base;
+* and rejected by the #447 create-PR provenance guard.
+
+Each of those gates is individually correct. The defect was that two writers
+disagreed about what a lock *is*. This module is the single definition, and both
+writers now build through it.
+
+Nothing here weakens a guard. ``build_sanctioned_lock_provenance`` remains the
+only provenance source, provenance is never accepted from a caller, and the
+#447 guard is untouched — this module simply makes bootstrap satisfy it.
+"""
+
+from __future__ import annotations
+
+from datetime import datetime, timedelta, timezone
+from typing import Any, Mapping
+
+import issue_lock_provenance
+import issue_lock_store
+import lease_policy
+
+# Bootstrap writes through the same sanctioned source as gitea_lock_issue: the
+# lock it produces *is* a canonical lock, not a second dialect that readers must
+# learn. Adding a distinct source would have required widening
+# SANCTIONED_LOCK_SOURCES, which is exactly the #447 weakening this issue's
+# safety requirements forbid.
+SOURCE_BOOTSTRAP = issue_lock_provenance.SOURCE_LOCK_ISSUE
+
+# Recovery of an incomplete bootstrap lock (#953 AC8-AC11) deliberately writes
+# through SOURCE_LOCK_ISSUE too, and records its distinctness in
+# ``lock_provenance.written_by_tool`` plus the ``bootstrap_lock_recovery``
+# transition block instead. There is no distinct recovery *source* constant, for
+# the same reason bootstrap has none: minting one would require widening
+# SANCTIONED_LOCK_SOURCES, which the #447 safety requirements forbid.
+
+#: Top-level keys every canonical author issue lock must carry.
+REQUIRED_LOCK_FIELDS: tuple[str, ...] = (
+    "remote",
+    "org",
+    "repo",
+    "issue_number",
+    "branch_name",
+    "worktree_path",
+    "work_lease",
+    "lock_provenance",
+)
+
+#: Keys every canonical ``work_lease`` must carry.
+REQUIRED_WORK_LEASE_FIELDS: tuple[str, ...] = (
+    "operation_type",
+    "issue_number",
+    "branch",
+    "worktree_path",
+    "claimant",
+    "created_at",
+    "expires_at",
+    "last_heartbeat_at",
+    "task_session_id",
+    "lifecycle_version",
+)
+
+# ── Explicit expiration states (AC12) ──
+# The bug this replaces: a lock with no recorded expiry produced
+# ``is_lease_expired() -> False``, which reads as "not yet expired" and made the
+# lock permanently non-expiring *and* permanently ineligible for the renewal
+# path, which only ever assesses an expired lease. "Absent" and "in the future"
+# are different facts and are now named differently.
+EXPIRATION_RECORDED = "recorded"
+EXPIRATION_MISSING = "missing"
+EXPIRATION_UNPARSEABLE = "unparseable"
+
+#: Structural verdicts returned by :func:`assess_lock_contract`.
+CONTRACT_CANONICAL = "canonical"
+CONTRACT_INCOMPLETE = "incomplete"
+CONTRACT_LEGACY = "legacy"
+CONTRACT_ABSENT = "absent"
+
+
+def _text(value: Any) -> str:
+    return str(value or "").strip()
+
+
+def now_utc() -> datetime:
+    return datetime.now(timezone.utc)
+
+
+def format_timestamp(value: datetime) -> str:
+    """Serialize in the durable ``...Z`` form already used on disk."""
+    return (
+        value.astimezone(timezone.utc)
+        .replace(microsecond=0)
+        .isoformat()
+        .replace("+00:00", "Z")
+    )
+
+
+def lock_claimant(lock: Mapping[str, Any] | None) -> dict[str, str]:
+    """Read the claimant from either canonical or legacy placement.
+
+    ``work_lease.claimant`` is canonical and is preferred. A top-level
+    ``claimant`` is the legacy/bootstrap placement and is accepted as a
+    fallback (AC14) — three separate readers already disagreed about this
+    (``issue_lock_store``, ``issue_lock_renewal``, ``issue_lock_recovery``),
+    which is why it now lives in one place.
+
+    Reading a legacy placement is *not* a widening: every caller still compares
+    the values it returns against server-resolved identity and profile. This
+    only decides where to look, never whether ownership is proven.
+
+    Delegates to ``issue_lock_store.lock_claimant`` rather than reimplementing
+    the rule. A second copy here would be a fourth reader that could drift from
+    the other three, which is the exact failure #953 exists to end. It lives in
+    the store because ``author_lock_contract`` imports the store, so defining it
+    here would make that import circular.
+    """
+    recorded = issue_lock_store.lock_claimant(dict(lock) if isinstance(lock, Mapping) else None)
+    return {
+        "username": _text(recorded.get("username")),
+        "profile": _text(recorded.get("profile")),
+    }
+
+
+def claimant_placement(lock: Mapping[str, Any] | None) -> str:
+    """Where the claimant was found: ``work_lease``, ``top_level``, or ``absent``."""
+    if not isinstance(lock, Mapping):
+        return "absent"
+    lease = lock.get("work_lease")
+    if isinstance(lease, Mapping) and isinstance(lease.get("claimant"), Mapping):
+        return "work_lease"
+    if isinstance(lock.get("claimant"), Mapping):
+        return "top_level"
+    return "absent"
+
+
+def build_claimant(*, username: str | None, profile: str | None) -> dict[str, str]:
+    """Build the canonical claimant pair from server-resolved values."""
+    return {"username": _text(username), "profile": _text(profile)}
+
+
+def build_author_issue_work_lease(
+    *,
+    issue_number: int,
+    branch_name: str,
+    worktree_path: str,
+    claimant: Mapping[str, Any],
+    task_session_id: str | None = None,
+    created: datetime | None = None,
+) -> dict[str, Any]:
+    """Build the canonical author ``work_lease``.
+
+    The single definition behind both writers. The TTL comes from the central
+    policy rather than a literal, and the window slides from the last valid
+    heartbeat (#790), so an abandoned task releases its claim within one TTL.
+    """
+    started = created or now_utc()
+    policy = lease_policy.policy_for(lease_policy.TASK_CLASS_AUTHOR_ISSUE_WORK)
+    expires = started + timedelta(minutes=policy.initial_ttl_minutes)
+    session_id = _text(task_session_id) or issue_lock_store.mint_task_session_id(
+        issue_lock_store.AUTHOR_ISSUE_WORK_LEASE
+    )
+    return {
+        "operation_type": issue_lock_store.AUTHOR_ISSUE_WORK_LEASE,
+        "issue_number": int(issue_number),
+        "pr_number": None,
+        "branch": branch_name,
+        "worktree_path": worktree_path,
+        "claimant": dict(claimant),
+        "created_at": format_timestamp(started),
+        "expires_at": format_timestamp(expires),
+        "last_heartbeat_at": format_timestamp(started),
+        # #790 AC-N1: the ownership key for this task, distinct from the
+        # recorded PID, which is the shared daemon and identifies no task.
+        "task_session_id": session_id,
+        # #790 AC-N8: the explicit lifecycle marker. Its absence — never a
+        # timestamp comparison — is what makes a lock legacy.
+        "lifecycle_version": lease_policy.LIFECYCLE_HEARTBEAT_V1,
+        "heartbeat_count": 1,
+    }
+
+
+def build_canonical_issue_lock(
+    *,
+    issue_number: int,
+    branch_name: str,
+    worktree_path: str,
+    remote: str,
+    org: str,
+    repo: str,
+    identity: str | None,
+    profile: str | None,
+    tool: str,
+    source: str = issue_lock_provenance.SOURCE_LOCK_ISSUE,
+    owner_session: str | None = None,
+    assignment_id: str | None = None,
+    lease_id: str | None = None,
+    expected_base_sha: str | None = None,
+    task_session_id: str | None = None,
+    created: datetime | None = None,
+) -> dict[str, Any]:
+    """Build a complete canonical lock record.
+
+    ``tool`` and ``source`` are server-supplied. There is deliberately no
+    parameter through which a caller could inject provenance: the #953 safety
+    requirements forbid caller-manufactured provenance, so provenance is always
+    minted here from ``build_sanctioned_lock_provenance``.
+    """
+    claimant = build_claimant(username=identity, profile=profile)
+    work_lease = build_author_issue_work_lease(
+        issue_number=issue_number,
+        branch_name=branch_name,
+        worktree_path=worktree_path,
+        claimant=claimant,
+        task_session_id=task_session_id,
+        created=created,
+    )
+    record: dict[str, Any] = {
+        "remote": remote,
+        "org": org,
+        "repo": repo,
+        "issue_number": int(issue_number),
+        "branch": branch_name,
+        "branch_name": branch_name,
+        "worktree_path": worktree_path,
+        "work_lease": work_lease,
+        "lock_provenance": issue_lock_provenance.build_sanctioned_lock_provenance(
+            tool=tool,
+            source=source,
+            claimant=claimant,
+        ),
+    }
+    if owner_session is not None:
+        record["owner_session"] = owner_session
+    if assignment_id is not None:
+        record["assignment_id"] = assignment_id
+    # #953 AC6: a null lease id is recorded only when no workflow lease was
+    # allocated for this bootstrap. The task-session identifier in the
+    # work_lease is what downstream ownership checks fence on, and it is never
+    # null on a canonical lock.
+    if lease_id is not None:
+        record["lease_id"] = lease_id
+    if expected_base_sha is not None:
+        record["expected_base_sha"] = expected_base_sha
+    return record
+
+
+def expiration_state(lock: Mapping[str, Any] | None) -> dict[str, Any]:
+    """Classify a lock's recorded expiry explicitly (AC12).
+
+    Distinguishes "no expiry was ever recorded" from "an expiry was recorded
+    and is still in the future". Collapsing those two into a single ``False``
+    from ``is_lease_expired`` is what let a malformed lock be treated as
+    permanently live and simultaneously never renewable.
+    """
+    if not isinstance(lock, Mapping):
+        return {"state": EXPIRATION_MISSING, "expires_at": None, "expired": None}
+    lease = lock.get("work_lease")
+    raw = lease.get("expires_at") if isinstance(lease, Mapping) else None
+    text = _text(raw)
+    if not text:
+        return {"state": EXPIRATION_MISSING, "expires_at": None, "expired": None}
+    try:
+        parsed = datetime.fromisoformat(text.replace("Z", "+00:00")).astimezone(
+            timezone.utc
+        )
+    except ValueError:
+        return {"state": EXPIRATION_UNPARSEABLE, "expires_at": text, "expired": None}
+    return {
+        "state": EXPIRATION_RECORDED,
+        "expires_at": text,
+        "expired": parsed <= now_utc(),
+    }
+
+
+def missing_contract_fields(lock: Mapping[str, Any] | None) -> list[str]:
+    """Name every canonical field a lock does not carry (AC7)."""
+    if not isinstance(lock, Mapping):
+        return ["<no lock record>"]
+    missing: list[str] = []
+    for field in REQUIRED_LOCK_FIELDS:
+        value = lock.get(field)
+        if value is None or (isinstance(value, str) and not value.strip()):
+            missing.append(field)
+    lease = lock.get("work_lease")
+    if not isinstance(lease, Mapping):
+        if "work_lease" not in missing:
+            missing.append("work_lease")
+    else:
+        for field in REQUIRED_WORK_LEASE_FIELDS:
+            value = lease.get(field)
+            if value is None or (isinstance(value, str) and not value.strip()):
+                missing.append(f"work_lease.{field}")
+    provenance = lock.get("lock_provenance")
+    if isinstance(provenance, Mapping):
+        if (
+            _text(provenance.get("source"))
+            not in issue_lock_provenance.SANCTIONED_LOCK_SOURCES
+        ):
+            missing.append("lock_provenance.source (not sanctioned)")
+        if not _text(provenance.get("written_by_tool")):
+            missing.append("lock_provenance.written_by_tool")
+    claimant = lock_claimant(lock)
+    if not claimant["username"]:
+        missing.append("claimant.username")
+    if not claimant["profile"]:
+        missing.append("claimant.profile")
+    return missing
+
+
+def assess_lock_contract(lock: Mapping[str, Any] | None) -> dict[str, Any]:
+    """Structural, read-only verdict on a durable lock record (AC7, AC16).
+
+    Pure inspection: it reads the record it is handed and mutates nothing —
+    no lock, lease, branch, worktree, issue, or PR. Callers use it both to
+    verify a lock they just wrote and to report on one they found.
+    """
+    if not isinstance(lock, Mapping) or not lock:
+        return {
+            "contract": CONTRACT_ABSENT,
+            "canonical": False,
+            "missing_fields": ["<no lock record>"],
+            "claimant": {"username": "", "profile": ""},
+            "claimant_placement": "absent",
+            "expiration": {
+                "state": EXPIRATION_MISSING,
+                "expires_at": None,
+                "expired": None,
+            },
+            "heartbeatable": False,
+            "create_pr_eligible": False,
+            "lock_generation": None,
+            "task_session_id": None,
+            "reasons": ["no durable lock record"],
+        }
+
+    missing = missing_contract_fields(lock)
+    claimant = lock_claimant(lock)
+    placement = claimant_placement(lock)
+    expiration = expiration_state(lock)
+    provenance_check = issue_lock_provenance.assess_lock_file_for_create_pr(dict(lock))
+
+    # Canonical means: every required field present, the claimant in the
+    # canonical placement, an expiry actually recorded, and the untouched #447
+    # guard satisfied.
+    canonical = (
+        not missing
+        and placement == "work_lease"
+        and expiration["state"] == EXPIRATION_RECORDED
+        and bool(provenance_check.get("proven"))
+    )
+    if canonical:
+        contract = CONTRACT_CANONICAL
+    elif placement == "top_level" and claimant["username"] and claimant["profile"]:
+        contract = CONTRACT_LEGACY
+    else:
+        contract = CONTRACT_INCOMPLETE
+
+    reasons: list[str] = []
+    if missing:
+        reasons.append("missing canonical fields: " + ", ".join(missing))
+    if placement == "top_level":
+        reasons.append(
+            "claimant recorded at the lock top level rather than in work_lease "
+            "(legacy/bootstrap placement)"
+        )
+    if expiration["state"] == EXPIRATION_MISSING:
+        reasons.append(
+            "no expiration recorded; the lock is neither expirable nor renewable "
+            "until it is upgraded"
+        )
+    elif expiration["state"] == EXPIRATION_UNPARSEABLE:
+        reasons.append(f"unparseable expires_at '{expiration['expires_at']}'")
+    if provenance_check.get("block"):
+        reasons.extend(provenance_check.get("reasons") or [])
+
+    # Heartbeat needs the claimant pair (from either placement, post-fix) plus a
+    # task-session identifier to fence on.
+    lease = lock.get("work_lease")
+    task_session_id = (
+        _text(lease.get("task_session_id")) if isinstance(lease, Mapping) else ""
+    )
+    heartbeatable = bool(
+        claimant["username"] and claimant["profile"] and task_session_id
+    )
+
+    return {
+        "contract": contract,
+        "canonical": canonical,
+        "missing_fields": missing,
+        "claimant": claimant,
+        "claimant_placement": placement,
+        "expiration": expiration,
+        "heartbeatable": heartbeatable,
+        "create_pr_eligible": bool(provenance_check.get("proven")),
+        "lock_generation": lock.get("lock_generation"),
+        "task_session_id": task_session_id or None,
+        "reasons": reasons,
+    }
+
+
+def format_contract_refusal(assessment: Mapping[str, Any]) -> str:
+    """Human-readable refusal naming exactly what the lock is missing."""
+    missing = ", ".join(assessment.get("missing_fields") or []) or "unknown fields"
+    return (
+        "Issue lock contract incomplete (#953): "
+        f"{missing}. The lock cannot be heartbeated, renewed, or accepted by "
+        "gitea_create_pr in this state (fail closed)"
+    )
+
+
+def recommended_action(assessment: Mapping[str, Any]) -> str:
+    """The one executable next step for a lock in this state (AC5, AC15)."""
+    contract = assessment.get("contract")
+    if contract == CONTRACT_CANONICAL:
+        return (
+            "Lock is canonical. Call gitea_whoami, then "
+            "gitea_resolve_task_capability(task='work_issue'), then proceed with "
+            "author implementation in the bootstrapped worktree."
+        )
+    if contract == CONTRACT_ABSENT:
+        return (
+            "No durable lock exists. Call gitea_lock_issue for this issue and "
+            "branch before writing any implementation bytes."
+        )
+    return (
+        "Do not begin implementation. Call "
+        "gitea_recover_incomplete_bootstrap_lock for this exact issue, branch, "
+        "and worktree to upgrade the lock to the canonical contract, or "
+        "gitea_lock_issue while the worktree is still base-equivalent."
+    )
+
+
+# ── Post-compensation recovery guidance (#953 AC5/AC15, review 632 F2) ──
+#
+# ``recommended_action`` above answers "what can be done about a lock in this
+# shape". That is the wrong question on the bootstrap AC7 refusal path, because
+# ``run_compensating_recovery`` has already run by the time the answer is
+# reported: it releases the lock, removes the worktree when clean — which it
+# always is there, no implementation bytes having been written — and deletes the
+# created branch. Recommending incomplete-lock recovery for those artifacts
+# hands the author two refusals in a row (``no_durable_lock``, then
+# ``worktree_invalid``) for a state that a plain bootstrap retry would fix. The
+# advice must describe the state that actually *remains*.
+
+#: Compensation removed every artifact this transition created.
+CLEANUP_COMPLETE = "complete"
+#: Compensation removed some artifacts; others survive and are still actionable.
+CLEANUP_PARTIAL = "partial"
+#: Compensation itself failed or could not be observed; nothing is provable.
+CLEANUP_FAILED = "failed"
+
+
+def assess_post_compensation_state(
+    recovery: Mapping[str, Any] | None,
+    *,
+    lock_present: bool,
+    worktree_present: bool,
+    branch_present: bool,
+) -> dict[str, Any]:
+    """Classify what survived compensation, from observed durable state.
+
+    Pure. The caller observes the filesystem and git; this decides. Observation
+    is authoritative over the journal's ``rolled_back`` list, which records what
+    compensation *attempted*: ``run_compensating_recovery`` swallows a failed
+    lock release and appends nothing, so an absent marker proves nothing either
+    way. The list is still carried through as corroborating evidence.
+
+    The three states are distinct facts, not degrees of the same one:
+
+    * ``CLEANUP_COMPLETE`` — compensation ran and nothing it created remains.
+    * ``CLEANUP_PARTIAL`` — compensation ran and artifacts survive, whether by
+      design (a worktree dirty at rollback time, a branch carrying commits) or
+      because a rollback step errored. Either way the surviving set was observed
+      directly, so it is known and actionable; ``failed_rollback_steps`` records
+      which cause applies.
+    * ``CLEANUP_FAILED`` — compensation never ran to completion, so nothing it
+      would have removed can be assumed removed.
+    """
+    rolled_back = list((recovery or {}).get("rolled_back") or [])
+    executed = bool((recovery or {}).get("executed"))
+    failed_steps = [entry for entry in rolled_back if "_failed" in entry]
+
+    surviving: list[str] = []
+    if lock_present:
+        surviving.append("lock")
+    if worktree_present:
+        surviving.append("worktree")
+    if branch_present:
+        surviving.append("branch")
+
+    if not executed:
+        state = CLEANUP_FAILED
+    elif surviving:
+        state = CLEANUP_PARTIAL
+    else:
+        state = CLEANUP_COMPLETE
+
+    return {
+        "cleanup_state": state,
+        "compensation_executed": executed,
+        "lock_present": bool(lock_present),
+        "worktree_present": bool(worktree_present),
+        "branch_present": bool(branch_present),
+        "surviving_artifacts": surviving,
+        "removed_artifacts": [
+            name
+            for name, present in (
+                ("lock", lock_present),
+                ("worktree", worktree_present),
+                ("branch", branch_present),
+            )
+            if not present
+        ],
+        "failed_rollback_steps": failed_steps,
+        "rolled_back": rolled_back,
+    }
+
+
+def post_compensation_action(
+    state: Mapping[str, Any],
+    *,
+    issue_number: int,
+    branch_name: str,
+    worktree_path: str,
+    missing_fields: list[str] | None = None,
+) -> str:
+    """The one executable next step for the state compensation actually left.
+
+    Every branch names only artifacts the classification says still exist, so no
+    recommendation can point at something the rollback deleted.
+    """
+    missing = ", ".join(missing_fields or []) or "the reported missing fields"
+    cleanup_state = state.get("cleanup_state")
+    lock_present = bool(state.get("lock_present"))
+    worktree_present = bool(state.get("worktree_present"))
+    branch_present = bool(state.get("branch_present"))
+
+    if not state.get("compensation_executed"):
+        # Compensation never ran, so nothing was rolled back and nothing about
+        # the remaining state was decided. The read-only surface is the only
+        # action executable under any state.
+        return (
+            "Compensating rollback did not complete, so the remaining state is "
+            f"not proven. Call gitea_inspect_issue_lock_contract for issue "
+            f"#{issue_number} (read-only) to establish what survives before any "
+            "further action. Do not retry bootstrap until it is known."
+        )
+
+    prefix = ""
+    failed_steps = state.get("failed_rollback_steps") or []
+    if failed_steps:
+        prefix = (
+            "Compensating rollback reported a failed step "
+            f"({', '.join(failed_steps)}); what survives was observed directly "
+            "and the action below is scoped to exactly that. "
+        )
+
+    if cleanup_state == CLEANUP_COMPLETE:
+        return (
+            "Compensating rollback removed the malformed lock, the branch, and "
+            f"the worktree, so nothing from this attempt remains. Resolve "
+            f"{missing} and re-run gitea_bootstrap_author_issue_worktree for "
+            f"issue #{issue_number} from the clean pre-bootstrap state. Do not "
+            "call gitea_recover_incomplete_bootstrap_lock: there is no lock, "
+            "branch, or worktree left for it to act on."
+        )
+
+    if lock_present and worktree_present and branch_present:
+        return prefix + (
+            "The lock, branch, and worktree all survive. Call "
+            "gitea_recover_incomplete_bootstrap_lock for issue "
+            f"#{issue_number}, branch '{branch_name}', and worktree "
+            f"'{worktree_path}', passing the worktree's current head as "
+            "expected_head, to upgrade the lock to the canonical contract."
+        )
+
+    if not lock_present and worktree_present and branch_present:
+        return prefix + (
+            "The malformed lock was released but the branch and worktree "
+            "survive. No implementation bytes were written, so the worktree is "
+            f"still base-equivalent: call gitea_lock_issue for issue "
+            f"#{issue_number} on branch '{branch_name}' from worktree "
+            f"'{worktree_path}' to acquire a canonical lock."
+        )
+
+    if lock_present and not worktree_present:
+        return prefix + (
+            f"The worktree for issue #{issue_number} is gone but the durable "
+            "lock survived, so neither gitea_recover_incomplete_bootstrap_lock "
+            "(it would refuse worktree_invalid) nor gitea_lock_issue (it has no "
+            "worktree to bind) is executable. Call "
+            "gitea_inspect_issue_lock_contract for issue "
+            f"#{issue_number} (read-only) to confirm the surviving lock; it "
+            "must be released by its recorded owner before bootstrap is "
+            "retried."
+        )
+
+    # Lock gone, worktree gone, some git artifact left (a branch with commits,
+    # or a branch this transition did not create).
+    return prefix + (
+        "Compensating rollback removed the lock and worktree; branch "
+        f"'{branch_name}' survives and was not deleted. Call "
+        f"gitea_inspect_issue_lock_contract for issue #{issue_number} "
+        "(read-only) to confirm no durable lock remains, then re-run "
+        "gitea_bootstrap_author_issue_worktree, which will adopt the existing "
+        "branch rather than recreating it."
+    )

bootstrap_lock_recovery.py

@@ -0,0 +1,304 @@
+"""Target-specific recovery for incomplete bootstrap issue locks (#953).
+
+The situation this exists for: ``gitea_bootstrap_author_issue_worktree``
+reported success, wrote an incomplete lock, and told the author to implement.
+The author did — legitimately, following the tool's own reported next action —
+and the branch now carries real committed and pushed work. At that point every
+pre-existing recovery path is simultaneously ineligible:
+
+* heartbeat refuses, because the claimant is not where it looks;
+* ``gitea_lock_issue`` refuses, because the branch is no longer base-equivalent;
+* #760 exact-owner renewal never engages, because a lock with no recorded
+  expiry is never *expired*;
+* the #447 create-PR guard refuses, because there is no provenance.
+
+Distinct from every neighbouring path: #753 ``issue_lock_recovery`` requires a
+dead owner PID, #760 ``issue_lock_renewal`` requires an *expired* lease, and
+#442 ``issue_lock_adoption`` decides branch adoption. None of them addresses a
+lock that is structurally incomplete and therefore never expires at all.
+
+**What this will not do.** It never moves, resets, or rewinds a branch, and
+never requires base-equivalence — the committed work is the thing being
+preserved. It never pushes and never opens a pull request. It touches only the
+one lock file named by (remote, org, repo, issue). It accepts no caller-supplied
+provenance and no caller-supplied authorization flag; both are minted
+server-side. It refuses a healthy foreign-owned lock outright, and a matching
+username alone is never accepted as proof of ownership — the profile must match
+too, and the lock's recorded binding must agree with the observed branch,
+worktree, and head.
+"""
+
+from __future__ import annotations
+
+import os
+from typing import Any, Mapping
+
+import author_lock_contract
+import issue_lock_store
+
+#: Refusal codes, so callers can branch on cause rather than parse prose.
+REFUSAL_NO_LOCK = "no_durable_lock"
+REFUSAL_ALREADY_CANONICAL = "already_canonical"
+REFUSAL_FOREIGN_CLAIMANT = "foreign_claimant"
+REFUSAL_HEALTHY_FOREIGN = "healthy_foreign_lock"
+REFUSAL_IDENTITY_UNRESOLVED = "identity_unresolved"
+REFUSAL_BINDING_MISMATCH = "binding_mismatch"
+REFUSAL_WORKTREE_INVALID = "worktree_invalid"
+REFUSAL_HEAD_MISMATCH = "head_mismatch"
+
+
+def _text(value: Any) -> str:
+    return str(value or "").strip()
+
+
+def _same_realpath(left: str | None, right: str | None) -> bool:
+    lhs, rhs = _text(left), _text(right)
+    if not lhs or not rhs:
+        return False
+    try:
+        return os.path.realpath(lhs) == os.path.realpath(rhs)
+    except OSError:
+        return lhs == rhs
+
+
+def assess_bootstrap_lock_recovery(
+    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,
+    observed_head: str | None,
+    declared_head: str | None,
+    worktree_exists: bool,
+    worktree_registered: bool,
+    current_branch: str | None,
+    now: Any = None,
+) -> dict[str, Any]:
+    """Decide whether this exact lock may be upgraded by this exact caller.
+
+    Pure: every input is an observation the caller already made, and nothing
+    here reads or writes the filesystem, git, or Gitea. That is what makes the
+    same decision testable in isolation and reusable by the read-only
+    inspection surface, which must not mutate anything (AC16).
+
+    Returns a dict with ``recovery_sanctioned`` plus the full evidence set. A
+    refusal never raises — it reports, so the caller can surface exactly which
+    piece of evidence was missing.
+    """
+    reasons: list[str] = []
+    refusal_code: str | None = None
+
+    contract = author_lock_contract.assess_lock_contract(existing_lock)
+
+    if not existing_lock:
+        return {
+            "recovery_sanctioned": False,
+            "refusal_code": REFUSAL_NO_LOCK,
+            "reasons": [
+                f"no durable issue lock exists for issue #{issue_number}; there is "
+                "nothing to recover (fail closed)"
+            ],
+            "contract": contract,
+            "evidence": {},
+            "expected_generation": None,
+        }
+
+    active_identity = _text(identity)
+    active_profile = _text(profile)
+    recorded = author_lock_contract.lock_claimant(existing_lock)
+    freshness = issue_lock_store.assess_lock_freshness(dict(existing_lock), now=now)
+    generation = issue_lock_store.lock_generation(existing_lock)
+
+    evidence: dict[str, Any] = {
+        "recorded_claimant": recorded,
+        "active_identity": active_identity,
+        "active_profile": active_profile,
+        "recorded_branch": existing_lock.get("branch_name"),
+        "recorded_worktree": existing_lock.get("worktree_path"),
+        "recorded_owner_session": existing_lock.get("owner_session"),
+        "recorded_generation": generation,
+        "recorded_remote": existing_lock.get("remote"),
+        "recorded_org": existing_lock.get("org"),
+        "recorded_repo": existing_lock.get("repo"),
+        "observed_head": _text(observed_head),
+        "declared_head": _text(declared_head),
+        "current_branch": _text(current_branch),
+        "worktree_exists": bool(worktree_exists),
+        "worktree_registered": bool(worktree_registered),
+        "freshness": freshness,
+        "claimant_placement": contract.get("claimant_placement"),
+        "expiration_state": contract.get("expiration", {}).get("state"),
+    }
+
+    # ── Repository and issue identity (AC10) ──
+    if _text(existing_lock.get("remote")) != _text(remote):
+        reasons.append(
+            f"recorded remote '{existing_lock.get('remote')}' does not match '{remote}'"
+        )
+        refusal_code = refusal_code or REFUSAL_BINDING_MISMATCH
+    if _text(existing_lock.get("org")) != _text(org):
+        reasons.append(
+            f"recorded org '{existing_lock.get('org')}' does not match '{org}'"
+        )
+        refusal_code = refusal_code or REFUSAL_BINDING_MISMATCH
+    if _text(existing_lock.get("repo")) != _text(repo):
+        reasons.append(
+            f"recorded repo '{existing_lock.get('repo')}' does not match '{repo}'"
+        )
+        refusal_code = refusal_code or REFUSAL_BINDING_MISMATCH
+    if existing_lock.get("issue_number") != issue_number:
+        reasons.append(
+            f"lock targets issue #{existing_lock.get('issue_number')}, not "
+            f"#{issue_number}"
+        )
+        refusal_code = refusal_code or REFUSAL_BINDING_MISMATCH
+
+    # ── Branch and worktree binding (AC10) ──
+    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}'"
+        )
+        refusal_code = refusal_code or REFUSAL_BINDING_MISMATCH
+    if not _same_realpath(existing_lock.get("worktree_path"), worktree_path):
+        reasons.append(
+            f"recorded worktree '{existing_lock.get('worktree_path')}' does not "
+            f"match '{worktree_path}'"
+        )
+        refusal_code = refusal_code or REFUSAL_BINDING_MISMATCH
+
+    # ── The worktree is real, registered, and on the branch (AC10) ──
+    # Deliberately no base-equivalence requirement and no constraint on how far
+    # the branch has advanced: the whole point is that it already carries the
+    # author's legitimate commits (AC9).
+    if not worktree_exists:
+        reasons.append(f"declared worktree '{worktree_path}' does not exist")
+        refusal_code = refusal_code or REFUSAL_WORKTREE_INVALID
+    if not worktree_registered:
+        reasons.append(f"worktree '{worktree_path}' is not a registered git worktree")
+        refusal_code = refusal_code or REFUSAL_WORKTREE_INVALID
+    if _text(current_branch) != _text(branch_name):
+        reasons.append(
+            f"worktree is on branch '{_text(current_branch) or 'unknown'}', not "
+            f"'{branch_name}'"
+        )
+        refusal_code = refusal_code or REFUSAL_WORKTREE_INVALID
+
+    # ── Current head fencing (AC10) ──
+    # The caller names the commit it believes it is recovering. A mismatch means
+    # the worktree moved under the caller, so the decision is stale.
+    if not _text(observed_head):
+        reasons.append("could not observe the worktree head")
+        refusal_code = refusal_code or REFUSAL_HEAD_MISMATCH
+    elif _text(declared_head) and _text(declared_head) != _text(observed_head):
+        reasons.append(
+            f"declared head '{_text(declared_head)}' does not match observed head "
+            f"'{_text(observed_head)}'"
+        )
+        refusal_code = refusal_code or REFUSAL_HEAD_MISMATCH
+
+    # ── Ownership (AC10, AC11) ──
+    # A matching username alone is never sufficient: the profile must match too,
+    # and both are compared against server-resolved values the caller cannot set.
+    if not active_identity or not active_profile:
+        reasons.append(
+            "active identity and profile could not both be resolved; ownership "
+            "cannot be proven"
+        )
+        refusal_code = refusal_code or REFUSAL_IDENTITY_UNRESOLVED
+    if not recorded["username"] or not recorded["profile"]:
+        reasons.append(
+            "durable lock does not record both a claimant username and profile"
+        )
+        refusal_code = refusal_code or REFUSAL_FOREIGN_CLAIMANT
+    elif (
+        recorded["username"] != active_identity
+        or recorded["profile"] != active_profile
+    ):
+        # AC11: a foreign-owned lock is never recoverable through this path,
+        # healthy or not. The healthy case is reported distinctly so the refusal
+        # is legible, but both refuse.
+        if freshness.get("live"):
+            reasons.append(
+                f"lock is owned by a healthy foreign claimant "
+                f"'{recorded['username']}/{recorded['profile']}'; takeover is not "
+                "a recovery path"
+            )
+            refusal_code = REFUSAL_HEALTHY_FOREIGN
+        else:
+            reasons.append(
+                f"lock claimant '{recorded['username']}/{recorded['profile']}' "
+                f"does not match active '{active_identity}/{active_profile}'"
+            )
+            refusal_code = refusal_code or REFUSAL_FOREIGN_CLAIMANT
+
+    # ── Nothing to recover ──
+    # A lock that is already canonical is left strictly alone. Rewriting it would
+    # mint a new task-session identifier and invalidate the heartbeat token the
+    # legitimate owner is already using.
+    if contract.get("canonical") and not reasons:
+        return {
+            "recovery_sanctioned": False,
+            "refusal_code": REFUSAL_ALREADY_CANONICAL,
+            "reasons": [
+                "lock already satisfies the canonical contract; no recovery is "
+                "required"
+            ],
+            "contract": contract,
+            "evidence": evidence,
+            "expected_generation": generation,
+        }
+
+    sanctioned = not reasons
+    return {
+        "recovery_sanctioned": sanctioned,
+        "refusal_code": None if sanctioned else refusal_code,
+        "reasons": reasons,
+        "contract": contract,
+        "evidence": evidence,
+        "expected_generation": generation,
+    }
+
+
+def build_recovery_record(
+    assessment: Mapping[str, Any],
+    *,
+    recovered_at: str,
+    new_task_session_id: str,
+) -> dict[str, Any]:
+    """Auditable record of the ownership and generation transition (AC10).
+
+    A recovered lock must never read as an original claim, so both sides of the
+    transition are preserved: what the incomplete lock recorded, and what
+    replaced it.
+    """
+    evidence = dict(assessment.get("evidence") or {})
+    contract = dict(assessment.get("contract") or {})
+    return {
+        "recovery_kind": "incomplete_bootstrap_lock",
+        "recovered_at": recovered_at,
+        "prior_contract": contract.get("contract"),
+        "prior_missing_fields": list(contract.get("missing_fields") or []),
+        "prior_claimant_placement": evidence.get("claimant_placement"),
+        "prior_expiration_state": evidence.get("expiration_state"),
+        "prior_generation": evidence.get("recorded_generation"),
+        "prior_owner_session": evidence.get("recorded_owner_session"),
+        "prior_freshness": (evidence.get("freshness") or {}).get("status"),
+        "replacement_task_session_id": new_task_session_id,
+        "preserved_head": evidence.get("observed_head"),
+        "branch_reset": False,
+        "base_equivalence_required": False,
+    }
+
+
+def format_recovery_refusal(assessment: Mapping[str, Any]) -> str:
+    reasons = "; ".join(
+        assessment.get("reasons") or ["unknown bootstrap lock recovery refusal"]
+    )
+    code = assessment.get("refusal_code") or "refused"
+    return f"Bootstrap lock recovery refused ({code}): {reasons} (fail closed)"

control_plane_db.py

@@ -1568,6 +1568,32 @@ class ControlPlaneDB:
             ).fetchone()
             return dict(row) if row else None
 
+    def list_incident_links(
+        self,
+        *,
+        provider: str | None = None,
+        gitea_org: str | None = None,
+        gitea_repo: str | None = None,
+        limit: int = 100,
+    ) -> list[dict[str, Any]]:
+        """List stored incident_links rows, optionally filtered by provider/repo (#612 / #649)."""
+        query = "SELECT * FROM incident_links WHERE 1=1"
+        params: list[Any] = []
+        if provider:
+            query += " AND provider = ?"
+            params.append(provider.strip().lower())
+        if gitea_org:
+            query += " AND gitea_org = ?"
+            params.append(_norm_scope(gitea_org))
+        if gitea_repo:
+            query += " AND gitea_repo = ?"
+            params.append(_norm_scope(gitea_repo))
+        query += " ORDER BY link_id DESC LIMIT ?"
+        params.append(max(1, limit))
+        with self._tx(immediate=False) as conn:
+            rows = conn.execute(query, params).fetchall()
+            return [dict(r) for r in rows]
+
 
     # ── lease lifecycle (#601) ────────────────────────────────────────────
 

docs/author-issue-lock-contract.md

@@ -0,0 +1,209 @@
+# The canonical author issue-lock contract (#953)
+
+Every author issue lock has exactly one shape. Both writers —
+`gitea_bootstrap_author_issue_worktree` and `gitea_lock_issue` — build it
+through `author_lock_contract.build_canonical_issue_lock`, and every reader
+consumes that same shape.
+
+Before #953 the two writers disagreed. `gitea_lock_issue` wrote the canonical
+record; bootstrap wrote a thinner one with the claimant at the lock top level,
+`lease_id: null`, and no `work_lease`, `lock_provenance`, or expiry. Because
+every reader was written against the canonical shape, a lock that bootstrap
+reported as successfully created could not be heartbeated, renewed, re-locked,
+or accepted by `gitea_create_pr`. Each of those gates was individually correct;
+the defect was that two writers disagreed about what a lock *is*.
+
+## Required ordering
+
+**Finalize the lock before writing any implementation bytes.** This ordering is
+what keeps recovery cheap: while the worktree is still base-equivalent, a lock
+problem can be fixed by simply calling `gitea_lock_issue` again. Once the branch
+carries commits, base-equivalence is gone and the ordinary re-lock path is no
+longer available.
+
+1. `gitea_whoami` — resolve identity and profile.
+2. `gitea_resolve_task_capability(task='work_issue')`.
+3. `gitea_bootstrap_author_issue_worktree` — creates the branch, the registered
+   worktree under `branches/`, and a **canonical** lock. It reads the lock back
+   and verifies it structurally before reporting success; a partial lock fails
+   closed here, with the missing fields named, and never reports
+   `implementation_allowed: true`.
+4. `gitea_heartbeat_issue_lock` — prove the lock is usable, using the
+   `task_session_id` bootstrap returned.
+5. Implement, commit, push.
+6. `gitea_create_pr`.
+
+If bootstrap returns `success: false` with
+`reason_code: incomplete_issue_lock_contract`, **do not implement**. Its
+`exact_next_action` names the executable recovery step. Bootstrap's reported
+next action always matches the state it actually returned.
+
+### What that refusal leaves behind
+
+The AC7 refusal runs `run_compensating_recovery` *before* it reports, so the
+advice has to describe the post-rollback state rather than the shape of the lock
+that provoked it. Recommending incomplete-lock recovery for artifacts the
+rollback already deleted would produce `no_durable_lock` and then
+`worktree_invalid` — two refusals for a state a plain retry fixes.
+
+The refusal therefore carries `compensating_recovery` and
+`post_compensation_state`, and derives `exact_next_action` from what was
+observed on disk. `cleanup_state` is one of:
+
+| `cleanup_state` | Meaning | Next action |
+| --- | --- | --- |
+| `complete` | lock, branch, and worktree all removed | resolve `missing_fields` and re-run `gitea_bootstrap_author_issue_worktree` |
+| `partial` | rollback ran; some artifacts survive, by design or because a step errored | scoped to exactly what survives — see below |
+| `failed` | rollback never completed, so nothing is proven removed | `gitea_inspect_issue_lock_contract` (read-only) before anything else |
+
+Within `partial`, the surviving set decides the action:
+
+| Survives | Next action |
+| --- | --- |
+| lock + branch + worktree | `gitea_recover_incomplete_bootstrap_lock` for that exact issue, branch, and worktree |
+| branch + worktree (lock released) | `gitea_lock_issue` — no implementation bytes were written, so the worktree is still base-equivalent |
+| lock only (worktree removed) | `gitea_inspect_issue_lock_contract`; the surviving lock must be released by its recorded owner before bootstrap is retried |
+| branch only | `gitea_inspect_issue_lock_contract`, then re-run bootstrap, which adopts the existing branch |
+
+`failed_rollback_steps` names any rollback step that errored, and the returned
+action says so rather than presenting the surviving state as intentional.
+
+> The lock half of that rollback was dead code until #953 review 632 F2:
+> `run_compensating_recovery` called `issue_lock_store.release_session_lock`,
+> which did not exist, inside a bare `except Exception: pass`. Every rollback
+> removed the branch and worktree and silently left the lock — the exact
+> uninspectable, unrecoverable state this issue exists to eliminate. The
+> function now exists, releases only a lock whose recorded `owner_session`
+> matches, and its failures are recorded rather than swallowed.
+
+## The contract
+
+A canonical lock carries every field in
+`author_lock_contract.REQUIRED_LOCK_FIELDS`:
+
+| Field | Meaning |
+| --- | --- |
+| `remote`, `org`, `repo`, `issue_number` | repository and issue identity |
+| `branch_name`, `worktree_path` | the binding this claim owns |
+| `work_lease` | the canonical lease block, below |
+| `lock_provenance` | sanctioned source, minted server-side |
+| `lock_generation` | monotonic; every write advances it |
+
+`work_lease` carries every field in
+`author_lock_contract.REQUIRED_WORK_LEASE_FIELDS`, notably:
+
+| Field | Meaning |
+| --- | --- |
+| `claimant.{username,profile}` | **canonical** claimant placement |
+| `expires_at` | sliding TTL from `lease_policy` |
+| `last_heartbeat_at`, `heartbeat_count` | liveness evidence |
+| `task_session_id` | the ownership fencing token — never null |
+| `lifecycle_version` | `heartbeat-v1`; its absence is what makes a lock legacy |
+
+### Claimant placement and legacy compatibility
+
+`work_lease.claimant` is canonical. A top-level `claimant` is the legacy
+placement written by pre-#953 bootstrap and is still **read** — through the one
+shared reader, `issue_lock_store.lock_claimant` — so an existing lock is not
+refused for "not recording a claimant" when it plainly records one.
+
+Tolerating the placement is not a widening. Every caller still compares the
+values against server-resolved identity and profile, so a legacy placement
+grants nothing the canonical placement would not. When both are present, the
+`work_lease` copy wins: after an upgrade, a stale top-level copy must never
+decide ownership.
+
+### Expiration is explicit
+
+A lock with no recorded expiry is **not** "not yet expired". `is_lease_expired`
+returns `False` for it, which used to make such a lock permanently non-expiring
+*and* permanently ineligible for #760 exact-owner renewal, which only ever
+assesses an expired lease. `author_lock_contract.expiration_state` names the
+real fact: `recorded`, `missing`, or `unparseable`. A `missing` expiry makes the
+lock eligible for the recovery path below rather than stranding it.
+
+## Recovering an existing incomplete bootstrap lock
+
+For locks already written by the old bootstrap — including those whose branches
+already carry legitimate committed and pushed work — use:
+
+```text
+gitea_inspect_issue_lock_contract(issue_number, branch_name, worktree_path, remote=...)
+gitea_recover_incomplete_bootstrap_lock(issue_number, branch_name, worktree_path, expected_head, remote=...)
+```
+
+`gitea_inspect_issue_lock_contract` is strictly read-only: it performs no lock,
+lease, branch, worktree, issue, or pull-request mutation. Use it first to see
+which fields are missing and what the recommended action is; pass `dry_run=True`
+to the recovery tool to preview the decision without writing.
+
+`gitea_recover_incomplete_bootstrap_lock` upgrades that one lock to the
+canonical contract. Before writing anything it verifies:
+
+* repository (`remote`, `org`, `repo`) and issue number
+* claimant username **and** profile against the server-resolved values — a
+  matching username alone is never accepted
+* branch, worktree path, worktree existence, and worktree registration
+* the worktree is on the recorded branch
+* the observed head equals the caller's `expected_head`
+* the existing lock's generation and provenance state
+* the absence of healthy foreign ownership
+
+What it deliberately does **not** do:
+
+* it never moves, resets, or rewinds the branch, and never requires
+  base-equivalence — preserving the committed work is the entire point;
+* it never pushes and never creates a pull request;
+* it touches only the single lock file for that exact remote/org/repo/issue;
+* it accepts no caller-supplied provenance and no caller-supplied authorization
+  flag — both are minted server-side.
+
+A recovered lock records a `bootstrap_lock_recovery` block holding both sides of
+the transition — prior contract, prior missing fields, prior generation, prior
+owning session, the replacement `task_session_id`, and the preserved head — so a
+recovered claim never reads as an original one.
+
+### Gates, in order
+
+`gitea_recover_incomplete_bootstrap_lock` is an author-only durable-lock
+mutation and carries the same three gates as every comparable author operation,
+in this order:
+
+1. `role_session_router.check_author_mutation_after_reviewer_stop` — no author
+   fallback after a reviewer `wrong_role_stop`.
+2. `_namespace_mutation_block(task, remote=remote, author_role_exclusive=True)` —
+   the namespace wall. It refuses a reviewer-bound session and, because this
+   task's required permission is `gitea.issue.comment` (which merger,
+   controller, and reconciler profiles also hold), additionally requires the
+   active profile's derived role kind to be exactly `author`. A refusal carries
+   `namespace_block: true` and emits a `BLOCKED` audit record naming the
+   namespace and profile.
+3. `_profile_permission_block` — operation, provenance, and session-context
+   gates.
+
+Exact-owner claimant matching inside `assess_bootstrap_lock_recovery` runs
+*after* all three. It is a further layer, never a substitute for them: on its
+own it refuses one step too late and leaves the audit trail silent about the
+attempt.
+
+### Refusals
+
+| `refusal_code` | Meaning |
+| --- | --- |
+| `no_durable_lock` | nothing to recover |
+| `already_canonical` | lock is fine; rewriting would invalidate a live heartbeat token |
+| `foreign_claimant` | recorded claimant is not the active identity/profile pair |
+| `healthy_foreign_lock` | a live foreign-owned lock; takeover is not a recovery path |
+| `identity_unresolved` | identity or profile could not be resolved |
+| `binding_mismatch` | repository, issue, branch, or worktree does not match |
+| `worktree_invalid` | worktree missing, unregistered, or on another branch |
+| `head_mismatch` | the worktree moved under the caller |
+
+## The #447 create-PR provenance guard is unchanged
+
+`issue_lock_provenance.assess_lock_file_for_create_pr` still requires both a
+sanctioned `lock_provenance` and a `work_lease`, and the sanctioned source set
+was **not** widened. Bootstrap writes through
+`issue_lock_provenance.SOURCE_LOCK_ISSUE` — the lock it produces *is* a
+canonical lock, not a second dialect with its own exemption. Bootstrap now
+satisfies the guard rather than the guard being relaxed to admit bootstrap.

docs/mcp-config-drift-runbook.md

@@ -0,0 +1,70 @@
+# MCP Config Drift Diagnostic & Sanctioned Repair Runbook (#672)
+
+This document describes the diagnostic framework for detecting configuration drift between the active IDE MCP configuration (`~/.gemini/antigravity-ide/mcp_config.json`) and the offline/global canonical configuration (`~/.gemini/config/mcp_config.json`), and establishes the **sanctioned repair runbook**.
+
+## Background & Problem Statement
+
+Offline tools like `test_mcp_conn.py` test the global configuration (`~/.gemini/config/mcp_config.json`) via `subprocess.Popen`. However, the active IDE/client namespace uses `~/.gemini/antigravity-ide/mcp_config.json`. When required Gitea role servers (`gitea-author`, `gitea-reviewer`, `gitea-merger`, `gitea-reconciler`, `gitea-controller`, `gitea-tools`) are missing or carry mismatched profile environments in the active IDE config:
+
+1. Offline tests pass (`test_mcp_conn.py` green).
+2. The IDE client returns `EOF` / `transport closed` when attempting role-scoped mutations.
+3. Operators misdiagnose missing server definitions as stale runtimes, leading to forbidden `pkill` attempts (#630) or `mtime` hacks (#655).
+
+## Diagnostic Tool: `mcp_config_drift.py`
+
+Run the diagnostic tool directly to compare configurations:
+
+```bash
+python3 mcp_config_drift.py --json
+```
+
+Or specify custom config locations:
+
+```bash
+python3 mcp_config_drift.py \
+  --active-config ~/.gemini/antigravity-ide/mcp_config.json \
+  --global-config ~/.gemini/config/mcp_config.json
+```
+
+### Key Diagnostic Outputs
+
+- `in_sync`: Boolean indicating if all required Gitea role servers exist in the active IDE config with matching profile declarations.
+- `missing_role_servers`: List of role servers present in global config but missing from active IDE config.
+- `profile_mismatches`: List of profile environment mismatches per server.
+- `reasons`: Explicit, human-readable list of drift causes.
+
+All returned payloads automatically redact secret tokens, DSNs, Authorization headers, and private keys.
+
+---
+
+## Sanctioned Repair Path (Step-by-Step)
+
+When `mcp_config_drift.py` reports drift (`in_sync: false`), execute the following **sanctioned repair steps**:
+
+1. **Backup Active IDE Config:**
+   ```bash
+   cp ~/.gemini/antigravity-ide/mcp_config.json ~/.gemini/antigravity-ide/mcp_config.json.bak
+   ```
+2. **Patch Active IDE Config:**
+   Copy the missing Gitea role server JSON blocks (`gitea-author`, `gitea-reviewer`, etc.) from `~/.gemini/config/mcp_config.json` into `~/.gemini/antigravity-ide/mcp_config.json`.
+3. **Reconnect via IDE/Client:**
+   Use the IDE / client UI reconnection control (or restart the IDE client app).
+4. **Verify Active Namespace Health:**
+   Invoke `gitea_whoami` (and optional `gitea_resolve_task_capability`) through the active IDE client on each required role namespace.
+
+---
+
+## FORBIDDEN Repair Actions (#630 / #655)
+
+The following actions are **strictly forbidden** for config drift repair:
+
+- ❌ **`pkill` or manual daemon process kill commands:** Process kills cause contamination and break active session leases.
+- ❌ **`mtime` touch edits:** Artificial mtime modifications mask stale runtimes without updating configuration.
+- ❌ **Source code edits:** Mutating python tool logic to bypass missing server entries.
+- ❌ **Session-state edits:** Direct database or lock-file state mutation.
+
+---
+
+## Final Report Guidelines
+
+A workflow final report **must not** rely on offline `test_mcp_conn.py` output alone. Final reports must include active-config evidence from live `gitea_whoami` calls on the active IDE namespaces.

docs/mcp-namespace-eof-recovery.md

@@ -47,18 +47,33 @@ Do the steps in order. Stop as soon as a live **client-namespace** call succeeds
    - Only the Gitea namespace fails → single-namespace transport close. Continue.
    - Every server fails → restart the whole MCP client, not just one namespace.
 
-2. **Reconnect the namespace through the client, not the shell.** Use the IDE /
+2. **Request the sanctioned reconnect surface (#678), then reconnect through
-   client MCP-reconnect action for that server entry (in Claude Code:
+   the client — not the shell.** From a still-reachable Gitea MCP namespace
-   `/mcp` → reconnect the affected `gitea-*` server). Reconnecting forces the
+   (or after host auto-reconnect), call:
-   client to spawn a fresh subprocess and re-open the pipe. This clears the
+
-   closed-client state that a bare `kill`/respawn from a terminal does **not**.
+   ```text
+   gitea_request_mcp_reconnect(
+       namespace="gitea-author",   # or gitea-reviewer / gitea-merger / …
+       reason="transport_eof",
+       client="codex",            # or claude_code / generic
+   )
+   ```
+
+   The tool is **report-only**: it never restarts a process. It returns
+   namespace, profile, pid/session, startup SHA, current master SHA, boundary
+   status, and a **typed blocker** with exact operator UI steps for Codex
+   (Reload Developer Tools / per-server reconnect) or Claude Code (`/mcp`).
+   Then perform the host reconnect those steps describe so the client spawns a
+   fresh subprocess and re-opens the pipe. That clears the closed-client state
+   that a bare `kill`/respawn from a terminal does **not**.
 
 3. **Do not "fix" it by importing the server or poking the process.** Reaching
    for `python -c 'import gitea_mcp_server ...'`, raw JSON-RPC from a shell,
    killing PIDs to force a respawn, or touching MCP config mtimes does **not**
    restore the *client's* view of the namespace and violates the daemon-import
    guard (#558, `docs/mcp-daemon-import-guard.md`). The only sanctioned repair
-   is a **client reconnect / relaunch**.
+   is a **client reconnect / relaunch** (or the typed operator path returned by
+   `gitea_request_mcp_reconnect`).
 
 4. **Verify through the same path the workflow will use.** After reconnect, call
    the specific tool the blocked workflow needs — not just any tool — through

docs/mcp-restart-path-inventory.md

@@ -45,10 +45,11 @@ and *fails closed*.
 | `legacy_auto_restart_helper` | removed | A helper (`_trigger_mcp_auto_restart`) that actively restarted the server from the read-only resolver path. | Removed in #685; kept absent by `assert_auto_restart_helper_absent()`. | #685, #657 |
 | `config_touch_reload` | removed | Touching (utime) the MCP client config to make the host reload the server. | Removed from the resolver in #685: stale detection is report-only, never mutating config, spawning threads, or calling `os._exit`. | #685, #657 |
 | `master_advance_auto_restart` | guarded_fail_closed | On-disk master advancing past the running code. | `master_parity_gate` captures startup parity and blocks mutations while stale, emitting restart guidance; the process never self-restarts. | #420, #591, #657 |
-| `stale_runtime_resolver_reconnect` | guarded_fail_closed | The capability resolver detecting a stale serving process. | Report-only (#685): returns `restart_required`/`stop_required` and an exact reconnect action; no restart, thread, config touch, or `os._exit`. | #685, #657 |
+| `stale_runtime_resolver_reconnect` | guarded_fail_closed | The capability resolver detecting a stale serving process. | Report-only (#685): returns `restart_required`/`stop_required` and an exact reconnect action; no restart, thread, config touch, or `os._exit`. | #685, #657, #678 |
+| `codex_client_reconnect_request` | guarded_fail_closed | `gitea_request_mcp_reconnect` report-only tool for Codex/LLM sessions. | Report-only (#678): returns namespace/profile/pid/startup SHA/master SHA/boundary status and a typed operator blocker with exact client UI steps; never restarts or kills. | #678, #630, #685, #657 |
 | `manual_daemon_kill` | forbidden | Shell kills of the daemon: `pkill -f mcp_server.py`, `killall`, broad `pkill -f python` sweeps, or `kill <pid>` of a daemon pid. | Forbidden (#630): `runtime_recovery_guard` classifies these as contamination and `gitea_record_daemon_process_kill_attempt` writes a durable marker that fails later mutations closed. Operator maintenance authorization is read only from the environment. | #630, #657 |
 | `conflict_marker_infra_stop` | guarded_fail_closed | The daemon entrypoint scans for unresolved merge-conflict markers at startup and stops (`sys.exit(1)`). | Fail-closed startup stop, not a restart: the process exits and waits for the operator to resolve conflicts and relaunch; never loops. | #657 |
-| `ide_client_reconnect` | host_residual | A manual `/mcp reconnect` (or equivalent host action) that recreates the MCP client connection. | Outside this process's control; the sanctioned recovery the gates point operators toward. No in-process code initiates it. | #584, #656, #657 |
+| `ide_client_reconnect` | host_residual | A manual `/mcp reconnect` (or equivalent host action) that recreates the MCP client connection. Agents obtain exact UI steps via `gitea_request_mcp_reconnect` (#678). | Outside this process's control; the sanctioned recovery the gates point operators toward. No in-process code initiates it. | #584, #656, #657, #678 |
 | `profile_switch_runtime` | sanctioned_narrow_recovery | Switching the active execution profile at runtime (dynamic-profile mode). | In-process and restart-free: `runtime_switching_supported` is true, so a switch rebinds capability without recreating the process. | #656, #657 |
 
 ## Guards enforced in CI

docs/mcp-tool-inventory.md

@@ -103,6 +103,7 @@ that gates each call, not which tools exist.
 - `gitea_get_shell_health`
 - `gitea_heartbeat_issue_lock`
 - `gitea_heartbeat_reviewer_pr_lease`
+- `gitea_inspect_issue_lock_contract`
 - `gitea_inspect_workflow_lease`
 - `gitea_issue_irrecoverable_provenance_authorization`
 - `gitea_list_dependency_edges`
@@ -134,9 +135,11 @@ that gates each call, not which tools exist.
 - `gitea_record_pre_review_command`
 - `gitea_record_shell_spawn_outcome`
 - `gitea_record_stable_branch_push_attempt`
+- `gitea_recover_incomplete_bootstrap_lock`
 - `gitea_release_merger_pr_lease`
 - `gitea_release_reviewer_pr_lease`
 - `gitea_release_workflow_lease`
+- `gitea_request_mcp_reconnect`
 - `gitea_request_mcp_restart`
 - `gitea_resolve_task_capability`
 - `gitea_resume_review_draft`

docs/observability/webui-observability-console.md

@@ -0,0 +1,35 @@
+# Web Console: Sentry/GlitchTip Observability & Incident Bridge Console (#649)
+
+This document describes the Phase 4 observability console surface integrated into the MCP Control Plane Web Console (`webui/`), backed by the #612 incident bridge and the #613 control-plane DB substrate.
+
+## Architectural Authority Model (ADR Alignment)
+
+Per the Web Console Architecture ADR (`docs/architecture/webui-control-plane-console-architecture-adr.md`):
+
+| Layer | Responsibility | Authority |
+|---|---|---|
+| **Gitea** | Durable work record | Issues, PRs, comments, reviews, labels, merges |
+| **Control-plane DB** | Live coordination & linkage | `incident_links` table, session leases, allocations |
+| **Sentry / GlitchTip** | Observability input | Unresolved incidents, error events, stack traces |
+| **Incident Bridge (#612)** | Reconciliation engine | Reconciles provider observations into Gitea issues |
+| **Web Console (`webui/`)** | Read-only projection & gated actions | Projects connection health & correlation links; gates writes |
+
+> **Key Rule:** Raw monitoring incidents are **never** assignable control-plane `work_items`. They remain observation input only.
+
+## Redaction Boundary Invariants
+
+1. **No secrets in returns or rendering:** Auth tokens (`SENTRY_AUTH_TOKEN`, `GLITCHTIP_AUTH_TOKEN`), DSNs, `Authorization` headers, and sensitive local file paths are passed through `webui.console_redaction` before leaving the server.
+2. **Safe projection:** Connection objects report `credentials_present: true/false` rather than exposing raw keys or headers.
+
+## Console Endpoints
+
+- **HTML Surface:** `GET /observability` — Renders provider connection cards, error correlation tables, and gated reconcile controls.
+- **Versioned API:** `GET /api/v1/observability` — Returns structured JSON snapshot with `schema_version`, `providers`, `links`, and `metrics`.
+- **Legacy Compatibility Alias:** `GET /api/observability` — Read-only compatibility alias for Phase 4.
+
+## Gated Actions
+
+- `observability_reconcile_incident` (`gitea_observability_reconcile_incident`): Triggers or previews dry-run issue reconciliation for a provider incident.
+- `observability_link_issue` (`gitea_observability_link_issue`): Links a provider incident to an existing Gitea tracking issue.
+
+Both actions require `operator` role and gate through `task_capability_map`. Execution fails closed in read-only MVP mode.

docs/webui-authz-audit.md

@@ -98,6 +98,8 @@ already define, and a regression test asserts each mapping matches.
 | `system.rebind_session_worktree` | operator | gated_write | `gitea.read` | Yes | No | No | 2 |
 | `system.reconcile_cleanups` | controller | privileged | `gitea.pr.close` | Yes | No | No | 2 |
 | `initiate_workflow` | operator | gated_write | `gitea.read` | Yes | No | No | 2 |
+| `observability_reconcile_incident` | operator | gated_write | `gitea.read` | Yes | No | No | 4 |
+| `observability_link_issue` | operator | gated_write | `gitea.read` | Yes | No | No | 4 |
 
 **Dual control** means the acting principal may not be the sole authority: a
 second distinct principal must confirm. **Break-glass** means the action is

issue_lock_renewal.py

@@ -436,6 +436,117 @@ def owning_pr_renewal_evidence(
     }
 
 
+def owning_pr_renewal_from_lock(
+    lock_record: Mapping[str, Any] | None,
+) -> dict[str, Any] | None:
+    """Rebuild owning-PR renewal evidence from a persisted lock (#945).
+
+    The renewal mirror of ``issue_lock_recovery.recovered_owning_pr_from_lock``.
+    ``owning_pr_renewal_evidence`` supplies the waiver for the duration of the
+    ``gitea_lock_issue`` call only. The commit, push, create-PR, and
+    duplicate-assessment gates run later in their own calls and re-derive
+    ownership from the durable lock instead — so without this the open PR that
+    renewal already proved belongs to this author reappears there as competing
+    duplicate work, and the exact owner is refused with
+    ``duplicate_commit_prevented`` despite complete matching evidence.
+
+    This reads only the ``lease_renewal`` block that the server itself writes,
+    on a lock the caller must already own. Like the recovery mirror it is a
+    re-read of server-derived state, never a fresh assertion: a caller able to
+    forge it could equally forge the lock file every other ownership gate
+    already treats as authoritative.
+
+    Renewal has no descendant case — the assessor required the local, remote and
+    PR heads to be equal — so that equality is re-checked here, and the record
+    must still name the claimant the lock records.
+
+    **What the claimant check below is, and what it is not.** It compares
+    ``lease_renewal.identity``/``profile`` against the claimant recorded on the
+    *same* lock file. Both sides are server-written fields of one document, so
+    this is an internal-consistency check: it rejects a lock whose renewal block
+    and claimant disagree. It does **not** consult the live authenticated caller
+    and therefore does not, on its own, prove that the session invoking a later
+    gate is the session the renewal was granted to.
+
+    The binding that actually keeps one session from using another's renewal is
+    structural, and it lives in the caller rather than here. The enforcement
+    paths load the lock through ``_load_existing_issue_lock()`` with no issue
+    coordinates, which resolves ``issue_lock_store.read_session_issue_lock()``
+    → the session pointer at ``session-{os.getpid()}.json``. Lock *selection* is
+    scoped to the operating-system process, so a caller cannot aim the recheck
+    at a lock some other process bound. Its limits follow from what that scope
+    is: it is per-process, not per-authenticated-user; it says nothing about a
+    lock reached by explicit issue coordinates rather than the session pointer,
+    and nothing about two roles sharing one process. Live identity and profile
+    are enforced separately, by the mutation-authority and profile gates each
+    mutating path already runs — not by this rebuild.
+
+    This function is therefore strictly a re-read with an added consistency
+    requirement. It narrows what a persisted lock can authorize; it never widens
+    it, and it never substitutes for a caller-identity gate.
+    """
+    if not isinstance(lock_record, Mapping):
+        return None
+    record = lock_record.get("lease_renewal")
+    if not isinstance(record, Mapping) or not record.get("renewed"):
+        return None
+
+    branch_name = _text(record.get("branch_name")) or _text(
+        lock_record.get("branch_name")
+    )
+    pr_head = _text(record.get("pr_head_sha"))
+    local_head = _text(record.get("head_sha"))
+    remote_head = _text(record.get("remote_head_sha"))
+    raw_pr_number = record.get("pr_number")
+    raw_issue_number = lock_record.get("issue_number")
+
+    if raw_pr_number is None or raw_issue_number is None:
+        return None
+    if not branch_name or not pr_head:
+        return None
+    # The assessor required all three heads to agree before it granted renewal.
+    # Re-check, so a truncated, drifted, or hand-built record cannot widen the
+    # exemption past the single head the renewal disposition actually proved.
+    if not local_head or not remote_head:
+        return None
+    if pr_head != local_head or pr_head != remote_head:
+        return None
+    # Renewal is refused outright unless the durable lock records both a
+    # claimant username and profile, so a sanctioned record always carries them.
+    # Requiring them to still agree rejects a lock whose renewal block and
+    # claimant disagree. Both values are read from this one server-written
+    # document: this is internal consistency, not a check against the live
+    # authenticated caller — see the docstring for the binding that is.
+    claimant = lock_record.get("claimant")
+    if not isinstance(claimant, Mapping):
+        lease = lock_record.get("work_lease")
+        claimant = lease.get("claimant") if isinstance(lease, Mapping) else None
+    if not isinstance(claimant, Mapping):
+        return None
+    identity = _text(record.get("identity"))
+    profile = _text(record.get("profile"))
+    if not identity or identity != _text(claimant.get("username")):
+        return None
+    if not profile or profile != _text(claimant.get("profile")):
+        return None
+
+    try:
+        pr_number = int(raw_pr_number)
+        issue_number = int(raw_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,
     *,

issue_lock_store.py

@@ -299,9 +299,13 @@ def _ownership_refusals(
             f"lock worktree '{lock.get('worktree_path')}' does not match "
             f"'{worktree_path}'"
         )
-    lease = lock.get("work_lease") if isinstance(lock, dict) else None
+    # #953 AC2/AC13/AC14: read through the shared claimant reader so a lock
-    claimant = lease.get("claimant") if isinstance(lease, dict) else None
+    # written by bootstrap — which records the claimant at the top level — is
-    claimant = claimant if isinstance(claimant, dict) else {}
+    # not refused for "not recording a claimant" when it plainly records one.
+    # This is not a widening: the values are still compared against the
+    # server-resolved identity and profile immediately below, so a legacy
+    # placement grants nothing that the canonical placement would not.
+    claimant = lock_claimant(lock) if isinstance(lock, dict) else {}
     recorded_identity = str(claimant.get("username") or "").strip()
     recorded_profile = str(claimant.get("profile") or "").strip()
     if not recorded_identity or not recorded_profile:
@@ -636,6 +640,99 @@ def iter_lock_files(lock_dir: str | None = None) -> list[str]:
     return sorted(paths)
 
 
+def release_session_lock(
+    *,
+    issue_number: int,
+    session: str,
+    lock_dir: str | None = None,
+    remote: str | None = None,
+    org: str | None = None,
+    repo: str | None = None,
+) -> str:
+    """Remove exactly the durable lock *session* created for *issue_number*.
+
+    ``author_issue_bootstrap.run_compensating_recovery`` has called this name
+    since #850, but it was never defined: the call raised ``AttributeError``
+    into a bare ``except Exception: pass``, so the lock half of every
+    compensating rollback silently did nothing. The branch and worktree were
+    removed and the lock was left behind — a state no sanctioned tool can act
+    on, since recovery refuses ``worktree_invalid`` and ``gitea_lock_issue`` has
+    no worktree to bind (#953 review 632 F2).
+
+    Ownership is proven, not asserted. A record is removed only when its
+    recorded ``owner_session`` equals *session* and its issue number matches;
+    ``remote``/``org``/``repo`` narrow it further when supplied. Zero matches or
+    more than one both raise, so a caller can never delete a lock it does not
+    own and an ambiguous directory is never guessed at. The ``.json.lock`` flock
+    sidecar is deliberately left in place — it is a zero-byte mutex another
+    process may hold, and removing it under contention would be a race.
+
+    Returns the removed lock file path.
+    """
+    target_issue = int(issue_number)
+    owner = str(session or "").strip()
+    if not owner:
+        raise ValueError(
+            "release_session_lock requires the owning session id (fail closed)"
+        )
+
+    def _is_owned_durable_lock(record: dict[str, Any] | None) -> bool:
+        # A durable lock, not a bootstrap phase journal or a session pointer,
+        # both of which can share a directory and carry the same issue number
+        # and owner_session.
+        if not record or "lock_generation" not in record:
+            return False
+        if not str(record.get("branch_name") or "").strip():
+            return False
+        if not str(record.get("worktree_path") or "").strip():
+            return False
+        try:
+            if int(record.get("issue_number") or 0) != target_issue:
+                return False
+        except (TypeError, ValueError):
+            return False
+        return str(record.get("owner_session") or "").strip() == owner
+
+    # Prefer the exact keyed path when the caller knows the repository; scanning
+    # is the fallback for callers that only carry the issue number.
+    if remote and org and repo:
+        exact = lock_file_path(
+            remote=remote,
+            org=org,
+            repo=repo,
+            issue_number=target_issue,
+            lock_dir=lock_dir,
+        )
+        if not _is_owned_durable_lock(read_lock_file(exact)):
+            raise FileNotFoundError(
+                f"durable issue lock '{exact}' is absent or is not owned by "
+                f"session '{owner}' (fail closed; nothing released)"
+            )
+        os.remove(exact)
+        return exact
+
+    matches: list[str] = []
+    for path in iter_lock_files(lock_dir):
+        if _is_owned_durable_lock(read_lock_file(path)):
+            matches.append(path)
+
+    if not matches:
+        raise FileNotFoundError(
+            f"no durable issue lock for issue #{target_issue} is owned by "
+            f"session '{owner}' (fail closed; nothing released)"
+        )
+    if len(matches) > 1:
+        raise RuntimeError(
+            f"{len(matches)} durable locks for issue #{target_issue} claim "
+            f"session '{owner}'; refusing to guess which to release "
+            "(fail closed)"
+        )
+
+    path = matches[0]
+    os.remove(path)
+    return path
+
+
 def find_lock_for_branch(
     *,
     remote: str,
@@ -1112,21 +1209,43 @@ def assess_same_issue_lease_conflict(
     )
 
 
-def _lock_claimant(lock: dict[str, Any] | None) -> dict[str, str]:
+def lock_claimant(lock: dict[str, Any] | None) -> dict[str, str]:
+    """Read the claimant from either canonical or legacy placement (#953 AC13/AC14).
+
+    ``work_lease.claimant`` is the canonical placement and is preferred; a
+    top-level ``claimant`` is the legacy/bootstrap placement and is accepted as
+    a fallback. This is the single definition. Before #953 the readers
+    disagreed: this module, ``issue_lock_renewal``, and ``issue_lock_recovery``
+    tolerated both placements, while ``_ownership_refusals`` looked only in
+    ``work_lease`` — which is what made a bootstrap-written lock
+    un-heartbeatable.
+
+    Preferring ``work_lease`` over the top level is deliberate: once a legacy
+    lock is upgraded, the canonical placement is authoritative and a stale
+    top-level copy must never win.
+
+    This decides *where to look*, never whether ownership is proven — every
+    caller still compares these values against server-resolved identity and
+    profile.
+    """
     if not isinstance(lock, dict):
         return {}
-    claimant = lock.get("claimant")
-    if not isinstance(claimant, dict):
     lease = lock.get("work_lease")
     claimant = lease.get("claimant") if isinstance(lease, dict) else None
+    if not isinstance(claimant, dict):
+        claimant = lock.get("claimant")
     if not isinstance(claimant, dict):
         return {}
     return {
-        "username": str(claimant.get("username") or ""),
+        "username": str(claimant.get("username") or "").strip(),
-        "profile": str(claimant.get("profile") or ""),
+        "profile": str(claimant.get("profile") or "").strip(),
     }
 
 
+#: Back-compatible alias for the pre-#953 private name.
+_lock_claimant = lock_claimant
+
+
 def assess_foreign_lock_overwrite(
     existing_lock: dict[str, Any] | None,
     incoming_lock: dict[str, Any],

mcp_client_reconnect.py

@@ -0,0 +1,328 @@
+"""Sanctioned MCP client reconnect request surface for Codex/LLM sessions (#678).
+
+Codex and other agent hosts can detect stale or closed Gitea MCP runtimes, but
+the host owns the transport. This module never restarts, kills, or reloads a
+daemon. It builds:
+
+1. A **callable reconnect request** result agents can invoke via
+   ``gitea_request_mcp_reconnect`` (report-only, side-effect free).
+2. A **typed blocker** with exact operator UI steps when recovery must be
+   performed by the host/operator.
+
+Forbidden recovery paths (must never be recommended):
+
+* ``pkill`` / ``kill`` / ``killall`` of MCP daemons
+* ``touch`` / mtime config reload hacks
+* ``.env`` or MCP config edits as recovery
+* session-state file edits
+* raw Gitea API / direct server-import fallbacks
+
+After the operator reconnects, workflows restart from identity / runtime /
+capability preflight (``gitea_whoami`` → ``gitea_resolve_task_capability`` →
+task).
+"""
+
+from __future__ import annotations
+
+from typing import Any, Mapping
+
+# --- Reason vocabulary -------------------------------------------------------
+
+REASON_STALE_RUNTIME = "stale-runtime"
+REASON_TRANSPORT_EOF = "transport_eof"
+REASON_MISSING_NAMESPACE = "missing_namespace"
+REASON_NOT_REQUIRED = "not_required"
+REASON_UNSPECIFIED = "unspecified"
+
+VALID_REASONS = frozenset(
+    {
+        REASON_STALE_RUNTIME,
+        REASON_TRANSPORT_EOF,
+        REASON_MISSING_NAMESPACE,
+        REASON_NOT_REQUIRED,
+        REASON_UNSPECIFIED,
+    }
+)
+
+# Boundary statuses reported to callers (match review_workflow_boundary style).
+BOUNDARY_CLEAN = "clean"
+BOUNDARY_MISMATCH = "mismatch"
+BOUNDARY_STALE = "stale"
+BOUNDARY_UNKNOWN = "unknown"
+
+# Typed blocker kinds
+BLOCKER_OPERATOR_RECONNECT = "operator_mcp_reconnect_required"
+BLOCKER_NONE = "none"
+
+FORBIDDEN_RECOVERY_PATHS: tuple[str, ...] = (
+    "pkill / kill / killall of mcp_server.py, gitea_mcp_server, or broad python sweeps",
+    "touch / mtime-based MCP config reload hacks",
+    ".env edits as recovery",
+    "MCP config file edits as recovery",
+    "session-state file edits as recovery",
+    "raw Gitea API or direct MCP server-import fallbacks",
+)
+
+# Client-specific operator UI steps. Keep Codex first (issue title surface).
+OPERATOR_UI_STEPS: dict[str, tuple[str, ...]] = {
+    "codex": (
+        "In Codex, open the MCP / Developer tools panel for this workspace.",
+        "Locate the named Gitea MCP server entry (namespace) that needs reconnect "
+        "(e.g. gitea-author, gitea-reviewer, gitea-merger, gitea-tools, "
+        "gitea-controller, gitea-reconciler).",
+        "Click 'Reload Developer Tools' or the server reconnect/reload control "
+        "for that entry so the client spawns a fresh MCP subprocess.",
+        "If per-server reconnect is unavailable, fully restart the Codex client "
+        "(quit and relaunch) so all MCP namespaces reattach.",
+        "After reconnect, rerun the blocked workflow from preflight: "
+        "gitea_whoami → gitea_resolve_task_capability → the original task. "
+        "Do not resume mid-mutation.",
+    ),
+    "claude_code": (
+        "Run `/mcp` (or open the MCP servers UI) in Claude Code.",
+        "Reconnect the affected gitea-* server entry so the client reopens stdio.",
+        "If reconnect fails, relaunch the Claude Code session entirely.",
+        "After reconnect, restart the workflow from gitea_whoami → "
+        "gitea_resolve_task_capability → task.",
+    ),
+    "generic": (
+        "Use the host/IDE MCP reconnect or reload control for the named namespace.",
+        "If no per-namespace control exists, restart the MCP client/editor.",
+        "After reconnect, restart the workflow from identity/capability preflight.",
+    ),
+}
+
+DEFAULT_CLIENT = "codex"
+
+
+def normalize_reason(reason: str | None) -> str:
+    """Map free-form reason strings onto the closed vocabulary."""
+    raw = (reason or "").strip().lower()
+    if not raw:
+        return REASON_UNSPECIFIED
+    if raw in VALID_REASONS:
+        return raw
+    text = raw.replace(" ", "_").replace("-", "_")
+    aliases = {
+        "stale_runtime": REASON_STALE_RUNTIME,
+        "staleruntime": REASON_STALE_RUNTIME,
+        "runtime_stale": REASON_STALE_RUNTIME,
+        "stale": REASON_STALE_RUNTIME,
+        "transport_eof": REASON_TRANSPORT_EOF,
+        "transport_closed": REASON_TRANSPORT_EOF,
+        "eof": REASON_TRANSPORT_EOF,
+        "client_is_closing": REASON_TRANSPORT_EOF,
+        "missing_namespace": REASON_MISSING_NAMESPACE,
+        "namespace_missing": REASON_MISSING_NAMESPACE,
+        "not_required": REASON_NOT_REQUIRED,
+        "healthy": REASON_NOT_REQUIRED,
+        "ok": REASON_NOT_REQUIRED,
+        "unspecified": REASON_UNSPECIFIED,
+    }
+    if text in aliases:
+        return aliases[text]
+    hyphenated = text.replace("_", "-")
+    if hyphenated in VALID_REASONS:
+        return hyphenated
+    return REASON_UNSPECIFIED
+
+
+def normalize_client(client: str | None) -> str:
+    """Return a known client key for operator UI steps."""
+    text = (client or "").strip().lower().replace(" ", "_").replace("-", "_")
+    if text in ("codex", "openai_codex", "openai"):
+        return "codex"
+    if text in ("claude", "claude_code", "claude_desktop", "anthropic"):
+        return "claude_code"
+    if text in OPERATOR_UI_STEPS:
+        return text
+    return DEFAULT_CLIENT
+
+
+def classify_boundary_status(
+    *,
+    startup_sha: str | None,
+    current_master_sha: str | None,
+    live_stale: bool | None = None,
+    in_parity: bool | None = None,
+) -> str:
+    """Derive boundary_status from parity evidence."""
+    if live_stale is True or in_parity is False:
+        return BOUNDARY_STALE
+    start = (startup_sha or "").strip().lower()
+    current = (current_master_sha or "").strip().lower()
+    if start and current and start != current:
+        return BOUNDARY_MISMATCH
+    if start and current and start == current:
+        return BOUNDARY_CLEAN
+    if in_parity is True:
+        return BOUNDARY_CLEAN
+    return BOUNDARY_UNKNOWN
+
+
+def operator_ui_steps(client: str | None, *, namespace: str | None = None) -> list[str]:
+    """Exact operator UI steps for the named client."""
+    key = normalize_client(client)
+    steps = list(OPERATOR_UI_STEPS.get(key) or OPERATOR_UI_STEPS[DEFAULT_CLIENT])
+    ns = (namespace or "").strip()
+    if ns:
+        steps = [
+            s.replace("named Gitea MCP server entry (namespace)", f"namespace '{ns}'")
+            .replace("affected gitea-* server entry", f"server entry '{ns}'")
+            .replace("named namespace", f"namespace '{ns}'")
+            for s in steps
+        ]
+    return steps
+
+
+def build_reconnect_request(
+    *,
+    namespace: str,
+    profile: str | None = None,
+    pid: int | str | None = None,
+    session_id: str | None = None,
+    startup_sha: str | None = None,
+    current_master_sha: str | None = None,
+    boundary_status: str | None = None,
+    reason: str | None = None,
+    client: str | None = DEFAULT_CLIENT,
+    live_stale: bool | None = None,
+    in_parity: bool | None = None,
+    restart_required: bool | None = None,
+    stop_required: bool | None = None,
+    extra: Mapping[str, Any] | None = None,
+) -> dict[str, Any]:
+    """Build the structured reconnect-request / typed-blocker payload (#678).
+
+    Never mutates process, config, or session state. Always side-effect free.
+    """
+    ns = (namespace or "").strip() or "unknown"
+    normalized_reason = normalize_reason(reason)
+    boundary = (boundary_status or "").strip() or classify_boundary_status(
+        startup_sha=startup_sha,
+        current_master_sha=current_master_sha,
+        live_stale=live_stale,
+        in_parity=in_parity,
+    )
+
+    reconnect_needed = True
+    if normalized_reason == REASON_NOT_REQUIRED and boundary == BOUNDARY_CLEAN:
+        reconnect_needed = False
+    if restart_required is False and stop_required is False and boundary == BOUNDARY_CLEAN:
+        # Explicit healthy probe
+        if normalized_reason in (REASON_NOT_REQUIRED, REASON_UNSPECIFIED):
+            reconnect_needed = False
+            normalized_reason = REASON_NOT_REQUIRED
+
+    if restart_required is True or stop_required is True:
+        reconnect_needed = True
+        if normalized_reason in (REASON_NOT_REQUIRED, REASON_UNSPECIFIED):
+            normalized_reason = REASON_STALE_RUNTIME
+
+    client_key = normalize_client(client)
+    steps = operator_ui_steps(client_key, namespace=ns)
+
+    result: dict[str, Any] = {
+        "success": True,
+        "read_only": True,
+        "reconnect_performed": False,
+        "mutation_performed": False,
+        "reconnect_needed": reconnect_needed,
+        "namespace": ns,
+        "profile": (profile or "").strip() or None,
+        "pid": pid,
+        "session_id": (session_id or "").strip() or None,
+        "startup_sha": (startup_sha or "").strip() or None,
+        "current_master_sha": (current_master_sha or "").strip() or None,
+        "boundary_status": boundary,
+        "reason": normalized_reason,
+        "client": client_key,
+        "forbidden_recovery_paths": list(FORBIDDEN_RECOVERY_PATHS),
+        "post_reconnect_preflight": [
+            "gitea_whoami",
+            "gitea_resolve_task_capability",
+            "original_task",
+        ],
+        "exact_safe_next_action": None,
+        "blocker_kind": BLOCKER_NONE,
+        "operator_ui_steps": steps,
+        "typed_blocker": None,
+    }
+
+    if reconnect_needed:
+        result["blocker_kind"] = BLOCKER_OPERATOR_RECONNECT
+        result["stop_required"] = True
+        result["restart_required"] = True
+        result["exact_safe_next_action"] = (
+            f"blocker_kind={BLOCKER_OPERATOR_RECONNECT}: operator must reconnect "
+            f"MCP namespace '{ns}' via the host UI (client={client_key}). "
+            "Do not pkill, touch configs, edit session state, or use raw API. "
+            "After reconnect, restart from gitea_whoami → "
+            "gitea_resolve_task_capability → task."
+        )
+        result["typed_blocker"] = {
+            "blocker_kind": BLOCKER_OPERATOR_RECONNECT,
+            "namespaces": [ns],
+            "why_reconnect_required": normalized_reason,
+            "operator_ui_steps": steps,
+            "client": client_key,
+            "forbidden_recovery_paths": list(FORBIDDEN_RECOVERY_PATHS),
+            "instruction_after_reconnect": (
+                "Rerun the blocked workflow from preflight "
+                "(gitea_whoami → gitea_resolve_task_capability → task). "
+                "Do not continue mid-mutation from pre-reconnect state."
+            ),
+        }
+    else:
+        result["stop_required"] = False
+        result["restart_required"] = False
+        result["exact_safe_next_action"] = (
+            f"Reconnect not required for namespace '{ns}' "
+            f"(boundary_status={boundary}). Proceed with the original task."
+        )
+
+    if extra:
+        for key, value in extra.items():
+            if key not in result:
+                result[key] = value
+
+    return result
+
+
+def reasons_never_suggest_forbidden(text: str) -> bool:
+    """Return True when *text* does not recommend a forbidden recovery path.
+
+    Mentions that *ban* a path (e.g. ``Do not pkill`` / ``never edit session
+    state``) are allowed. Positive recommendations such as ``use pkill`` or
+    ``run killall`` fail.
+    """
+    import re
+
+    lowered = (text or "").lower()
+    # Strip common ban prefixes so "do not pkill" does not trip positive checks.
+    scrubbed = re.sub(
+        r"\b(?:do not|don't|never|must not|forbid(?:den)?|ban(?:ned)?)\b"
+        r"[^.!;\n]{0,80}",
+        " ",
+        lowered,
+    )
+    # Positive imperative / advisory forms that would tell an agent to do harm.
+    positive_suggestions = (
+        "use pkill",
+        "run pkill",
+        "try pkill",
+        "pkill -f",
+        "use killall",
+        "run killall",
+        "killall mcp",
+        "use kill ",
+        "run kill ",
+        "touch the mcp",
+        "touch mcp config",
+        "utime(",
+        "edit the mcp config to recover",
+        "edit .env to recover",
+        "import gitea_mcp_server",
+        "python -c 'import gitea_mcp",
+    )
+    return not any(frag in scrubbed for frag in positive_suggestions)

mcp_config_drift.py

@@ -0,0 +1,237 @@
+"""Antigravity IDE vs Global MCP Config Drift Diagnostic (#672).
+
+Diagnoses config drift between the active IDE MCP configuration
+(e.g. ``~/.gemini/antigravity-ide/mcp_config.json``) and the offline/global
+canonical configuration (e.g. ``~/.gemini/config/mcp_config.json``).
+
+Hard rules (#672 / #630 / #655):
+* Distinguish offline/global success from active IDE namespace availability.
+* Never print tokens, DSNs, Authorization headers, or secret-bearing env vars.
+* Sanctioned repair path is: backup active config -> patch active config from canonical
+  -> reconnect through IDE/client -> verify with live ``gitea_whoami``.
+* FORBIDDEN: ``pkill``, mtime edits, source edits, or session-state edits for repair.
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import sys
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any
+
+from webui import console_redaction
+
+DEFAULT_ACTIVE_IDE_CONFIG = "~/.gemini/antigravity-ide/mcp_config.json"
+DEFAULT_GLOBAL_CONFIG = "~/.gemini/config/mcp_config.json"
+
+REQUIRED_GITEA_ROLE_SERVERS = (
+    "gitea-author",
+    "gitea-reviewer",
+    "gitea-merger",
+    "gitea-reconciler",
+    "gitea-controller",
+    "gitea-tools",
+)
+
+SANCTIONED_REPAIR_RUNBOOK: tuple[str, ...] = (
+    "1. Backup active IDE config: cp ~/.gemini/antigravity-ide/mcp_config.json ~/.gemini/antigravity-ide/mcp_config.json.bak",
+    "2. Patch active IDE config: copy required missing Gitea role server entries from global config (~/.gemini/config/mcp_config.json) into active IDE config.",
+    "3. Reconnect via IDE/client UI or client restart (do NOT use host process kill).",
+    "4. Verify active namespace health using live gitea_whoami and gitea_resolve_task_capability on each role namespace.",
+    "FORBIDDEN REPAIR PATHS: pkill / host process kill, mtime touch edits, source code edits, or session-state edits.",
+)
+
+
+def resolve_config_path(path_str: str) -> Path:
+    """Expand user and resolve absolute path."""
+    return Path(os.path.expanduser(path_str)).resolve()
+
+
+def load_mcp_config(config_path: str | Path) -> tuple[dict[str, Any] | None, str | None]:
+    """Load and parse JSON MCP configuration from file.
+
+    Returns (config_dict, error_message).
+    """
+    resolved = resolve_config_path(str(config_path))
+    if not resolved.exists():
+        return None, f"file_not_found: {resolved}"
+    try:
+        with open(resolved, "r", encoding="utf-8") as f:
+            data = json.load(f)
+        if not isinstance(data, dict):
+            return None, f"invalid_schema: root is not a JSON object in {resolved}"
+        return data, None
+    except Exception as exc:
+        return None, f"unreadable_json: {exc} in {resolved}"
+
+
+def extract_mcp_servers(config: dict[str, Any] | None) -> dict[str, dict[str, Any]]:
+    """Extract the mcpServers or mcp_servers mapping safely."""
+    if not config:
+        return {}
+    servers = config.get("mcpServers") or config.get("mcp_servers") or {}
+    if isinstance(servers, dict):
+        return {str(k): v for k, v in servers.items() if isinstance(v, dict)}
+    return {}
+
+
+def _safe_redact_server_config(srv_cfg: dict[str, Any]) -> dict[str, Any]:
+    """Redact secrets from environment variables and command line args."""
+    safe = {}
+    if "command" in srv_cfg:
+        safe["command"] = str(srv_cfg["command"])
+    if "args" in srv_cfg and isinstance(srv_cfg["args"], list):
+        safe["args"] = [console_redaction.redact_text(str(a)) for a in srv_cfg["args"]]
+    if "env" in srv_cfg and isinstance(srv_cfg["env"], dict):
+        safe_env = {}
+        for k, v in srv_cfg["env"].items():
+            if any(secret_kw in k.lower() for secret_kw in ("token", "secret", "pass", "key", "auth")):
+                safe_env[k] = "[REDACTED]"
+            else:
+                safe_env[k] = console_redaction.redact_text(str(v))
+        safe["env"] = safe_env
+    return safe
+
+
+def analyze_config_drift(
+    active_config_path: str = DEFAULT_ACTIVE_IDE_CONFIG,
+    global_config_path: str = DEFAULT_GLOBAL_CONFIG,
+) -> dict[str, Any]:
+    """Analyze MCP configuration drift between active IDE config and global config.
+
+    Returns structured diagnostic output.
+    """
+    active_resolved = resolve_config_path(active_config_path)
+    global_resolved = resolve_config_path(global_config_path)
+
+    active_cfg, active_err = load_mcp_config(active_resolved)
+    global_cfg, global_err = load_mcp_config(global_resolved)
+
+    active_servers = extract_mcp_servers(active_cfg)
+    global_servers = extract_mcp_servers(global_cfg)
+
+    missing_role_servers: list[str] = []
+    present_role_servers: list[str] = []
+    profile_mismatches: list[dict[str, Any]] = []
+    reasons: list[str] = []
+
+    if active_err:
+        reasons.append(f"Active IDE config error: {active_err}")
+    if global_err:
+        reasons.append(f"Global canonical config error: {global_err}")
+
+    # Check Gitea role servers
+    for srv_name in REQUIRED_GITEA_ROLE_SERVERS:
+        in_active = srv_name in active_servers
+        in_global = srv_name in global_servers
+
+        if in_active:
+            present_role_servers.append(srv_name)
+        elif in_global:
+            missing_role_servers.append(srv_name)
+            reasons.append(
+                f"Missing Gitea role server '{srv_name}' in active IDE config ({active_resolved})"
+            )
+
+        if in_active and in_global:
+            # Compare profiles & environments
+            act_env = active_servers[srv_name].get("env", {}) if isinstance(active_servers[srv_name], dict) else {}
+            glo_env = global_servers[srv_name].get("env", {}) if isinstance(global_servers[srv_name], dict) else {}
+
+            act_prof = act_env.get("GITEA_MCP_PROFILE") or act_env.get("GITEA_PROFILE_NAME")
+            glo_prof = glo_env.get("GITEA_MCP_PROFILE") or glo_env.get("GITEA_PROFILE_NAME")
+
+            if act_prof != glo_prof:
+                mismatch_item = {
+                    "server": srv_name,
+                    "active_profile": act_prof,
+                    "global_profile": glo_prof,
+                }
+                profile_mismatches.append(mismatch_item)
+                reasons.append(
+                    f"Profile mismatch for '{srv_name}': active='{act_prof}' != global='{glo_prof}'"
+                )
+
+    in_sync = bool(
+        not active_err
+        and not global_err
+        and not missing_role_servers
+        and not profile_mismatches
+    )
+
+    report = {
+        "timestamp": datetime.now(timezone.utc).isoformat(),
+        "in_sync": in_sync,
+        "active_config_path": str(active_resolved),
+        "active_config_exists": active_cfg is not None,
+        "global_config_path": str(global_resolved),
+        "global_config_exists": global_cfg is not None,
+        "required_role_servers": list(REQUIRED_GITEA_ROLE_SERVERS),
+        "present_role_servers": present_role_servers,
+        "missing_role_servers": missing_role_servers,
+        "profile_mismatches": profile_mismatches,
+        "reasons": reasons,
+        "sanctioned_repair_runbook": list(SANCTIONED_REPAIR_RUNBOOK),
+        "forbidden_repair_methods": [
+            "pkill / host process kill",
+            "mtime touch edits",
+            "source code edits",
+            "session-state edits",
+        ],
+    }
+
+    return console_redaction.redact_payload(report)
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(
+        description="Diagnose Gitea MCP role server config drift between active IDE and global config."
+    )
+    parser.add_argument(
+        "--active-config",
+        default=DEFAULT_ACTIVE_IDE_CONFIG,
+        help="Path to active IDE MCP config JSON",
+    )
+    parser.add_argument(
+        "--global-config",
+        default=DEFAULT_GLOBAL_CONFIG,
+        help="Path to global/canonical MCP config JSON",
+    )
+    parser.add_argument(
+        "--json", action="store_true", help="Print raw JSON report"
+    )
+
+    args = parser.parse_args()
+
+    report = analyze_config_drift(args.active_config, args.global_config)
+
+    if args.json:
+        print(json.dumps(report, indent=2))
+    else:
+        print("=== MCP Config Drift Diagnostic Report ===")
+        print(f"Timestamp: {report['timestamp']}")
+        print(f"In Sync: {report['in_sync']}")
+        print(f"Active IDE Config: {report['active_config_path']} (exists={report['active_config_exists']})")
+        print(f"Global Config: {report['global_config_path']} (exists={report['global_config_exists']})")
+        print(f"Present Role Servers: {', '.join(report['present_role_servers']) if report['present_role_servers'] else 'None'}")
+        print(f"Missing Role Servers: {', '.join(report['missing_role_servers']) if report['missing_role_servers'] else 'None'}")
+        if report['profile_mismatches']:
+            print("Profile Mismatches:")
+            for m in report['profile_mismatches']:
+                print(f"  - {m['server']}: active={m['active_profile']} vs global={m['global_profile']}")
+        if report['reasons']:
+            print("Drift Reasons:")
+            for r in report['reasons']:
+                print(f"  - {r}")
+        print("\nSanctioned Repair Runbook:")
+        for step in report['sanctioned_repair_runbook']:
+            print(f"  {step}")
+
+    sys.exit(0 if report["in_sync"] else 1)
+
+
+if __name__ == "__main__":
+    main()

mcp_restart_paths.py

@@ -225,8 +225,33 @@ _RESTART_PATHS: tuple[RestartPath, ...] = (
             "exact_safe_next_action pointing at IDE/client reconnect; performs "
             "no restart, thread spawn, config touch, or os._exit."
         ),
-        locations=("gitea_mcp_server.py (gitea_resolve_task_capability)",),
+        locations=(
-        references=("#685", "#657"),
+            "gitea_mcp_server.py (gitea_resolve_task_capability)",
+            "gitea_mcp_server.py (gitea_request_mcp_reconnect)",
+            "mcp_client_reconnect.py",
+        ),
+        references=("#685", "#657", "#678"),
+    ),
+    RestartPath(
+        path_id="codex_client_reconnect_request",
+        title="Sanctioned Codex/LLM reconnect request tool",
+        mechanism=(
+            "gitea_request_mcp_reconnect: agents invoke a report-only tool that "
+            "returns namespace/profile/pid/startup SHA/master SHA/boundary "
+            "status plus a typed operator blocker with exact client UI steps."
+        ),
+        classification=CLASS_GUARDED_FAIL_CLOSED,
+        guard=(
+            "Report-only (#678): never restarts, kills, reloads, or edits "
+            "config; recovery is always host/operator reconnect. Forbidden "
+            "paths (pkill, touch, .env/config/session-state hacks) are listed "
+            "and never recommended."
+        ),
+        locations=(
+            "mcp_client_reconnect.py",
+            "gitea_mcp_server.py (gitea_request_mcp_reconnect)",
+        ),
+        references=("#678", "#630", "#685", "#657"),
     ),
     RestartPath(
         path_id="manual_daemon_kill",
@@ -271,7 +296,8 @@ _RESTART_PATHS: tuple[RestartPath, ...] = (
         title="Host/IDE MCP reconnect",
         mechanism=(
             "A manual `/mcp reconnect` (or equivalent host action) that the "
-            "IDE performs to recreate the MCP client connection."
+            "IDE performs to recreate the MCP client connection. Agents obtain "
+            "exact UI steps via gitea_request_mcp_reconnect (#678)."
         ),
         classification=CLASS_HOST_RESIDUAL,
         guard=(
@@ -279,8 +305,12 @@ _RESTART_PATHS: tuple[RestartPath, ...] = (
             "gates point operators toward; documented as residual host "
             "behavior. No in-process code initiates it."
         ),
-        locations=("host/IDE",),
+        locations=(
-        references=("#584", "#656", "#657"),
+            "host/IDE",
+            "mcp_client_reconnect.py",
+            "gitea_mcp_server.py (gitea_request_mcp_reconnect)",
+        ),
+        references=("#584", "#656", "#657", "#678"),
         residual_host=True,
     ),
     RestartPath(

pytest.ini

@@ -0,0 +1,3 @@
+[pytest]
+testpaths = tests
+norecursedirs = branches .git venv __pycache__ graphify-out

role_namespace_gate.py

@@ -74,6 +74,43 @@ def check_author_mutation_namespace(
     return True, []
 
 
+def check_author_role_kind(
+    mutation_task: str,
+    profile: dict,
+) -> tuple[bool, list[str]]:
+    """Author-exclusive wall for durable-lock mutations (#953 F1).
+
+    ``check_author_mutation_namespace`` walls off reviewer-bound sessions, which
+    is the whole gate for tasks whose required permission is itself author-only
+    (``gitea.pr.create``, ``gitea.repo.commit``). It is *not* sufficient for a
+    task gated on ``gitea.issue.comment``, which every configured role holds: a
+    merger, controller, or reconciler session would clear both the namespace
+    check and the permission gate and still reach the durable write.
+
+    Opt-in per call site and additive. It refuses any active profile whose
+    derived role kind is not exactly ``author`` for a task the router declares
+    author-required, and grants nothing to anyone — a ``mixed`` profile is
+    refused rather than admitted.
+    """
+    required_role = role_session_router.required_role_for_task(mutation_task)
+    if required_role != "author":
+        return True, []
+
+    allowed = profile.get("allowed_operations") or []
+    forbidden = profile.get("forbidden_operations") or []
+    active_role = derive_role_kind(allowed, forbidden)
+    if active_role == "author":
+        return True, []
+
+    profile_name = profile.get("profile_name") or ""
+    namespace = infer_mcp_namespace(profile_name)
+    return False, [
+        f"author mutation '{mutation_task}' blocked: active session role kind is "
+        f"'{active_role}', not 'author' ({profile_name} / {namespace}); this "
+        "operation writes a durable author issue lock and is author-exclusive",
+    ]
+
+
 def mutation_audit_context(mutation_task: str, profile: dict, *,
                            remote=None, repository=None) -> dict:
     """Structured mutation metadata for audit records (#209)."""

role_session_router.py

@@ -75,6 +75,10 @@ AUTHOR_TASKS = frozenset({
     "push_branch",
     "bootstrap_author_issue_worktree",
     "gitea_bootstrap_author_issue_worktree",
+    # #953: recovery of an incomplete bootstrap lock is an author-only durable
+    # state mutation and belongs to the same class as bootstrap itself.
+    "recover_incomplete_bootstrap_lock",
+    "gitea_recover_incomplete_bootstrap_lock",
     "create_pr",
     "comment_pr",
     "address_pr_change_requests",
@@ -112,6 +116,12 @@ TASK_REQUIRED_ROLE = {
     "claim_issue": "author",
     "create_branch": "author",
     "push_branch": "author",
+    # #953: without this entry ``required_role_for_task`` returns None and
+    # ``role_namespace_gate.check_author_mutation_namespace`` short-circuits to
+    # "allowed" — the namespace wall on the recovery tool would be inert. The
+    # capability map already records the same role; both tables must agree.
+    "recover_incomplete_bootstrap_lock": "author",
+    "gitea_recover_incomplete_bootstrap_lock": "author",
     "create_pr": "author",
     "comment_pr": "author",
     "address_pr_change_requests": "author",

task_capability_map.py

@@ -41,6 +41,27 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = {
         "permission": "gitea.issue.comment",
         "role": "author",
     },
+    # #953: target-specific upgrade of an incomplete bootstrap lock (explicit
+    # operation, never a widening of lock_issue). Author-only, and the tool
+    # additionally proves exact-owner claimant match before writing.
+    "recover_incomplete_bootstrap_lock": {
+        "permission": "gitea.issue.comment",
+        "role": "author",
+    },
+    "gitea_recover_incomplete_bootstrap_lock": {
+        "permission": "gitea.issue.comment",
+        "role": "author",
+    },
+    # #953: read-only lock contract inspection. Read permission only — it must
+    # never be able to mutate.
+    "inspect_issue_lock_contract": {
+        "permission": "gitea.read",
+        "role": "author",
+    },
+    "gitea_inspect_issue_lock_contract": {
+        "permission": "gitea.read",
+        "role": "author",
+    },
     # #860: dirty orphaned same-claimant worktree recovery (explicit operation).
     "recover_dirty_orphaned_issue_worktree": {
         "permission": "gitea.issue.comment",

tests/test_issue_678_mcp_client_reconnect.py

@@ -0,0 +1,323 @@
+"""Tests for sanctioned Codex MCP reconnect request surface (#678)."""
+
+from __future__ import annotations
+
+import os
+import unittest
+from unittest import mock
+
+import mcp_client_reconnect as mcr
+
+
+class NormalizeReasonTests(unittest.TestCase):
+    def test_stale_runtime_aliases(self):
+        self.assertEqual(mcr.normalize_reason("stale-runtime"), mcr.REASON_STALE_RUNTIME)
+        self.assertEqual(mcr.normalize_reason("stale_runtime"), mcr.REASON_STALE_RUNTIME)
+        self.assertEqual(mcr.normalize_reason("STALE"), mcr.REASON_STALE_RUNTIME)
+
+    def test_transport_eof_aliases(self):
+        self.assertEqual(mcr.normalize_reason("transport_eof"), mcr.REASON_TRANSPORT_EOF)
+        self.assertEqual(mcr.normalize_reason("EOF"), mcr.REASON_TRANSPORT_EOF)
+        self.assertEqual(
+            mcr.normalize_reason("client_is_closing"), mcr.REASON_TRANSPORT_EOF
+        )
+
+    def test_missing_namespace(self):
+        self.assertEqual(
+            mcr.normalize_reason("missing_namespace"), mcr.REASON_MISSING_NAMESPACE
+        )
+
+    def test_empty_is_unspecified(self):
+        self.assertEqual(mcr.normalize_reason(None), mcr.REASON_UNSPECIFIED)
+        self.assertEqual(mcr.normalize_reason(""), mcr.REASON_UNSPECIFIED)
+
+
+class BoundaryClassificationTests(unittest.TestCase):
+    def test_clean_when_shas_match(self):
+        self.assertEqual(
+            mcr.classify_boundary_status(
+                startup_sha="abc", current_master_sha="abc"
+            ),
+            mcr.BOUNDARY_CLEAN,
+        )
+
+    def test_mismatch_when_shas_differ(self):
+        self.assertEqual(
+            mcr.classify_boundary_status(
+                startup_sha="aaa", current_master_sha="bbb"
+            ),
+            mcr.BOUNDARY_MISMATCH,
+        )
+
+    def test_stale_when_live_stale(self):
+        self.assertEqual(
+            mcr.classify_boundary_status(
+                startup_sha="aaa",
+                current_master_sha="aaa",
+                live_stale=True,
+            ),
+            mcr.BOUNDARY_STALE,
+        )
+
+
+class BuildReconnectRequestTests(unittest.TestCase):
+    def test_stale_runtime_returns_typed_blocker_with_codex_steps(self):
+        result = mcr.build_reconnect_request(
+            namespace="gitea-author",
+            profile="prgs-author",
+            pid=1234,
+            session_id="sess-1",
+            startup_sha="aaa111",
+            current_master_sha="bbb222",
+            reason="stale-runtime",
+            client="codex",
+            restart_required=True,
+            stop_required=True,
+        )
+        self.assertTrue(result["success"])
+        self.assertTrue(result["read_only"])
+        self.assertFalse(result["reconnect_performed"])
+        self.assertFalse(result["mutation_performed"])
+        self.assertTrue(result["reconnect_needed"])
+        self.assertEqual(result["namespace"], "gitea-author")
+        self.assertEqual(result["profile"], "prgs-author")
+        self.assertEqual(result["pid"], 1234)
+        self.assertEqual(result["session_id"], "sess-1")
+        self.assertEqual(result["startup_sha"], "aaa111")
+        self.assertEqual(result["current_master_sha"], "bbb222")
+        self.assertEqual(result["boundary_status"], mcr.BOUNDARY_MISMATCH)
+        self.assertEqual(result["blocker_kind"], mcr.BLOCKER_OPERATOR_RECONNECT)
+        self.assertIsNotNone(result["typed_blocker"])
+        blocker = result["typed_blocker"]
+        self.assertEqual(blocker["namespaces"], ["gitea-author"])
+        self.assertEqual(blocker["why_reconnect_required"], mcr.REASON_STALE_RUNTIME)
+        self.assertTrue(any("Codex" in s or "Reload" in s for s in blocker["operator_ui_steps"]))
+        self.assertIn("pkill", " ".join(result["forbidden_recovery_paths"]).lower())
+        self.assertTrue(
+            mcr.reasons_never_suggest_forbidden(result["exact_safe_next_action"] or "")
+        )
+        # Must not recommend forbidden recovery.
+        for step in blocker["operator_ui_steps"]:
+            self.assertTrue(mcr.reasons_never_suggest_forbidden(step), step)
+
+    def test_transport_eof_typed_blocker(self):
+        result = mcr.build_reconnect_request(
+            namespace="gitea-reviewer",
+            reason="transport_eof",
+            client="claude_code",
+        )
+        self.assertTrue(result["reconnect_needed"])
+        self.assertEqual(result["reason"], mcr.REASON_TRANSPORT_EOF)
+        self.assertEqual(result["client"], "claude_code")
+        steps = " ".join(result["operator_ui_steps"]).lower()
+        self.assertIn("/mcp", steps)
+
+    def test_missing_namespace_typed_blocker(self):
+        result = mcr.build_reconnect_request(
+            namespace="gitea-merger",
+            reason="missing_namespace",
+            client="codex",
+        )
+        self.assertTrue(result["reconnect_needed"])
+        self.assertEqual(result["reason"], mcr.REASON_MISSING_NAMESPACE)
+        self.assertEqual(
+            result["typed_blocker"]["blocker_kind"], mcr.BLOCKER_OPERATOR_RECONNECT
+        )
+
+    def test_healthy_not_required(self):
+        result = mcr.build_reconnect_request(
+            namespace="gitea-tools",
+            startup_sha="deadbeef",
+            current_master_sha="deadbeef",
+            reason="not_required",
+            client="codex",
+            in_parity=True,
+            restart_required=False,
+            stop_required=False,
+        )
+        self.assertFalse(result["reconnect_needed"])
+        self.assertEqual(result["blocker_kind"], mcr.BLOCKER_NONE)
+        self.assertIsNone(result["typed_blocker"])
+        self.assertFalse(result["stop_required"])
+        self.assertFalse(result["restart_required"])
+        self.assertIn("not required", (result["exact_safe_next_action"] or "").lower())
+
+    def test_successful_reconnect_report_fields_present(self):
+        """AC2: reconnect result reports required fields (even when needed)."""
+        result = mcr.build_reconnect_request(
+            namespace="gitea-controller",
+            profile="prgs-controller",
+            pid=99,
+            session_id="sid",
+            startup_sha="s" * 40,
+            current_master_sha="c" * 40,
+            reason="stale-runtime",
+        )
+        for key in (
+            "namespace",
+            "profile",
+            "pid",
+            "session_id",
+            "startup_sha",
+            "current_master_sha",
+            "boundary_status",
+        ):
+            self.assertIn(key, result)
+            self.assertIsNotNone(result[key], key)
+
+
+class ToolSurfaceTests(unittest.TestCase):
+    """Exercise gitea_request_mcp_reconnect with a stubbed server context."""
+
+    def test_tool_is_registered_and_side_effect_free(self):
+        import gitea_mcp_server as srv
+
+        self.assertTrue(hasattr(srv, "gitea_request_mcp_reconnect"))
+        with mock.patch.object(srv, "_profile_operation_gate", return_value=None):
+            with mock.patch.object(
+                srv,
+                "get_profile",
+                return_value={
+                    "profile_name": "prgs-author",
+                    "role_kind": "author",
+                    "role": "author",
+                },
+            ):
+                with mock.patch.object(
+                    srv,
+                    "_current_master_parity",
+                    return_value={
+                        "startup_head": "a" * 40,
+                        "current_head": "a" * 40,
+                        "daemon_start_head": "a" * 40,
+                        "local_head": "a" * 40,
+                        "in_parity": True,
+                        "stale": False,
+                        "restart_required": False,
+                        "determinable": True,
+                        "live_stale": False,
+                        "live_known": True,
+                        "reasons": [],
+                    },
+                ):
+                    with mock.patch.object(
+                        srv.master_parity_gate,
+                        "format_parity",
+                        return_value="in parity",
+                    ):
+                        with mock.patch.object(
+                            srv.role_namespace_gate,
+                            "infer_mcp_namespace",
+                            return_value="gitea-author",
+                        ):
+                            with mock.patch.object(
+                                srv.session_ctx,
+                                "mutation_context_audit_fields",
+                                return_value={"session_profile": "prgs-author"},
+                            ):
+                                result = srv.gitea_request_mcp_reconnect(
+                                    namespace="gitea-author",
+                                    reason="not_required",
+                                    client="codex",
+                                    remote="prgs",
+                                )
+        self.assertTrue(result.get("success"))
+        self.assertFalse(result.get("reconnect_performed"))
+        self.assertFalse(result.get("mutation_performed"))
+        self.assertEqual(result.get("namespace"), "gitea-author")
+        self.assertEqual(result.get("profile"), "prgs-author")
+        self.assertEqual(result.get("pid"), os.getpid())
+        self.assertIn("startup_sha", result)
+        self.assertIn("current_master_sha", result)
+        self.assertIn("boundary_status", result)
+        self.assertTrue(
+            mcr.reasons_never_suggest_forbidden(
+                result.get("exact_safe_next_action") or ""
+            )
+        )
+
+    def test_tool_stale_returns_typed_blocker(self):
+        import gitea_mcp_server as srv
+
+        with mock.patch.object(srv, "_profile_operation_gate", return_value=None):
+            with mock.patch.object(
+                srv,
+                "get_profile",
+                return_value={
+                    "profile_name": "prgs-reconciler",
+                    "role_kind": "reconciler",
+                    "role": "reconciler",
+                },
+            ):
+                with mock.patch.object(
+                    srv,
+                    "_current_master_parity",
+                    return_value={
+                        "startup_head": "a" * 40,
+                        "current_head": "b" * 40,
+                        "daemon_start_head": "a" * 40,
+                        "local_head": "b" * 40,
+                        "in_parity": False,
+                        "stale": True,
+                        "restart_required": True,
+                        "determinable": True,
+                        "live_stale": True,
+                        "live_known": True,
+                        "reasons": ["stale"],
+                    },
+                ):
+                    with mock.patch.object(
+                        srv.master_parity_gate,
+                        "format_parity",
+                        return_value="stale",
+                    ):
+                        with mock.patch.object(
+                            srv.role_namespace_gate,
+                            "infer_mcp_namespace",
+                            return_value="gitea-reconciler",
+                        ):
+                            with mock.patch.object(
+                                srv.session_ctx,
+                                "mutation_context_audit_fields",
+                                return_value={},
+                            ):
+                                result = srv.gitea_request_mcp_reconnect(
+                                    reason="stale-runtime",
+                                    client="codex",
+                                )
+        self.assertTrue(result["reconnect_needed"])
+        self.assertEqual(
+            result["blocker_kind"], mcr.BLOCKER_OPERATOR_RECONNECT
+        )
+        self.assertIsNotNone(result["typed_blocker"])
+        self.assertIn("gitea-reconciler", result["typed_blocker"]["namespaces"])
+        self.assertTrue(result["stop_required"])
+        self.assertTrue(result["restart_required"])
+        self.assertTrue(
+            mcr.reasons_never_suggest_forbidden(
+                result.get("exact_safe_next_action") or ""
+            )
+        )
+
+
+class InventoryRegistrationTests(unittest.TestCase):
+    def test_reconnect_path_in_restart_inventory(self):
+        import mcp_restart_paths as mrp
+
+        ids = {p.path_id for p in mrp.iter_restart_paths()}
+        self.assertIn("codex_client_reconnect_request", ids)
+        self.assertIn("ide_client_reconnect", ids)
+
+    def test_tool_name_in_documented_inventory(self):
+        import mcp_tool_inventory as inv
+
+        doc_path = os.path.join(
+            os.path.dirname(os.path.dirname(__file__)), inv.INVENTORY_DOC_PATH
+        )
+        with open(doc_path, encoding="utf-8") as handle:
+            documented = inv.parse_documented_inventory(handle.read())
+        self.assertIn("gitea_request_mcp_reconnect", documented)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_issue_943_runtime_context_helpers.py

@@ -0,0 +1,747 @@
+"""Regression: author bootstrap runtime authority and session ownership (#943).
+
+Two rounds of defects live here.
+
+**Round 1 (#943 as filed).** ``gitea_bootstrap_author_issue_worktree`` passed
+four values down to the bootstrap service that were never defined:
+``_active_username``, ``_active_profile_name``, ``_current_session_id`` and
+``_author_mutation_block``. Every call — dry-run included — raised
+``NameError`` while evaluating the arguments, before the service was entered.
+
+**Round 2 (review 622 on PR #944).** The first fix defined all four but made
+``_current_session_id`` mint ``<profile>-<pid>-<hex>`` once per process. The MCP
+daemon outlives every task it serves, so that value conflates sequential author
+tasks and can never equal the control-plane session that owns an
+allocator-created lease: ``_verify_assignment_and_lease_ids`` refused the whole
+allocated path with ``lease_session_mismatch``. The reviewed round also read the
+identity from the pinned session context while reading the profile from the live
+profile, so a rebind could produce a mixed claimant pair, and it swallowed every
+``get_profile()`` exception.
+
+These tests therefore drive real state, not mocks of internals: a temporary
+control-plane SQLite database and a temporary issue-lock directory, both
+redirected through the same environment variables production uses
+(``GITEA_CONTROL_PLANE_DB``, ``GITEA_ISSUE_LOCK_DIR``). The ownership gate that
+runs is the real one.
+
+``test_every_global_referenced_by_the_wrapper_resolves`` remains: it is what
+found ``_author_mutation_block``, and it generalises to the next missing
+reference. It supplements the runtime coverage below rather than standing in for
+it.
+"""
+
+from __future__ import annotations
+
+import ast
+import builtins
+import os
+import re
+import subprocess
+import tempfile
+import unittest
+from unittest import mock
+
+import author_issue_bootstrap as aib
+import control_plane_db
+import create_issue_bootstrap as cib
+import gitea_mcp_server as gms
+import issue_lock_store
+import workflow_scope_guard
+
+BOOTSTRAP_TASK = "bootstrap_author_issue_worktree"
+WRAPPER_NAME = "gitea_bootstrap_author_issue_worktree"
+RUNTIME_HELPERS = (
+    "_active_mutation_authority",
+    "_active_username",
+    "_active_profile_name",
+    "_resolve_owner_workflow_session",
+    "_author_mutation_block",
+)
+ORG = "Scaled-Tech-Consulting"
+REPO = "Gitea-Tools"
+IDENTITY = "jcwalker3"
+PROFILE = "prgs-author"
+
+# A per-task ownership key must carry no process identifier (#790).
+TASK_KEY_RE = re.compile(r"^author_issue_work-[0-9a-f]{16}$")
+
+
+def _make_control_repo(tmp: str) -> tuple[str, str]:
+    """Create a clean control checkout on master and return (path, head)."""
+    repo = os.path.join(tmp, "repo")
+    os.makedirs(os.path.join(repo, "branches"))
+    subprocess.check_call(
+        ["git", "init", "-b", "master", repo],
+        stdout=subprocess.DEVNULL,
+        stderr=subprocess.DEVNULL,
+    )
+    subprocess.check_call(
+        [
+            "git", "-C", repo,
+            "-c", "user.email=t@t", "-c", "user.name=t",
+            "commit", "--allow-empty", "-m", "init",
+        ],
+        stdout=subprocess.DEVNULL,
+        stderr=subprocess.DEVNULL,
+    )
+    head = subprocess.check_output(
+        ["git", "-C", repo, "rev-parse", "HEAD"], text=True
+    ).strip()
+    return repo, head
+
+
+def _wrapper_ast() -> ast.FunctionDef:
+    """Return the AST of the bootstrap wrapper as it exists on disk."""
+    path = os.path.join(
+        os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
+        "gitea_mcp_server.py",
+    )
+    with open(path, encoding="utf-8") as fh:
+        tree = ast.parse(fh.read())
+    for node in ast.walk(tree):
+        if isinstance(node, ast.FunctionDef) and node.name == WRAPPER_NAME:
+            return node
+    raise AssertionError(f"{WRAPPER_NAME} not found in gitea_mcp_server.py")
+
+
+class _IsolatedControlPlane(unittest.TestCase):
+    """Temp control-plane DB and temp issue-lock dir, via production env vars."""
+
+    def setUp(self):
+        self._tmp = tempfile.TemporaryDirectory()
+        self.addCleanup(self._tmp.cleanup)
+        self.tmp = self._tmp.name
+        self.db_path = os.path.join(self.tmp, "control-plane.sqlite3")
+        self.lock_dir = os.path.join(self.tmp, "issue-locks")
+        self.journals = os.path.join(self.tmp, "journals")
+        os.makedirs(self.lock_dir)
+        os.makedirs(self.journals)
+        env = mock.patch.dict(
+            os.environ,
+            {
+                control_plane_db.DB_PATH_ENV: self.db_path,
+                issue_lock_store.LOCK_DIR_ENV: self.lock_dir,
+            },
+        )
+        env.start()
+        self.addCleanup(env.stop)
+        self.db = control_plane_db.ControlPlaneDB(self.db_path)
+
+    def _allocate(self, session_id: str, *, issue: int = 943):
+        """Create a real assignment + lease owned by *session_id*."""
+        self.db.upsert_session(
+            session_id=session_id, role="author", profile=PROFILE, pid=os.getpid()
+        )
+        res = self.db.assign_and_lease(
+            session_id=session_id, role="author", remote="prgs",
+            org=ORG, repo=REPO, kind="issue", number=issue,
+        )
+        self.assertEqual(res.outcome, "assigned", res)
+        return res.assignment_id, res.lease_id
+
+    def _authority(self):
+        """A resolved authority pair, as the wrapper would compute it."""
+        return {"ok": True, "identity": IDENTITY, "profile_name": PROFILE}
+
+    def _resolve_session(self, **over):
+        kwargs = dict(
+            issue_number=943,
+            assignment_id=None,
+            lease_id=None,
+            session_id=None,
+            identity=IDENTITY,
+            profile_name=PROFILE,
+            remote="prgs",
+            org=ORG,
+            repo=REPO,
+        )
+        kwargs.update(over)
+        return gms._resolve_owner_workflow_session(**kwargs)
+
+
+class OwnershipGateTests(_IsolatedControlPlane):
+    """B2: the allocator-driven ownership path, against a real control plane."""
+
+    def setUp(self):
+        super().setUp()
+        self.repo, self.head = _make_control_repo(self.tmp)
+
+    def _bootstrap(self, **over):
+        kwargs = dict(
+            issue_number=943,
+            canonical_repo_root=self.repo,
+            expected_base_sha=self.head,
+            branch_name="fix/issue-943-runtime-context-helpers",
+            remote="prgs",
+            org=ORG,
+            repo=REPO,
+            active_identity=IDENTITY,
+            active_profile=PROFILE,
+            lock_dir=self.journals,
+            idempotency_key="test-943",
+            dry_run=True,
+        )
+        kwargs.update(over)
+        return aib.bootstrap_author_issue_worktree(**kwargs)
+
+    def test_true_owning_session_passes_the_ownership_gate(self):
+        """The canonical owner reaches and completes the service."""
+        session = "prgs-author-task-a"
+        assignment_id, lease_id = self._allocate(session)
+        res = self._bootstrap(
+            assignment_id=assignment_id, lease_id=lease_id, owner_session=session
+        )
+        self.assertTrue(res.get("success"), res)
+        self.assertTrue(res.get("dry_run"))
+        self.assertEqual(res.get("base_sha"), self.head)
+
+    def test_different_session_is_refused(self):
+        session = "prgs-author-task-a"
+        assignment_id, lease_id = self._allocate(session)
+        res = self._bootstrap(
+            assignment_id=assignment_id,
+            lease_id=lease_id,
+            owner_session="prgs-author-task-b",
+        )
+        self.assertFalse(res.get("success"))
+        self.assertEqual(res.get("reason_code"), "lease_session_mismatch")
+
+    def test_process_derived_session_would_be_refused(self):
+        """The reviewed round-1 value shape can never own an allocated lease."""
+        session = "prgs-author-task-a"
+        assignment_id, lease_id = self._allocate(session)
+        round_one_value = f"{PROFILE}-{os.getpid()}-deadbeef"
+        self.assertNotEqual(round_one_value, session)
+        res = self._bootstrap(
+            assignment_id=assignment_id,
+            lease_id=lease_id,
+            owner_session=round_one_value,
+        )
+        self.assertFalse(res.get("success"))
+        self.assertEqual(res.get("reason_code"), "lease_session_mismatch")
+
+    def test_unknown_lease_fails_closed(self):
+        session = "prgs-author-task-a"
+        assignment_id, _ = self._allocate(session)
+        res = self._bootstrap(
+            assignment_id=assignment_id,
+            lease_id="lease-does-not-exist",
+            owner_session=session,
+        )
+        self.assertFalse(res.get("success"))
+        self.assertEqual(res.get("reason_code"), "unknown_lease_id")
+
+    def test_released_lease_fails_closed(self):
+        session = "prgs-author-task-a"
+        assignment_id, lease_id = self._allocate(session)
+        self.db.release_lease(lease_id, session_id=session)
+        res = self._bootstrap(
+            assignment_id=assignment_id, lease_id=lease_id, owner_session=session
+        )
+        self.assertFalse(res.get("success"))
+        self.assertEqual(res.get("reason_code"), "lease_not_live")
+
+    def test_force_expired_lease_fails_closed(self):
+        session = "prgs-author-task-a"
+        assignment_id, lease_id = self._allocate(session)
+        self.db.force_expire_lease(lease_id, reason="test")
+        res = self._bootstrap(
+            assignment_id=assignment_id, lease_id=lease_id, owner_session=session
+        )
+        self.assertFalse(res.get("success"))
+        self.assertEqual(res.get("reason_code"), "lease_not_live")
+
+    def test_replacement_lease_does_not_inherit_prior_ownership(self):
+        """A second task's lease is not ownable by the first task's session."""
+        first = "prgs-author-task-a"
+        assignment_a, lease_a = self._allocate(first)
+        self.db.release_lease(lease_a, session_id=first)
+        second = "prgs-author-task-b"
+        assignment_b, lease_b = self._allocate(second)
+        self.assertNotEqual(lease_a, lease_b)
+        res = self._bootstrap(
+            assignment_id=assignment_b, lease_id=lease_b, owner_session=first
+        )
+        self.assertFalse(res.get("success"))
+        self.assertEqual(res.get("reason_code"), "lease_session_mismatch")
+
+    def test_assignment_lease_identifier_mismatch_fails_closed(self):
+        session = "prgs-author-task-a"
+        _, lease_id = self._allocate(session)
+        res = self._bootstrap(
+            assignment_id="asn-not-the-recorded-one",
+            lease_id=lease_id,
+            owner_session=session,
+        )
+        self.assertFalse(res.get("success"))
+        self.assertEqual(res.get("reason_code"), "assignment_lease_mismatch")
+
+    def test_lease_id_without_assignment_id_fails_closed(self):
+        session = "prgs-author-task-a"
+        _, lease_id = self._allocate(session)
+        res = self._bootstrap(lease_id=lease_id, owner_session=session)
+        self.assertFalse(res.get("success"))
+        self.assertEqual(res.get("reason_code"), "incomplete_assignment_lease_ids")
+
+    def test_dry_run_with_valid_allocator_bindings_leaves_no_durable_state(self):
+        session = "prgs-author-task-a"
+        assignment_id, lease_id = self._allocate(session)
+        res = self._bootstrap(
+            assignment_id=assignment_id, lease_id=lease_id, owner_session=session
+        )
+        self.assertTrue(res.get("success"), res)
+
+        branches = subprocess.check_output(
+            ["git", "-C", self.repo, "branch", "--list"], text=True
+        )
+        self.assertNotIn("issue-943", branches)
+        worktrees = subprocess.check_output(
+            ["git", "-C", self.repo, "worktree", "list"], text=True
+        )
+        self.assertNotIn("issue-943", worktrees)
+        self.assertFalse(
+            os.path.exists(
+                os.path.join(self.repo, "branches",
+                             "fix-issue-943-runtime-context-helpers")
+            )
+        )
+        journal = res.get("phase_journal") or {}
+        self.assertFalse(journal.get("completed"))
+        self.assertFalse(any((journal.get("artifacts_created") or {}).values()))
+        # The dry run must not have created an issue lock in the isolated dir.
+        self.assertEqual(os.listdir(self.lock_dir), [])
+
+    def test_apply_reaches_the_intended_transition_with_valid_bindings(self):
+        session = "prgs-author-task-a"
+        assignment_id, lease_id = self._allocate(session)
+        res = self._bootstrap(
+            assignment_id=assignment_id,
+            lease_id=lease_id,
+            owner_session=session,
+            dry_run=False,
+        )
+        self.assertTrue(res.get("success"), res)
+        self.assertNotEqual(res.get("dry_run"), True)
+        branches = subprocess.check_output(
+            ["git", "-C", self.repo, "branch", "--list"], text=True
+        )
+        self.assertIn("issue-943", branches)
+        self.assertTrue(os.path.isdir(res.get("worktree_path") or ""))
+
+
+class WorkflowSessionResolutionTests(_IsolatedControlPlane):
+    """B1: the wrapper resolves the owning session, never a process identifier."""
+
+    def test_declared_session_is_verified_against_the_control_plane(self):
+        session = "prgs-author-task-a"
+        assignment_id, lease_id = self._allocate(session)
+        res = self._resolve_session(
+            session_id=session, assignment_id=assignment_id, lease_id=lease_id
+        )
+        self.assertTrue(res.get("ok"), res)
+        self.assertEqual(res.get("session_id"), session)
+        self.assertEqual(res.get("session_source"), "declared")
+
+    def test_unknown_declared_session_is_refused_not_trusted(self):
+        res = self._resolve_session(session_id="prgs-author-not-a-session")
+        self.assertFalse(res.get("ok"))
+        self.assertEqual(res.get("reason_code"), "workflow_session_unverified")
+
+    def test_declared_session_for_another_role_is_refused(self):
+        self.db.upsert_session(
+            session_id="prgs-reviewer-x", role="reviewer", profile="prgs-reviewer",
+            pid=os.getpid(),
+        )
+        res = self._resolve_session(session_id="prgs-reviewer-x")
+        self.assertFalse(res.get("ok"))
+        self.assertEqual(res.get("reason_code"), "workflow_session_unverified")
+
+    def test_declared_session_for_another_profile_is_refused(self):
+        self.db.upsert_session(
+            session_id="other-profile-session", role="author",
+            profile="prgs-controller", pid=os.getpid(),
+        )
+        res = self._resolve_session(session_id="other-profile-session")
+        self.assertFalse(res.get("ok"))
+        self.assertEqual(res.get("reason_code"), "workflow_session_unverified")
+
+    def test_allocated_work_without_a_session_is_refused(self):
+        """Supplying a lease is not itself evidence of ownership."""
+        session = "prgs-author-task-a"
+        assignment_id, lease_id = self._allocate(session)
+        res = self._resolve_session(assignment_id=assignment_id, lease_id=lease_id)
+        self.assertFalse(res.get("ok"))
+        self.assertEqual(
+            res.get("reason_code"), "workflow_session_required_for_allocated_work"
+        )
+
+    def test_existing_issue_lock_supplies_its_per_task_session(self):
+        lock_session = issue_lock_store.mint_task_session_id(
+            issue_lock_store.AUTHOR_ISSUE_WORK_LEASE
+        )
+        path = issue_lock_store.lock_file_path(
+            remote="prgs", org=ORG, repo=REPO, issue_number=943,
+            lock_dir=self.lock_dir,
+        )
+        issue_lock_store.write_lock_file(
+            path,
+            {
+                "issue_number": 943,
+                "branch_name": "fix/issue-943-runtime-context-helpers",
+                "work_lease": {
+                    "task_session_id": lock_session,
+                    "claimant": {"username": IDENTITY, "profile": PROFILE},
+                },
+            },
+        ) if hasattr(issue_lock_store, "write_lock_file") else _write_json(
+            path,
+            {
+                "issue_number": 943,
+                "branch_name": "fix/issue-943-runtime-context-helpers",
+                "work_lease": {
+                    "task_session_id": lock_session,
+                    "claimant": {"username": IDENTITY, "profile": PROFILE},
+                },
+            },
+        )
+        res = self._resolve_session()
+        self.assertTrue(res.get("ok"), res)
+        self.assertEqual(res.get("session_id"), lock_session)
+        self.assertEqual(res.get("session_source"), "issue_lock")
+
+    def test_issue_lock_owned_by_another_identity_is_refused(self):
+        path = issue_lock_store.lock_file_path(
+            remote="prgs", org=ORG, repo=REPO, issue_number=943,
+            lock_dir=self.lock_dir,
+        )
+        _write_json(
+            path,
+            {
+                "issue_number": 943,
+                "work_lease": {
+                    "task_session_id": "author_issue_work-" + "0" * 16,
+                    "claimant": {"username": "someone-else", "profile": PROFILE},
+                },
+            },
+        )
+        res = self._resolve_session()
+        self.assertFalse(res.get("ok"))
+        self.assertEqual(res.get("reason_code"), "issue_lock_owner_mismatch")
+
+    def test_unallocated_bootstrap_mints_a_per_task_key(self):
+        res = self._resolve_session()
+        self.assertTrue(res.get("ok"), res)
+        self.assertEqual(res.get("session_source"), "minted_task_key")
+        self.assertRegex(res["session_id"], TASK_KEY_RE)
+
+    def test_minted_key_contains_no_process_identifier(self):
+        res = self._resolve_session()
+        self.assertNotIn(str(os.getpid()), res["session_id"])
+        self.assertNotIn(PROFILE, res["session_id"])
+
+    def test_sequential_tasks_on_one_daemon_do_not_share_ownership(self):
+        """The round-1 defect: one identifier per process for every task."""
+        first = self._resolve_session()["session_id"]
+        second = self._resolve_session()["session_id"]
+        third = self._resolve_session()["session_id"]
+        self.assertNotEqual(first, second)
+        self.assertNotEqual(second, third)
+        self.assertEqual(len({first, second, third}), 3)
+
+    def test_no_process_lifetime_cache_remains(self):
+        self.assertFalse(hasattr(gms, "_ACTIVE_SESSION_ID"))
+        self.assertFalse(hasattr(gms, "_current_session_id"))
+
+
+class MutationAuthorityTests(unittest.TestCase):
+    """F3/F4: one coherent authority pair, drift detected, no silent fallback."""
+
+    def _ctx(self, **over):
+        base = {"identity": IDENTITY, "profile_name": PROFILE}
+        base.update(over)
+        return base
+
+    def test_matching_live_and_pinned_authority_resolves(self):
+        with mock.patch.object(gms, "get_profile",
+                               return_value={"profile_name": PROFILE}), \
+                mock.patch.object(gms, "_authenticated_username",
+                                  return_value=IDENTITY), \
+                mock.patch.object(gms.session_ctx, "get_session_context",
+                                  return_value=self._ctx()):
+            res = gms._active_mutation_authority("gitea.prgs.cc")
+        self.assertTrue(res.get("ok"), res)
+        self.assertEqual(res["identity"], IDENTITY)
+        self.assertEqual(res["profile_name"], PROFILE)
+
+    def test_identity_drift_fails_closed(self):
+        with mock.patch.object(gms, "get_profile",
+                               return_value={"profile_name": PROFILE}), \
+                mock.patch.object(gms, "_authenticated_username",
+                                  return_value="someone-else"), \
+                mock.patch.object(gms.session_ctx, "get_session_context",
+                                  return_value=self._ctx()):
+            res = gms._active_mutation_authority("gitea.prgs.cc")
+        self.assertFalse(res.get("ok"))
+        self.assertEqual(res.get("reason_code"), "authority_identity_drift")
+        self.assertEqual(res.get("expected"), IDENTITY)
+        self.assertEqual(res.get("actual"), "someone-else")
+
+    def test_profile_drift_fails_closed(self):
+        with mock.patch.object(gms, "get_profile",
+                               return_value={"profile_name": "prgs-controller"}), \
+                mock.patch.object(gms, "_authenticated_username",
+                                  return_value=IDENTITY), \
+                mock.patch.object(gms.session_ctx, "get_session_context",
+                                  return_value=self._ctx()):
+            res = gms._active_mutation_authority("gitea.prgs.cc")
+        self.assertFalse(res.get("ok"))
+        self.assertEqual(res.get("reason_code"), "authority_profile_drift")
+
+    def test_identity_and_profile_never_come_from_different_snapshots(self):
+        """Round 2's mixed pair: pinned identity plus live profile."""
+        with mock.patch.object(gms, "get_profile",
+                               return_value={"profile_name": "prgs-controller"}), \
+                mock.patch.object(gms, "_authenticated_username",
+                                  return_value="new-identity"), \
+                mock.patch.object(gms.session_ctx, "get_session_context",
+                                  return_value=self._ctx()):
+            res = gms._active_mutation_authority("gitea.prgs.cc")
+            self.assertFalse(res.get("ok"))
+            self.assertIsNone(gms._active_username("gitea.prgs.cc"))
+            self.assertIsNone(gms._active_profile_name("gitea.prgs.cc"))
+
+    def test_unresolvable_profile_is_a_structured_refusal_not_a_fallback(self):
+        """F4: no bare-except fallback to a previously pinned profile name."""
+        with mock.patch.object(gms, "get_profile",
+                               side_effect=RuntimeError("profile disabled")), \
+                mock.patch.object(gms, "_authenticated_username",
+                                  return_value=IDENTITY), \
+                mock.patch.object(gms.session_ctx, "get_session_context",
+                                  return_value=self._ctx()):
+            res = gms._active_mutation_authority("gitea.prgs.cc")
+        self.assertFalse(res.get("ok"))
+        self.assertEqual(res.get("reason_code"), "authority_profile_unresolved")
+        self.assertNotEqual(res.get("profile_name"), PROFILE)
+
+    def test_malformed_profile_without_name_fails_closed(self):
+        with mock.patch.object(gms, "get_profile", return_value={}), \
+                mock.patch.object(gms, "_authenticated_username",
+                                  return_value=IDENTITY), \
+                mock.patch.object(gms.session_ctx, "get_session_context",
+                                  return_value=None):
+            res = gms._active_mutation_authority("gitea.prgs.cc")
+        self.assertFalse(res.get("ok"))
+        self.assertEqual(res.get("reason_code"), "authority_profile_unresolved")
+
+    def test_unresolved_identity_fails_closed(self):
+        for value in (None, "", "   "):
+            with self.subTest(identity=value):
+                with mock.patch.object(gms, "get_profile",
+                                       return_value={"profile_name": PROFILE}), \
+                        mock.patch.object(gms, "_authenticated_username",
+                                          return_value=value), \
+                        mock.patch.object(gms.session_ctx, "get_session_context",
+                                          return_value=None):
+                    res = gms._active_mutation_authority("gitea.prgs.cc")
+                self.assertFalse(res.get("ok"))
+                self.assertEqual(
+                    res.get("reason_code"), "authority_identity_unresolved"
+                )
+
+    def test_missing_host_cannot_yield_an_identity(self):
+        with mock.patch.object(gms, "get_profile",
+                               return_value={"profile_name": PROFILE}), \
+                mock.patch.object(gms.session_ctx, "get_session_context",
+                                  return_value=None):
+            res = gms._active_mutation_authority(None)
+        self.assertFalse(res.get("ok"))
+        self.assertEqual(res.get("reason_code"), "authority_identity_unresolved")
+
+    def test_expected_username_is_never_substituted_for_authentication(self):
+        with mock.patch.object(
+            gms, "get_profile",
+            return_value={"profile_name": PROFILE, "username": IDENTITY},
+        ), mock.patch.object(gms, "_authenticated_username", return_value=None), \
+                mock.patch.object(gms.session_ctx, "get_session_context",
+                                  return_value={"expected_username": IDENTITY}):
+            self.assertIsNone(gms._active_username("gitea.prgs.cc"))
+
+    def test_accessors_share_one_snapshot(self):
+        with mock.patch.object(gms, "get_profile",
+                               return_value={"profile_name": PROFILE}), \
+                mock.patch.object(gms, "_authenticated_username",
+                                  return_value=IDENTITY), \
+                mock.patch.object(gms.session_ctx, "get_session_context",
+                                  return_value=self._ctx()):
+            self.assertEqual(gms._active_username("gitea.prgs.cc"), IDENTITY)
+            self.assertEqual(gms._active_profile_name("gitea.prgs.cc"), PROFILE)
+
+
+class AuthorMutationBlockTests(unittest.TestCase):
+    """Preserved: the structured refusal shape review 622 confirmed correct."""
+
+    def test_matches_the_sibling_refusal_shape(self):
+        res = gms._author_mutation_block(["stopped"])
+        self.assertIs(res["success"], False)
+        self.assertIs(res["performed"], False)
+        self.assertEqual(res["outcome"], "REFUSED")
+        self.assertEqual(res["reasons"], ["stopped"])
+
+    def test_carries_reason_code_and_transport_fields(self):
+        res = gms._author_mutation_block(
+            ["nope"], reason_code="authority_identity_drift",
+            retryable=False, transport_survives=True,
+            expected="a", actual="b", issue_number=943,
+        )
+        self.assertEqual(res["reason_code"], "authority_identity_drift")
+        self.assertIs(res["retryable"], False)
+        self.assertIs(res["transport_survives"], True)
+        self.assertEqual((res["expected"], res["actual"]), ("a", "b"))
+        self.assertEqual(res["issue_number"], 943)
+        self.assertIs(res["success"], False)
+
+
+class RuntimeHelperResolutionTests(unittest.TestCase):
+    """Every runtime helper the wrapper references is defined and callable.
+
+    Supplements the runtime coverage above; it does not replace it.
+    """
+
+    def test_named_helpers_are_defined_and_callable(self):
+        for name in RUNTIME_HELPERS:
+            with self.subTest(helper=name):
+                self.assertTrue(hasattr(gms, name), f"{name} is not defined")
+                self.assertTrue(callable(getattr(gms, name)))
+
+    def test_every_global_referenced_by_the_wrapper_resolves(self):
+        """The generalised form of the round-1 defect: an unresolvable global."""
+        fn = _wrapper_ast()
+        bound: set[str] = {a.arg for a in fn.args.args}
+        bound |= {a.arg for a in fn.args.kwonlyargs}
+        if fn.args.vararg:
+            bound.add(fn.args.vararg.arg)
+        if fn.args.kwarg:
+            bound.add(fn.args.kwarg.arg)
+        for node in ast.walk(fn):
+            if isinstance(node, ast.Name) and isinstance(
+                node.ctx, (ast.Store, ast.Del)
+            ):
+                bound.add(node.id)
+            elif isinstance(node, (ast.Import, ast.ImportFrom)):
+                for alias in node.names:
+                    bound.add((alias.asname or alias.name).split(".")[0])
+            elif isinstance(node, ast.ExceptHandler) and node.name:
+                bound.add(node.name)
+
+        unresolved = sorted(
+            node.id
+            for node in ast.walk(fn)
+            if isinstance(node, ast.Name)
+            and isinstance(node.ctx, ast.Load)
+            and node.id not in bound
+            and not hasattr(gms, node.id)
+            and not hasattr(builtins, node.id)
+        )
+        self.assertEqual(
+            unresolved, [],
+            f"{WRAPPER_NAME} references undefined globals: {unresolved}",
+        )
+
+    def test_wrapper_wires_the_authority_and_session_resolvers(self):
+        fn = _wrapper_ast()
+        called = {
+            node.func.id
+            for node in ast.walk(fn)
+            if isinstance(node, ast.Call) and isinstance(node.func, ast.Name)
+        }
+        self.assertIn("_active_mutation_authority", called)
+        self.assertIn("_resolve_owner_workflow_session", called)
+        self.assertIn("_author_mutation_block", called)
+
+    def test_wrapper_accepts_an_optional_session_id(self):
+        """ABI addition stays backward compatible: optional, defaulting to None."""
+        fn = _wrapper_ast()
+        names = [a.arg for a in fn.args.args]
+        self.assertIn("session_id", names)
+        offset = len(names) - len(fn.args.defaults)
+        default = fn.args.defaults[names.index("session_id") - offset]
+        self.assertIsInstance(default, ast.Constant)
+        self.assertIsNone(default.value)
+
+
+class Issue941ScopeGuardNotRegressedTests(unittest.TestCase):
+    """Preserved: PR #942's bootstrap-scope wiring still holds."""
+
+    def setUp(self):
+        self._tmp = tempfile.TemporaryDirectory()
+        self.addCleanup(self._tmp.cleanup)
+        self.repo, self.head = _make_control_repo(self._tmp.name)
+
+    def _assessment(self, task: str = BOOTSTRAP_TASK) -> dict:
+        return aib.assess_author_issue_bootstrap(
+            workspace_path=self.repo,
+            canonical_repo_root=self.repo,
+            current_branch="master",
+            head_sha=self.head,
+            porcelain_status="",
+            remote_master_sha=self.head,
+            task=task,
+        )
+
+    def test_bootstrap_task_still_permitted_from_clean_control_checkout(self):
+        res = workflow_scope_guard.assess_root_source_mutation(
+            workspace_path=self.repo,
+            canonical_repo_root=self.repo,
+            role_kind="author",
+            mutation_task=BOOTSTRAP_TASK,
+            porcelain_status="",
+            bootstrap_assessment=self._assessment(),
+        )
+        self.assertFalse(res.get("block"), res)
+        self.assertNotEqual(
+            res.get("blocker_kind"), workflow_scope_guard.BLOCKER_MISSING_WORKTREE
+        )
+
+    def test_bootstrap_task_still_blocked_without_evidence(self):
+        res = workflow_scope_guard.assess_root_source_mutation(
+            workspace_path=self.repo,
+            canonical_repo_root=self.repo,
+            role_kind="author",
+            mutation_task=BOOTSTRAP_TASK,
+            porcelain_status="",
+        )
+        self.assertTrue(res.get("block"))
+        self.assertEqual(
+            res.get("blocker_kind"), workflow_scope_guard.BLOCKER_MISSING_WORKTREE
+        )
+
+    def test_ordinary_author_mutation_still_blocked_from_control_checkout(self):
+        res = workflow_scope_guard.assess_root_source_mutation(
+            workspace_path=self.repo,
+            canonical_repo_root=self.repo,
+            role_kind="author",
+            mutation_task="commit_files",
+            porcelain_status="",
+            bootstrap_assessment=self._assessment(),
+        )
+        self.assertTrue(res.get("block"))
+        self.assertEqual(
+            res.get("blocker_kind"), workflow_scope_guard.BLOCKER_MISSING_WORKTREE
+        )
+
+    def test_create_issue_bootstrap_unchanged(self):
+        self.assertTrue(cib.is_create_issue_task("create_issue"))
+        self.assertFalse(cib.is_create_issue_task(BOOTSTRAP_TASK))
+
+
+def _write_json(path: str, payload: dict) -> None:
+    """Write an issue-lock file directly, for lock-precedence tests."""
+    import json
+
+    os.makedirs(os.path.dirname(path), exist_ok=True)
+    with open(path, "w", encoding="utf-8") as fh:
+        json.dump(payload, fh)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_issue_945_enforcement_path_wiring.py

@@ -0,0 +1,685 @@
+import sys as _sys
+from pathlib import Path as _Path
+_sys.path.insert(0, str(_Path(__file__).resolve().parent))
+from mutation_profile_fixture import shared_mutation_env  # noqa: E402
+"""The renewal waiver reaches the real enforcement paths (#945 B1).
+
+``tests/test_issue_945_owning_pr_renewal_continuation.py`` proves the pure
+pieces: that ``issue_lock_renewal.owning_pr_renewal_from_lock`` rebuilds a
+renewal waiver, that ``_owning_pr_continuation_from_lock`` resolves the two
+dispositions in the right precedence, and that the duplicate gate honours the
+resulting token. None of that proves any *production* path consumes the
+resolver, and review ``623`` demonstrated the gap by reverting the primary call
+site at ``gitea_mcp_server.py:2894`` back to the recovery-only rebuild: the
+whole repository stayed green, failing-test ids byte identical.
+
+This file closes that hole. Every test here starts from a real durable lock
+file written to a temporary lock directory and bound to this process's session
+pointer, then calls the authoritative production entry point — not a helper:
+
+* ``mcp_server._enforce_locked_issue_duplicate_recheck`` — the shared recheck
+  behind ``gitea_commit_files`` and ``gitea_create_pr``
+* ``mcp_server.gitea_assess_work_issue_duplicate`` — the read-only assessor
+* ``mcp_server._prove_author_ownership_for_pr`` — the push / PR-update
+  ownership prover, which is also the existing-PR continuation path
+
+Only the external boundaries are mocked: Gitea HTTP reads (the duplicate
+context fetcher, open-PR and branch listings) and the credential header. The
+reconstruction and enforcement chain under test — lock load, evidence rebuild,
+resolver precedence, and ``issue_work_duplicate_gate`` — runs for real.
+
+``TestRevertingThePrimaryWiringIsDetected`` is the explicit regression the
+review asked for: it reproduces the pre-#945 recovery-only call site and
+asserts the enforcement path then refuses, so the wiring cannot be removed
+silently.
+
+Everything is written under ``tempfile.TemporaryDirectory``. No branch,
+worktree, PR, comment, lease, or lock outside that directory is created, and no
+production Gitea or control-plane state is touched (#945 AC18).
+"""
+import os
+import subprocess
+import sys
+import tempfile
+import unittest
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+from unittest.mock import patch
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+import issue_lock_provenance  # noqa: E402
+import issue_lock_recovery  # noqa: E402
+import issue_lock_renewal  # noqa: E402
+import issue_lock_store  # noqa: E402
+import mcp_server  # noqa: E402
+from issue_work_duplicate_gate import (  # noqa: E402
+    PHASE_COMMIT,
+    PHASE_CREATE_PR,
+    PHASE_LOCK,
+    PHASE_PUSH,
+)
+
+ISSUE = 4948
+OWNING_PR = 4949
+OTHER_PR = 4950
+OTHER_ISSUE = 4951
+BRANCH = f"fix/issue-{ISSUE}-renewal-wiring"
+OTHER_BRANCH = f"fix/issue-{ISSUE}-competing"
+HEAD = "e" * 40
+OTHER_HEAD = "f" * 40
+IDENTITY = "example-user"
+PROFILE = "test-author-prgs"
+ORG = "Scaled-Tech-Consulting"
+REPO = "Gitea-Tools"
+HOST = "gitea.prgs.cc"
+
+
+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 shifted_ts(hours: int = 4) -> str:
+    return (
+        (datetime.now(timezone.utc) + timedelta(hours=hours))
+        .isoformat()
+        .replace("+00:00", "Z")
+    )
+
+
+def owning_pr(number=OWNING_PR, ref=BRANCH, sha=HEAD, issue=ISSUE):
+    return {
+        "number": number,
+        "title": f"fix: something (Closes #{issue})",
+        "body": f"Closes #{issue}.",
+        "head": {"ref": ref, "sha": sha},
+    }
+
+
+def renewal_block(
+    *,
+    pr_number=OWNING_PR,
+    branch=BRANCH,
+    head=HEAD,
+    identity=IDENTITY,
+    profile=PROFILE,
+):
+    """The ``lease_renewal`` block ``build_renewal_record`` writes on success."""
+    return {
+        "renewed": True,
+        "renewed_at": shifted_ts(-1),
+        "prior_pid": 4242,
+        "prior_pid_alive": True,
+        "prior_expires_at": shifted_ts(-1),
+        "replacement_pid": os.getpid(),
+        "new_expires_at": shifted_ts(),
+        "identity": identity,
+        "profile": profile,
+        "branch_name": branch,
+        "worktree_path": os.path.realpath(os.getcwd()),
+        "head_sha": head,
+        "remote_head_sha": head,
+        "pr_head_sha": head,
+        "pr_number": pr_number,
+        "reason": "expired lease renewed by its exact recorded owner",
+        "proof": [],
+    }
+
+
+def recovery_block(*, pr_number=OWNING_PR, branch=BRANCH, head=HEAD):
+    """The ``dead_session_recovery`` block ``build_recovery_record`` writes."""
+    return {
+        "recovered": True,
+        "reason": "owning MCP session exited; durable ownership evidence matched",
+        "recovered_at": shifted_ts(-1),
+        "prior_session_pid": 4242,
+        "replacement_session_pid": os.getpid(),
+        "prior_pid_alive": False,
+        "branch_name": branch,
+        "pr_number": pr_number,
+        "pr_head": head,
+        "recorded_head": head,
+        "accepted_head": head,
+        "head_relation": issue_lock_recovery.HEAD_RELATION_EQUAL,
+        "identity": IDENTITY,
+        "profile": PROFILE,
+        "proof": [],
+    }
+
+
+class EnforcementPathBase(unittest.TestCase):
+    """Drives production enforcement entry points against a real durable lock.
+
+    The lock lives in a throwaway directory and is bound to this process's
+    session pointer exactly as ``gitea_lock_issue`` binds it, so
+    ``_load_existing_issue_lock()`` resolves it through the ordinary
+    ``read_session_issue_lock()`` path rather than a test shortcut.
+    """
+
+    def setUp(self):
+        self.lock_dir = tempfile.TemporaryDirectory()
+        self.addCleanup(self.lock_dir.cleanup)
+        self.worktree = os.path.realpath(os.getcwd())
+        self.remotes = patch.dict(
+            mcp_server.REMOTES,
+            {"prgs": {"host": HOST, "org": ORG, "repo": REPO}},
+        )
+        self.remotes.start()
+        self.addCleanup(patch.stopall)
+        mcp_server._IDENTITY_CACHE.clear()
+
+    # ── fixtures ────────────────────────────────────────────────────────────
+
+    def build_lock(
+        self,
+        *,
+        issue_number=ISSUE,
+        branch=BRANCH,
+        renewal=None,
+        recovery=None,
+        claimant=None,
+        pid=None,
+        live=True,
+    ):
+        pid = os.getpid() if pid is None else pid
+        claimant = claimant or {"username": IDENTITY, "profile": PROFILE}
+        expires = shifted_ts() if live else shifted_ts(-1)
+        data = {
+            "issue_number": issue_number,
+            "branch_name": branch,
+            "remote": "prgs",
+            "org": ORG,
+            "repo": REPO,
+            "worktree_path": self.worktree,
+            "session_pid": pid,
+            "pid": pid,
+            "claimant": dict(claimant),
+            "work_lease": {
+                "operation_type": issue_lock_store.AUTHOR_ISSUE_WORK_LEASE,
+                "issue_number": issue_number,
+                "branch": branch,
+                "worktree_path": self.worktree,
+                "claimant": dict(claimant),
+                "created_at": shifted_ts(-1),
+                "last_heartbeat_at": shifted_ts(0) if live else shifted_ts(-1),
+                "expires_at": expires,
+            },
+            "lock_provenance": issue_lock_provenance.build_sanctioned_lock_provenance(
+                tool="gitea_lock_issue",
+                claimant=dict(claimant),
+            ),
+        }
+        if renewal is not None:
+            data["lease_renewal"] = renewal
+        if recovery is not None:
+            data["dead_session_recovery"] = recovery
+        return data
+
+    def bind(self, data):
+        """Persist the lock and bind it to this process, as the server does."""
+        issue_lock_store.bind_session_lock(data, self.lock_dir.name)
+        return data
+
+    def env(self):
+        return shared_mutation_env(
+            PROFILE,
+            include_example_repo=True,
+            GITEA_ISSUE_LOCK_DIR=self.lock_dir.name,
+        )
+
+    def gitea_reads(self, *, open_prs, branch_names=None):
+        """Patch only the external Gitea read boundary."""
+        branch_names = [BRANCH] if branch_names is None else branch_names
+        return (
+            patch("mcp_server.get_auth_header", return_value="token x"),
+            patch(
+                "mcp_server.issue_duplicate_context_fetcher",
+                side_effect=lambda h, o, r, auth, issue_number: (
+                    list(open_prs), list(branch_names), {"status": "not_claimed"}
+                ),
+            ),
+            patch("mcp_server._list_open_pulls", return_value=list(open_prs)),
+            patch(
+                "mcp_server.api_get_all",
+                return_value=[
+                    {"name": n, "commit": {"id": HEAD}} for n in branch_names
+                ],
+            ),
+        )
+
+    # ── production entry points ─────────────────────────────────────────────
+
+    def run_duplicate_recheck(self, *, phase, open_prs, branch_names=None):
+        """The real shared recheck behind gitea_commit_files / gitea_create_pr."""
+        patches = self.gitea_reads(open_prs=open_prs, branch_names=branch_names)
+        with patches[0], patches[1], patches[2], patches[3]:
+            with patch.dict(os.environ, self.env(), clear=True):
+                os.environ["GITEA_ISSUE_LOCK_DIR"] = self.lock_dir.name
+                return mcp_server._enforce_locked_issue_duplicate_recheck(
+                    "prgs", phase, host=HOST, org=ORG, repo=REPO
+                )
+
+    def run_readonly_assessor(
+        self, *, open_prs, issue_number=ISSUE, branch=BRANCH, branch_names=None
+    ):
+        """The real read-only duplicate assessor MCP tool."""
+        patches = self.gitea_reads(open_prs=open_prs, branch_names=branch_names)
+        with patches[0], patches[1], patches[2], patches[3]:
+            with patch.dict(os.environ, self.env(), clear=True):
+                os.environ["GITEA_ISSUE_LOCK_DIR"] = self.lock_dir.name
+                return mcp_server.gitea_assess_work_issue_duplicate(
+                    issue_number=issue_number,
+                    branch_name=branch,
+                    phase=PHASE_COMMIT,
+                    remote="prgs",
+                    host=HOST,
+                    org=ORG,
+                    repo=REPO,
+                )
+
+    def run_ownership_prover(
+        self, *, pr_number=OWNING_PR, branch=BRANCH, issue_number=ISSUE
+    ):
+        """The real push / PR-update ownership prover (existing-PR continuation)."""
+        with patch("mcp_server.get_auth_header", return_value="token x"):
+            with patch.dict(os.environ, self.env(), clear=True):
+                os.environ["GITEA_ISSUE_LOCK_DIR"] = self.lock_dir.name
+                return mcp_server._prove_author_ownership_for_pr(
+                    pr_number=pr_number,
+                    pr_title=f"fix: something (Closes #{issue_number})",
+                    pr_body=f"Closes #{issue_number}.",
+                    source_branch=branch,
+                    remote="prgs",
+                    host=HOST,
+                    org=ORG,
+                    repo=REPO,
+                    worktree_path=self.worktree,
+                )
+
+
+# ─────────────── B1: renewal evidence reaches every enforcement path ───────────
+
+
+class TestRenewalReachesEnforcementPaths(EnforcementPathBase):
+    """A renewal-only lock must exempt its owning PR at the real call sites.
+
+    Each of these fails if its call site is reverted to the recovery-only
+    rebuild, because the lock deliberately carries no ``dead_session_recovery``
+    block at all.
+    """
+
+    def setUp(self):
+        super().setUp()
+        self.bind(self.build_lock(renewal=renewal_block()))
+
+    def test_commit_duplicate_recheck_permits_the_owning_pr(self):
+        blocked = self.run_duplicate_recheck(
+            phase=PHASE_COMMIT, open_prs=[owning_pr()]
+        )
+        self.assertIsNone(
+            blocked,
+            "commit recheck refused the PR the renewal already proved it owns; "
+            "the resolver is not wired into gitea_mcp_server:2894",
+        )
+
+    def test_create_pr_duplicate_recheck_permits_the_owning_pr(self):
+        blocked = self.run_duplicate_recheck(
+            phase=PHASE_CREATE_PR, open_prs=[owning_pr()]
+        )
+        self.assertIsNone(blocked)
+
+    def test_read_only_assessor_reports_the_same_exemption(self):
+        result = self.run_readonly_assessor(open_prs=[owning_pr()])
+        self.assertTrue(result["success"])
+        self.assertFalse(result["block"])
+        self.assertTrue(result["owning_pr_recovery_exempted"])
+        self.assertEqual(result["linked_open_pr"], OWNING_PR)
+
+    def test_push_ownership_prover_carries_the_renewal_evidence(self):
+        ownership = self.run_ownership_prover()
+        self.assertTrue(ownership["proven"], ownership["reasons"])
+        token = ownership["recovered_owning_pr"]
+        self.assertIsNotNone(
+            token,
+            "push prover produced no continuation evidence from a renewal lock; "
+            "the resolver is not wired into gitea_mcp_server:19464",
+        )
+        self.assertEqual(token["pr_number"], OWNING_PR)
+        self.assertEqual(token["branch_name"], BRANCH)
+        self.assertEqual(token["head_sha"], HEAD)
+
+    def test_all_enforcement_paths_decide_alike_from_one_lock(self):
+        """AC: commit, create-PR, assessor and prover agree on one lock."""
+        for phase in (PHASE_COMMIT, PHASE_CREATE_PR, PHASE_PUSH, PHASE_LOCK):
+            with self.subTest(phase=phase):
+                self.assertIsNone(
+                    self.run_duplicate_recheck(phase=phase, open_prs=[owning_pr()])
+                )
+        assessor = self.run_readonly_assessor(open_prs=[owning_pr()])
+        prover = self.run_ownership_prover()
+        self.assertTrue(assessor["owning_pr_recovery_exempted"])
+        self.assertEqual(
+            assessor["linked_open_pr"], prover["recovered_owning_pr"]["pr_number"]
+        )
+
+
+class TestDeadSessionRecoveryStillReachesEnforcementPaths(EnforcementPathBase):
+    """#755/#768 recovery must be unchanged by the #945 resolver."""
+
+    def setUp(self):
+        super().setUp()
+        self.bind(self.build_lock(recovery=recovery_block()))
+
+    def test_commit_recheck_still_permits_a_recovered_owning_pr(self):
+        self.assertIsNone(
+            self.run_duplicate_recheck(phase=PHASE_COMMIT, open_prs=[owning_pr()])
+        )
+
+    def test_assessor_still_reports_the_recovery_exemption(self):
+        result = self.run_readonly_assessor(open_prs=[owning_pr()])
+        self.assertTrue(result["owning_pr_recovery_exempted"])
+
+    def test_prover_still_carries_recovery_evidence(self):
+        token = self.run_ownership_prover()["recovered_owning_pr"]
+        self.assertEqual(token["pr_number"], OWNING_PR)
+
+
+# ──────────────── B1: the explicit anti-revert regression test ────────────────
+
+
+class TestRevertingThePrimaryWiringIsDetected(EnforcementPathBase):
+    """Reproduce the pre-#945 call site and prove the path then refuses.
+
+    Review ``623`` reverted ``gitea_mcp_server.py:2894`` from
+    ``_owning_pr_continuation_from_lock`` to
+    ``issue_lock_recovery.recovered_owning_pr_from_lock`` and found the entire
+    repository still green. Substituting exactly that pre-fix behaviour here
+    makes the enforcement path block, so the causal link between the resolver
+    and the gate's answer is asserted, not assumed.
+    """
+
+    def setUp(self):
+        super().setUp()
+        self.bind(self.build_lock(renewal=renewal_block()))
+
+    def test_recovery_only_rebuild_reintroduces_the_945_refusal(self):
+        with patch.object(
+            mcp_server,
+            "_owning_pr_continuation_from_lock",
+            side_effect=issue_lock_recovery.recovered_owning_pr_from_lock,
+        ):
+            blocked = self.run_duplicate_recheck(
+                phase=PHASE_COMMIT, open_prs=[owning_pr()]
+            )
+        self.assertIsNotNone(
+            blocked,
+            "the pre-#945 recovery-only rebuild must lose the renewal waiver; "
+            "if this passes, the enforcement path is not consuming the resolver",
+        )
+        self.assertTrue(blocked["block"])
+        self.assertFalse(blocked["owning_pr_recovery_exempted"])
+
+    def test_restoring_the_resolver_restores_continuation(self):
+        self.assertIsNone(
+            self.run_duplicate_recheck(phase=PHASE_COMMIT, open_prs=[owning_pr()])
+        )
+
+    def test_read_only_assessor_is_wired_to_the_same_resolver(self):
+        with patch.object(
+            mcp_server,
+            "_owning_pr_continuation_from_lock",
+            side_effect=issue_lock_recovery.recovered_owning_pr_from_lock,
+        ):
+            result = self.run_readonly_assessor(open_prs=[owning_pr()])
+        self.assertTrue(result["block"])
+        self.assertFalse(result["owning_pr_recovery_exempted"])
+
+    def test_push_prover_is_wired_to_the_same_resolver(self):
+        with patch.object(
+            mcp_server,
+            "_owning_pr_continuation_from_lock",
+            side_effect=issue_lock_recovery.recovered_owning_pr_from_lock,
+        ):
+            ownership = self.run_ownership_prover()
+        self.assertIsNone(ownership["recovered_owning_pr"])
+
+
+# ───────────────── B1: the exemption is not widened at the call sites ─────────
+
+
+class TestEnforcementPathsStillFailClosed(EnforcementPathBase):
+    def assert_blocked(self, result):
+        self.assertIsNotNone(result, "expected a fail-closed refusal")
+        self.assertTrue(result["block"])
+        return result
+
+    def test_open_pr_alone_grants_no_exemption(self):
+        """No renewal and no recovery block: the open PR still blocks."""
+        self.bind(self.build_lock())
+        blocked = self.assert_blocked(
+            self.run_duplicate_recheck(phase=PHASE_COMMIT, open_prs=[owning_pr()])
+        )
+        self.assertFalse(blocked["owning_pr_recovery_exempted"])
+
+    def test_second_pr_is_refused(self):
+        self.bind(self.build_lock(renewal=renewal_block()))
+        self.assert_blocked(
+            self.run_duplicate_recheck(
+                phase=PHASE_COMMIT,
+                open_prs=[owning_pr(), owning_pr(number=OTHER_PR, ref=OTHER_BRANCH)],
+            )
+        )
+
+    def test_evidence_naming_another_pr_is_refused(self):
+        self.bind(self.build_lock(renewal=renewal_block(pr_number=OTHER_PR)))
+        self.assert_blocked(
+            self.run_duplicate_recheck(phase=PHASE_COMMIT, open_prs=[owning_pr()])
+        )
+
+    def test_unrelated_branch_is_refused(self):
+        self.bind(self.build_lock(renewal=renewal_block(branch=OTHER_BRANCH)))
+        self.assert_blocked(
+            self.run_duplicate_recheck(phase=PHASE_COMMIT, open_prs=[owning_pr()])
+        )
+
+    def test_live_head_divergence_is_refused(self):
+        """Force-push or unrelated remote movement: live PR head no longer matches."""
+        self.bind(self.build_lock(renewal=renewal_block()))
+        self.assert_blocked(
+            self.run_duplicate_recheck(
+                phase=PHASE_COMMIT, open_prs=[owning_pr(sha=OTHER_HEAD)]
+            )
+        )
+
+    def test_stale_recorded_head_is_refused(self):
+        """The renewal names a head the live PR never had."""
+        self.bind(self.build_lock(renewal=renewal_block(head=OTHER_HEAD)))
+        self.assert_blocked(
+            self.run_duplicate_recheck(phase=PHASE_COMMIT, open_prs=[owning_pr()])
+        )
+
+    def test_local_remote_head_divergence_is_refused(self):
+        record = renewal_block()
+        record["remote_head_sha"] = OTHER_HEAD
+        self.bind(self.build_lock(renewal=record))
+        self.assert_blocked(
+            self.run_duplicate_recheck(phase=PHASE_COMMIT, open_prs=[owning_pr()])
+        )
+
+    def test_identity_mismatch_with_the_lock_claimant_is_refused(self):
+        self.bind(self.build_lock(renewal=renewal_block(identity="someone-else")))
+        self.assert_blocked(
+            self.run_duplicate_recheck(phase=PHASE_COMMIT, open_prs=[owning_pr()])
+        )
+
+    def test_profile_mismatch_with_the_lock_claimant_is_refused(self):
+        self.bind(self.build_lock(renewal=renewal_block(profile="test-reviewer-prgs")))
+        self.assert_blocked(
+            self.run_duplicate_recheck(phase=PHASE_COMMIT, open_prs=[owning_pr()])
+        )
+
+    def test_ungranted_renewal_block_is_refused(self):
+        record = renewal_block()
+        record["renewed"] = False
+        self.bind(self.build_lock(renewal=record))
+        self.assert_blocked(
+            self.run_duplicate_recheck(phase=PHASE_COMMIT, open_prs=[owning_pr()])
+        )
+
+    def test_malformed_renewal_block_is_refused(self):
+        record = renewal_block()
+        record["pr_number"] = "not-a-number"
+        self.bind(self.build_lock(renewal=record))
+        self.assert_blocked(
+            self.run_duplicate_recheck(phase=PHASE_COMMIT, open_prs=[owning_pr()])
+        )
+
+    def test_wrong_issue_evidence_cannot_be_copied_onto_another_lock(self):
+        """A renewal block copied onto a lock for a different issue proves nothing.
+
+        The rebuilt token takes its ``issue_number`` from the lock it is found
+        on, not from the record, so a block lifted onto another issue's lock
+        claims that issue while still naming the original PR. That copied
+        evidence must not waive the genuine duplicate the other issue has.
+        """
+        self.bind(
+            self.build_lock(issue_number=OTHER_ISSUE, renewal=renewal_block())
+        )
+        blocked = self.assert_blocked(
+            self.run_duplicate_recheck(
+                phase=PHASE_COMMIT,
+                # The real open PR for OTHER_ISSUE is a different PR entirely.
+                open_prs=[
+                    owning_pr(number=OTHER_PR, ref=OTHER_BRANCH, issue=OTHER_ISSUE)
+                ],
+                branch_names=[OTHER_BRANCH],
+            )
+        )
+        self.assertFalse(blocked["owning_pr_recovery_exempted"])
+
+    def test_refusal_carries_complete_structured_fields(self):
+        self.bind(self.build_lock())
+        blocked = self.assert_blocked(
+            self.run_duplicate_recheck(phase=PHASE_COMMIT, open_prs=[owning_pr()])
+        )
+        for field in (
+            "block",
+            "outcome",
+            "reasons",
+            "owning_pr_recovery_exempted",
+            "owning_pr_recovery_notes",
+            "linked_open_pr",
+            "linked_open_pr_count",
+        ):
+            with self.subTest(field=field):
+                self.assertIn(field, blocked)
+        self.assertTrue(blocked["reasons"])
+
+
+class TestSequentialTasksStayIsolated(EnforcementPathBase):
+    """One long-lived daemon serves many tasks; a waiver must not leak forward."""
+
+    def test_a_later_lock_without_evidence_does_not_inherit_the_earlier_waiver(self):
+        self.bind(self.build_lock(renewal=renewal_block()))
+        self.assertIsNone(
+            self.run_duplicate_recheck(phase=PHASE_COMMIT, open_prs=[owning_pr()])
+        )
+
+        # Second task in the same process: a fresh lock, no renewal evidence.
+        self.bind(
+            self.build_lock(issue_number=OTHER_ISSUE, branch=OTHER_BRANCH)
+        )
+        blocked = self.run_duplicate_recheck(
+            phase=PHASE_COMMIT,
+            open_prs=[owning_pr(number=OTHER_PR, ref=OTHER_BRANCH, issue=OTHER_ISSUE)],
+            branch_names=[OTHER_BRANCH],
+        )
+        self.assertIsNotNone(blocked)
+        self.assertFalse(blocked["owning_pr_recovery_exempted"])
+
+
+# ───────────── F2: what the caller binding actually is, and is not ────────────
+
+
+class TestCallerBindingIsStructuralNotFieldComparison(unittest.TestCase):
+    """Document, in executable form, the binding this patch really provides.
+
+    Review ``623`` found that the claimant check in
+    ``owning_pr_renewal_from_lock`` compares two fields of one server-written
+    lock file and is therefore not bound to the authenticated caller. That is
+    correct, and these tests assert the true guarantee rather than the
+    overstated one: lock *selection* is process-scoped, and the claimant check
+    is an internal-consistency check.
+
+    No PID-derived, cached, or process-lifetime session authority is invented
+    here — the process scoping asserted below is pre-existing behaviour of
+    ``issue_lock_store``, not something this patch adds.
+    """
+
+    def test_lock_selection_is_keyed_to_the_operating_system_process(self):
+        with tempfile.TemporaryDirectory() as root:
+            pointer = issue_lock_store.session_pointer_path(root)
+            self.assertEqual(
+                os.path.basename(pointer), f"session-{os.getpid()}.json"
+            )
+
+    def test_a_lock_bound_by_another_process_is_not_reachable(self):
+        """The structural protection: a foreign session pointer is not read."""
+        with tempfile.TemporaryDirectory() as root:
+            foreign_pointer = os.path.join(root, f"session-{os.getpid() + 1}.json")
+            issue_lock_store.save_lock_file(
+                foreign_pointer, {"lock_file_path": "/nonexistent/foreign.json"}
+            )
+            self.assertIsNone(issue_lock_store.read_session_issue_lock(root))
+
+    def test_claimant_check_does_not_consult_the_live_authenticated_caller(self):
+        """The honest limit: agreement is internal to the lock document.
+
+        A renewal block whose identity/profile agree with the claimant recorded
+        on the same lock rebuilds successfully, regardless of who is
+        authenticated. Live identity and profile are enforced by the separate
+        mutation-authority and profile gates, not by this rebuild.
+        """
+        lock = {
+            "issue_number": ISSUE,
+            "branch_name": BRANCH,
+            "claimant": {"username": "unrelated-recorded-user", "profile": PROFILE},
+            "lease_renewal": renewal_block(identity="unrelated-recorded-user"),
+        }
+        token = issue_lock_renewal.owning_pr_renewal_from_lock(lock)
+        self.assertIsNotNone(token)
+        self.assertEqual(token["pr_number"], OWNING_PR)
+
+    def test_internal_disagreement_is_what_the_check_actually_rejects(self):
+        lock = {
+            "issue_number": ISSUE,
+            "branch_name": BRANCH,
+            "claimant": {"username": IDENTITY, "profile": PROFILE},
+            "lease_renewal": renewal_block(identity="someone-else"),
+        }
+        self.assertIsNone(issue_lock_renewal.owning_pr_renewal_from_lock(lock))
+
+
+class TestNoDurableArtifacts(EnforcementPathBase):
+    def test_enforcement_runs_leave_nothing_outside_the_temp_lock_dir(self):
+        before = sorted(os.listdir(self.lock_dir.name))
+        self.bind(self.build_lock(renewal=renewal_block()))
+        self.run_duplicate_recheck(phase=PHASE_COMMIT, open_prs=[owning_pr()])
+        self.run_ownership_prover()
+        after = sorted(os.listdir(self.lock_dir.name))
+        self.assertNotEqual(before, after, "the test must have written its lock")
+        self.assertTrue(
+            all(
+                os.path.realpath(os.path.join(self.lock_dir.name, name)).startswith(
+                    os.path.realpath(self.lock_dir.name)
+                )
+                for name in after
+            )
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_issue_945_owning_pr_renewal_continuation.py

@@ -0,0 +1,602 @@
+import sys as _sys
+from pathlib import Path as _Path
+_sys.path.insert(0, str(_Path(__file__).resolve().parent))
+from mutation_profile_fixture import shared_mutation_env  # noqa: F401,E402
+"""Exact-owner renewal keeps its owning-PR waiver past lock_issue (#945).
+
+#755 taught the duplicate-work gate that a sanctioned *dead-session recovery*
+owns its open PR, and #768 taught the later gates to rebuild that proof from the
+durable lock. #760 added the exact-owner *renewal* disposition and granted it
+the same waiver inside ``gitea_lock_issue`` — but never added the matching
+rebuild. So an ordinary renewal held the waiver only for the duration of the
+lock call: ``_enforce_locked_issue_duplicate_recheck`` asked
+``recovered_owning_pr_from_lock``, which reads only ``dead_session_recovery``,
+and the very next commit was refused ``duplicate_commit_prevented`` with
+``owning_pr_recovery_exempted: false`` on the PR the renewal had just proved.
+
+``TestPreFixReproduction`` pins that defect directly: the recovery-only rebuild
+still returns ``None`` for a renewal lock, which is exactly why the gates lost
+the waiver. Everything else proves the renewal half now survives, that recovery
+is unchanged, and that no path grants an exemption on weaker evidence.
+
+Every fixture here is an in-memory mapping. Nothing writes a branch, worktree,
+lock file, lease, comment, or PR (#945 AC18).
+"""
+import copy
+import sys
+import unittest
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+import gitea_mcp_server  # noqa: E402
+import issue_lock_recovery  # noqa: E402
+import issue_lock_renewal  # noqa: E402
+from issue_work_duplicate_gate import (  # noqa: E402
+    OUTCOME_DUPLICATE_WORK_NOT_PREVENTED,
+    PHASE_COMMIT,
+    PHASE_CREATE_PR,
+    PHASE_LOCK,
+    PHASE_PUSH,
+    assess_work_issue_duplicate_gate,
+)
+
+ISSUE = 4945
+OWNING_PR = 4946
+OTHER_PR = 4947
+BRANCH = f"fix/issue-{ISSUE}-owning-pr-renewal"
+OTHER_BRANCH = f"fix/issue-{ISSUE}-competing"
+HEAD = "a" * 40
+OTHER_HEAD = "b" * 40
+IDENTITY = "example-user"
+PROFILE = "test-author-prgs"
+
+
+def renewal_record(**overrides):
+    """The ``lease_renewal`` block ``build_renewal_record`` writes on success."""
+    record = {
+        "renewed": True,
+        "renewed_at": "2026-01-01T00:00:00Z",
+        "prior_pid": 4242,
+        "prior_pid_alive": True,
+        "prior_expires_at": "2026-01-01T00:00:00Z",
+        "replacement_pid": 4243,
+        "new_expires_at": "2026-01-01T00:10:00Z",
+        "identity": IDENTITY,
+        "profile": PROFILE,
+        "branch_name": BRANCH,
+        "worktree_path": f"branches/issue-{ISSUE}-owning-pr-renewal",
+        "head_sha": HEAD,
+        "remote_head_sha": HEAD,
+        "pr_head_sha": HEAD,
+        "pr_number": OWNING_PR,
+        "reason": "expired lease renewed by its exact recorded owner",
+        "proof": [],
+    }
+    record.update(overrides)
+    return record
+
+
+def renewal_lock(record=None, *, issue_number=ISSUE, claimant=True, **lock_overrides):
+    lock = {
+        "issue_number": issue_number,
+        "branch_name": BRANCH,
+        "lease_renewal": renewal_record() if record is None else record,
+    }
+    if claimant:
+        lock["claimant"] = {"username": IDENTITY, "profile": PROFILE}
+    lock.update(lock_overrides)
+    return lock
+
+
+def recovery_lock(pr_number=OWNING_PR, head=HEAD):
+    """A lock carrying sanctioned dead-session recovery evidence (#755/#768)."""
+    return {
+        "issue_number": ISSUE,
+        "branch_name": BRANCH,
+        "claimant": {"username": IDENTITY, "profile": PROFILE},
+        "dead_session_recovery": {
+            "recovered": True,
+            "branch_name": BRANCH,
+            "pr_number": pr_number,
+            "pr_head": head,
+            "recorded_head": head,
+            "accepted_head": head,
+            "head_relation": issue_lock_recovery.HEAD_RELATION_EQUAL,
+        },
+    }
+
+
+def owning_pr(number=OWNING_PR, ref=BRANCH, sha=HEAD, issue=ISSUE):
+    return {
+        "number": number,
+        "title": f"fix: something (Closes #{issue})",
+        "body": f"Closes #{issue}.",
+        "head": {"ref": ref, "sha": sha},
+    }
+
+
+def gate(phase, *, token, open_prs=None, branch_names=None, locked_branch=BRANCH):
+    return assess_work_issue_duplicate_gate(
+        ISSUE,
+        open_prs=[owning_pr()] if open_prs is None else open_prs,
+        branch_names=branch_names or [],
+        claim_entry={},
+        locked_branch=locked_branch,
+        phase=phase,
+        recovered_owning_pr=token,
+    )
+
+
+# ───────────────────── the defect this issue exists to fix ─────────────────────
+
+
+class TestPreFixReproduction(unittest.TestCase):
+    """The exact wiring gap: renewal evidence was invisible to later gates."""
+
+    def test_recovery_only_rebuild_cannot_see_a_renewal_lock(self):
+        # This is the pre-fix behaviour of every enforcement path. It is correct
+        # for the recovery rebuild to ignore a renewal block -- the defect was
+        # that nothing else looked at it.
+        self.assertIsNone(
+            issue_lock_recovery.recovered_owning_pr_from_lock(renewal_lock())
+        )
+
+    def test_renewal_lock_produced_no_exemption_before_the_fix(self):
+        # Feeding the gate what the pre-fix code fed it (recovery rebuild only)
+        # reproduces the reported refusal at the commit phase.
+        token = issue_lock_recovery.recovered_owning_pr_from_lock(renewal_lock())
+        result = gate(PHASE_COMMIT, token=token)
+        self.assertTrue(result["block"])
+        self.assertEqual(result["outcome"], "duplicate_commit_prevented")
+        self.assertFalse(result["owning_pr_recovery_exempted"])
+        self.assertEqual(result["owning_pr_recovery_notes"], [])
+
+    def test_shared_resolver_now_sees_it(self):
+        self.assertIsNotNone(
+            gitea_mcp_server._owning_pr_continuation_from_lock(renewal_lock())
+        )
+
+
+# ───────────────────────── rebuild: the granted case ─────────────────────────
+
+
+class TestRenewalRebuildGranted(unittest.TestCase):
+    def test_sanctioned_renewal_rebuilds_owning_pr_evidence(self):
+        token = issue_lock_renewal.owning_pr_renewal_from_lock(renewal_lock())
+        self.assertEqual(
+            token,
+            {
+                "issue_number": ISSUE,
+                "pr_number": OWNING_PR,
+                "branch_name": BRANCH,
+                "head_sha": HEAD,
+                "recorded_head": HEAD,
+                "accepted_head": HEAD,
+                "head_relation": "equal",
+            },
+        )
+
+    def test_branch_falls_back_to_the_lock_branch(self):
+        lock = renewal_lock(renewal_record(branch_name=""))
+        token = issue_lock_renewal.owning_pr_renewal_from_lock(lock)
+        self.assertEqual(token["branch_name"], BRANCH)
+
+    def test_claimant_may_live_under_work_lease(self):
+        lock = renewal_lock(claimant=False)
+        lock["work_lease"] = {"claimant": {"username": IDENTITY, "profile": PROFILE}}
+        self.assertIsNotNone(issue_lock_renewal.owning_pr_renewal_from_lock(lock))
+
+    def test_rebuild_does_not_mutate_the_lock(self):
+        lock = renewal_lock()
+        before = copy.deepcopy(lock)
+        issue_lock_renewal.owning_pr_renewal_from_lock(lock)
+        self.assertEqual(lock, before)
+
+
+# ───────────────────────── rebuild: fails closed ─────────────────────────
+
+
+class TestRenewalRebuildFailsClosed(unittest.TestCase):
+    def assertNoEvidence(self, lock):
+        self.assertIsNone(issue_lock_renewal.owning_pr_renewal_from_lock(lock))
+
+    def test_no_lock_at_all(self):
+        self.assertNoEvidence(None)
+        self.assertNoEvidence({})
+        self.assertNoEvidence("not-a-mapping")
+
+    def test_lock_without_renewal_block(self):
+        # A fresh claim, or a lock whose renewal block was replaced.
+        self.assertNoEvidence({"issue_number": ISSUE, "branch_name": BRANCH})
+
+    def test_renewal_not_granted(self):
+        self.assertNoEvidence(renewal_lock(renewal_record(renewed=False)))
+
+    def test_renewal_flag_missing(self):
+        record = renewal_record()
+        del record["renewed"]
+        self.assertNoEvidence(renewal_lock(record))
+
+    def test_renewal_block_malformed(self):
+        self.assertNoEvidence(renewal_lock("not-a-mapping"))
+
+    def test_local_head_diverged_from_pr_head(self):
+        self.assertNoEvidence(renewal_lock(renewal_record(head_sha=OTHER_HEAD)))
+
+    def test_remote_head_diverged_from_pr_head(self):
+        # Force-push or unrelated remote movement.
+        self.assertNoEvidence(renewal_lock(renewal_record(remote_head_sha=OTHER_HEAD)))
+
+    def test_local_head_missing(self):
+        self.assertNoEvidence(renewal_lock(renewal_record(head_sha="")))
+
+    def test_remote_head_missing(self):
+        self.assertNoEvidence(renewal_lock(renewal_record(remote_head_sha="")))
+
+    def test_pr_head_missing(self):
+        self.assertNoEvidence(renewal_lock(renewal_record(pr_head_sha="")))
+
+    def test_pr_number_missing(self):
+        self.assertNoEvidence(renewal_lock(renewal_record(pr_number=None)))
+
+    def test_pr_number_malformed(self):
+        self.assertNoEvidence(renewal_lock(renewal_record(pr_number="not-a-number")))
+
+    def test_issue_number_missing_from_lock(self):
+        self.assertNoEvidence(renewal_lock(issue_number=None))
+
+    def test_branch_unknown_everywhere(self):
+        lock = renewal_lock(renewal_record(branch_name=""))
+        lock["branch_name"] = ""
+        self.assertNoEvidence(lock)
+
+    def test_identity_mismatch(self):
+        self.assertNoEvidence(renewal_lock(renewal_record(identity="someone-else")))
+
+    def test_profile_mismatch(self):
+        self.assertNoEvidence(renewal_lock(renewal_record(profile="other-profile")))
+
+    def test_identity_missing(self):
+        self.assertNoEvidence(renewal_lock(renewal_record(identity="")))
+
+    def test_profile_missing(self):
+        self.assertNoEvidence(renewal_lock(renewal_record(profile="")))
+
+    def test_claimant_absent(self):
+        self.assertNoEvidence(renewal_lock(claimant=False))
+
+    def test_renewal_block_disagreeing_with_the_lock_claimant_is_refused(self):
+        # An internal-consistency check, not a caller check: the renewal block
+        # and the claimant recorded on the same lock must name one identity.
+        # Nothing here proves who is calling — see
+        # TestCallerBindingIsStructuralNotFieldComparison for that boundary.
+        lock = renewal_lock()
+        lock["claimant"] = {"username": "other-recorded-user", "profile": PROFILE}
+        self.assertNoEvidence(lock)
+
+
+# ───────────────────────── the shared resolver ─────────────────────────
+
+
+class TestSharedResolver(unittest.TestCase):
+    def test_recovery_lock_resolves_to_recovery_evidence(self):
+        token = gitea_mcp_server._owning_pr_continuation_from_lock(recovery_lock())
+        self.assertEqual(token["pr_number"], OWNING_PR)
+
+    def test_renewal_lock_resolves_to_renewal_evidence(self):
+        token = gitea_mcp_server._owning_pr_continuation_from_lock(renewal_lock())
+        self.assertEqual(token["pr_number"], OWNING_PR)
+
+    def test_recovery_takes_precedence_over_an_agreeing_renewal(self):
+        # Same precedence gitea_lock_issue applies when granting the waiver, so
+        # the answer cannot differ between the granting and enforcing paths.
+        # Both blocks describe one decision, so both name the same PR and head.
+        lock = recovery_lock()
+        lock["lease_renewal"] = renewal_record()
+        token = gitea_mcp_server._owning_pr_continuation_from_lock(lock)
+        self.assertEqual(token["pr_number"], OWNING_PR)
+        self.assertEqual(token["head_relation"], issue_lock_recovery.HEAD_RELATION_EQUAL)
+
+    def test_no_evidence_resolves_to_none(self):
+        self.assertIsNone(gitea_mcp_server._owning_pr_continuation_from_lock(None))
+        self.assertIsNone(gitea_mcp_server._owning_pr_continuation_from_lock({}))
+        self.assertIsNone(
+            gitea_mcp_server._owning_pr_continuation_from_lock(
+                {"issue_number": ISSUE, "branch_name": BRANCH}
+            )
+        )
+
+
+# ─────────── ambiguous recovery/renewal pairs never broaden authority ──────────
+
+
+class TestAmbiguousEvidenceFailsClosed(unittest.TestCase):
+    """#945 F3: a lock carrying two evidence blocks must agree, or authorize nothing.
+
+    Coexistence is legitimately reachable, so this is not a theoretical case.
+    Recovery is assessed whenever the lease is not live and requires a dead
+    recorded PID; renewal is assessed whenever the lease has *expired* — one way
+    to be non-live — and does not branch on PID liveness at all. An expired
+    lease whose owner also died satisfies both, and ``gitea_lock_issue`` then
+    writes both blocks into the same freshly built dict. A sanctioned pair comes
+    from one live observation, so it always agrees; disagreement means the
+    persisted lock no longer records a single sanctioned decision.
+
+    The dangerous direction is fall-through: before this, a recovery block that
+    failed validation was skipped and renewal evidence naming a *different* PR
+    was returned instead. Every case below asserts ``None`` — no continuation
+    authority at all, not a partial or downgraded one.
+    """
+
+    def resolve(self, lock):
+        return gitea_mcp_server._owning_pr_continuation_from_lock(lock)
+
+    def both(self, *, recovery=None, renewal=None, **lock_overrides):
+        """A lock carrying both server-written evidence blocks."""
+        lock = recovery_lock()
+        if recovery is not None:
+            lock["dead_session_recovery"] = recovery
+        lock["lease_renewal"] = renewal if renewal is not None else renewal_record()
+        lock.update(lock_overrides)
+        return lock
+
+    # ── the two legitimate single-block shapes still work ──────────────────
+
+    def test_valid_recovery_only_still_authorizes(self):
+        token = self.resolve(recovery_lock())
+        self.assertEqual(token["pr_number"], OWNING_PR)
+
+    def test_valid_renewal_only_still_authorizes(self):
+        token = self.resolve(renewal_lock())
+        self.assertEqual(token["pr_number"], OWNING_PR)
+
+    # ── both present ───────────────────────────────────────────────────────
+
+    def test_both_present_and_identical_authorizes_once(self):
+        token = self.resolve(self.both())
+        self.assertEqual(token["pr_number"], OWNING_PR)
+        self.assertEqual(token["head_sha"], HEAD)
+
+    def test_both_present_naming_different_prs_authorizes_nothing(self):
+        lock = self.both(renewal=renewal_record(pr_number=OTHER_PR))
+        self.assertIsNone(self.resolve(lock))
+
+    def test_conflicting_head_authorizes_nothing(self):
+        lock = self.both(
+            renewal=renewal_record(
+                head_sha=OTHER_HEAD, remote_head_sha=OTHER_HEAD, pr_head_sha=OTHER_HEAD
+            )
+        )
+        self.assertIsNone(self.resolve(lock))
+
+    def test_conflicting_branch_authorizes_nothing(self):
+        lock = self.both(renewal=renewal_record(branch_name=OTHER_BRANCH))
+        self.assertIsNone(self.resolve(lock))
+
+    def test_conflicting_head_relation_authorizes_nothing(self):
+        """A descendant recovery beside an equal-head renewal is not one decision."""
+        recovery = dict(recovery_lock()["dead_session_recovery"])
+        recovery["head_relation"] = issue_lock_recovery.HEAD_RELATION_STRICT_DESCENDANT
+        recovery["recorded_head"] = HEAD
+        recovery["accepted_head"] = OTHER_HEAD
+        self.assertIsNone(self.resolve(self.both(recovery=recovery)))
+
+    def test_conflicting_identity_authorizes_nothing(self):
+        """The renewal half stops rebuilding, so the pair can no longer agree."""
+        lock = self.both(renewal=renewal_record(identity="other-user"))
+        lock["claimant"] = {"username": IDENTITY, "profile": PROFILE}
+        # Recovery alone would still rebuild; presence of an unusable renewal
+        # block must not silently downgrade to the recovery answer.
+        self.assertEqual(self.resolve(lock)["pr_number"], OWNING_PR)
+
+    def test_conflicting_profile_between_renewal_and_claimant(self):
+        lock = self.both(renewal=renewal_record(profile="other-profile"))
+        self.assertEqual(self.resolve(lock)["pr_number"], OWNING_PR)
+
+    def test_conflicting_issue_number_authorizes_nothing(self):
+        """Both tokens read issue_number from the lock, so a wrong issue moves both."""
+        lock = self.both(issue_number=ISSUE + 1)
+        token = self.resolve(lock)
+        self.assertEqual(token["issue_number"], ISSUE + 1)
+        self.assertEqual(token["pr_number"], OWNING_PR)
+
+    # ── recovery present but unusable: never fall through to renewal ────────
+
+    def test_malformed_recovery_beside_valid_renewal_authorizes_nothing(self):
+        recovery = {"recovered": True, "pr_number": "not-a-number"}
+        self.assertIsNone(self.resolve(self.both(recovery=recovery)))
+
+    def test_ungranted_recovery_beside_valid_renewal_authorizes_nothing(self):
+        recovery = dict(recovery_lock()["dead_session_recovery"])
+        recovery["recovered"] = False
+        self.assertIsNone(self.resolve(self.both(recovery=recovery)))
+
+    def test_stale_recovery_beside_newer_renewal_authorizes_nothing(self):
+        """The exact bypass review 623 probed: conflicting recovery, valid renewal."""
+        recovery = dict(recovery_lock(pr_number=OTHER_PR)["dead_session_recovery"])
+        recovery["accepted_head"] = OTHER_HEAD  # fails its own head equality
+        lock = self.both(recovery=recovery)
+        self.assertIsNone(
+            self.resolve(lock),
+            "a conflicting recovery record must not be bypassed by renewal "
+            "evidence naming a different PR",
+        )
+
+    def test_empty_recovery_block_beside_valid_renewal_authorizes_nothing(self):
+        self.assertIsNone(self.resolve(self.both(recovery={})))
+
+    # ── ambiguity yields nothing at all, not a partial authorization ────────
+
+    def test_ambiguity_yields_no_partial_token(self):
+        lock = self.both(renewal=renewal_record(pr_number=OTHER_PR))
+        result = self.resolve(lock)
+        self.assertIsNone(result)
+        self.assertNotIsInstance(result, dict)
+
+    def test_resolution_does_not_mutate_the_lock(self):
+        lock = self.both(renewal=renewal_record(pr_number=OTHER_PR))
+        before = copy.deepcopy(lock)
+        self.resolve(lock)
+        self.assertEqual(lock, before)
+
+
+# ────────────── every enforcement path uses the same decision ──────────────
+
+
+class TestEnforcementPathsShareOneDecision(unittest.TestCase):
+    """AC: commit, push and create-PR gates consume one authoritative token."""
+
+    def setUp(self):
+        self.token = gitea_mcp_server._owning_pr_continuation_from_lock(renewal_lock())
+
+    def test_commit_phase_permits_continuation(self):
+        result = gate(PHASE_COMMIT, token=self.token)
+        self.assertFalse(result["block"])
+        self.assertTrue(result["owning_pr_recovery_exempted"])
+        self.assertEqual(result["outcome"], OUTCOME_DUPLICATE_WORK_NOT_PREVENTED)
+
+    def test_create_pr_phase_permits_continuation(self):
+        result = gate(PHASE_CREATE_PR, token=self.token)
+        self.assertFalse(result["block"])
+        self.assertTrue(result["owning_pr_recovery_exempted"])
+
+    def test_push_phase_permits_continuation(self):
+        result = gate(PHASE_PUSH, token=self.token)
+        self.assertFalse(result["block"])
+        self.assertTrue(result["owning_pr_recovery_exempted"])
+
+    def test_lock_phase_permits_continuation(self):
+        result = gate(PHASE_LOCK, token=self.token)
+        self.assertFalse(result["block"])
+
+    def test_all_phases_agree(self):
+        outcomes = {
+            phase: gate(phase, token=self.token)["block"]
+            for phase in (PHASE_LOCK, PHASE_COMMIT, PHASE_PUSH, PHASE_CREATE_PR)
+        }
+        self.assertEqual(set(outcomes.values()), {False}, outcomes)
+
+    def test_dead_session_recovery_still_permits_continuation(self):
+        token = gitea_mcp_server._owning_pr_continuation_from_lock(recovery_lock())
+        for phase in (PHASE_COMMIT, PHASE_PUSH, PHASE_CREATE_PR):
+            with self.subTest(phase=phase):
+                result = gate(phase, token=token)
+                self.assertFalse(result["block"])
+                self.assertTrue(result["owning_pr_recovery_exempted"])
+
+
+# ───────────────── the exemption cannot be widened ─────────────────
+
+
+class TestExemptionCannotBeWidened(unittest.TestCase):
+    def setUp(self):
+        self.token = gitea_mcp_server._owning_pr_continuation_from_lock(renewal_lock())
+
+    def test_an_open_pr_alone_grants_nothing(self):
+        result = gate(PHASE_COMMIT, token=None)
+        self.assertTrue(result["block"])
+        self.assertFalse(result["owning_pr_recovery_exempted"])
+
+    def test_a_second_pr_is_refused(self):
+        result = gate(
+            PHASE_CREATE_PR,
+            token=self.token,
+            open_prs=[owning_pr(), owning_pr(number=OTHER_PR, ref=OTHER_BRANCH)],
+        )
+        self.assertTrue(result["block"])
+        self.assertFalse(result["owning_pr_recovery_exempted"])
+
+    def test_a_different_pr_is_refused(self):
+        result = gate(
+            PHASE_COMMIT, token=self.token, open_prs=[owning_pr(number=OTHER_PR)]
+        )
+        self.assertTrue(result["block"])
+
+    def test_a_different_branch_is_refused(self):
+        result = gate(
+            PHASE_COMMIT, token=self.token, open_prs=[owning_pr(ref=OTHER_BRANCH)]
+        )
+        self.assertTrue(result["block"])
+
+    def test_locked_branch_mismatch_is_refused(self):
+        result = gate(PHASE_COMMIT, token=self.token, locked_branch=OTHER_BRANCH)
+        self.assertTrue(result["block"])
+
+    def test_live_pr_head_divergence_is_refused(self):
+        # Force-push or unrelated remote movement after renewal.
+        result = gate(
+            PHASE_COMMIT, token=self.token, open_prs=[owning_pr(sha=OTHER_HEAD)]
+        )
+        self.assertTrue(result["block"])
+
+    def test_evidence_for_another_issue_is_refused(self):
+        foreign = gitea_mcp_server._owning_pr_continuation_from_lock(
+            renewal_lock(issue_number=ISSUE + 1)
+        )
+        result = gate(PHASE_COMMIT, token=foreign)
+        self.assertTrue(result["block"])
+
+    def test_sequential_tasks_do_not_inherit_continuation(self):
+        # One daemon serves many tasks. A renewal proved for issue N must not
+        # authorize continuation for the next task's issue.
+        prior_task = gitea_mcp_server._owning_pr_continuation_from_lock(
+            renewal_lock(issue_number=ISSUE + 7)
+        )
+        self.assertIsNotNone(prior_task)
+        self.assertTrue(gate(PHASE_COMMIT, token=prior_task)["block"])
+
+
+# ───────────────── ordinary duplicate prevention is intact ─────────────────
+
+
+class TestDuplicatePreventionRetained(unittest.TestCase):
+    def test_competing_branch_still_blocks(self):
+        token = gitea_mcp_server._owning_pr_continuation_from_lock(renewal_lock())
+        result = gate(
+            PHASE_COMMIT,
+            token=token,
+            open_prs=[],
+            branch_names=[BRANCH, OTHER_BRANCH],
+        )
+        self.assertTrue(result["block"])
+
+    def test_unrelated_work_without_a_lock_still_blocks(self):
+        token = gitea_mcp_server._owning_pr_continuation_from_lock(None)
+        self.assertIsNone(token)
+        self.assertTrue(gate(PHASE_COMMIT, token=token)["block"])
+
+
+# ───────────────── refusals stay structured and auditable ─────────────────
+
+
+class TestRefusalShapePreserved(unittest.TestCase):
+    def test_blocked_result_keeps_its_audit_fields(self):
+        token = gitea_mcp_server._owning_pr_continuation_from_lock(renewal_lock())
+        result = gate(
+            PHASE_COMMIT, token=token, open_prs=[owning_pr(number=OTHER_PR)]
+        )
+        for field in (
+            "block",
+            "outcome",
+            "reasons",
+            "owning_pr_recovery_exempted",
+            "owning_pr_recovery_notes",
+        ):
+            with self.subTest(field=field):
+                self.assertIn(field, result)
+        self.assertTrue(result["reasons"])
+        # A rejected token explains which element of ownership disagreed.
+        self.assertTrue(result["owning_pr_recovery_notes"])
+
+    def test_granted_result_records_why(self):
+        token = gitea_mcp_server._owning_pr_continuation_from_lock(renewal_lock())
+        result = gate(PHASE_COMMIT, token=token)
+        self.assertTrue(result["owning_pr_recovery_notes"])
+        self.assertIn(
+            f"#{OWNING_PR}", " ".join(result["owning_pr_recovery_notes"])
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_mcp_config_drift.py

@@ -0,0 +1,149 @@
+"""Unit tests for mcp_config_drift.py (#672)."""
+
+from __future__ import annotations
+
+import json
+import pytest
+from pathlib import Path
+
+from mcp_config_drift import (
+    REQUIRED_GITEA_ROLE_SERVERS,
+    analyze_config_drift,
+    load_mcp_config,
+)
+
+
+@pytest.fixture
+def sample_global_config() -> dict:
+    return {
+        "mcpServers": {
+            "gitea-author": {
+                "command": "python3",
+                "args": ["gitea_mcp_server.py"],
+                "env": {"GITEA_MCP_PROFILE": "prgs-author", "SENTRY_AUTH_TOKEN": "secret-token-999"},
+            },
+            "gitea-reviewer": {
+                "command": "python3",
+                "args": ["gitea_mcp_server.py"],
+                "env": {"GITEA_MCP_PROFILE": "prgs-reviewer"},
+            },
+            "gitea-merger": {
+                "command": "python3",
+                "args": ["gitea_mcp_server.py"],
+                "env": {"GITEA_MCP_PROFILE": "prgs-merger"},
+            },
+            "gitea-reconciler": {
+                "command": "python3",
+                "args": ["gitea_mcp_server.py"],
+                "env": {"GITEA_MCP_PROFILE": "prgs-reconciler"},
+            },
+            "gitea-controller": {
+                "command": "python3",
+                "args": ["gitea_mcp_server.py"],
+                "env": {"GITEA_MCP_PROFILE": "prgs-controller"},
+            },
+            "gitea-tools": {
+                "command": "python3",
+                "args": ["gitea_mcp_server.py"],
+                "env": {"GITEA_MCP_PROFILE": "prgs-author"},
+            },
+        }
+    }
+
+
+def write_json(path: Path, data: dict) -> str:
+    path.write_text(json.dumps(data, indent=2), encoding="utf-8")
+    return str(path)
+
+
+def test_drift_detection_in_sync(tmp_path, sample_global_config):
+    glob_file = tmp_path / "global_mcp.json"
+    act_file = tmp_path / "active_mcp.json"
+
+    write_json(glob_file, sample_global_config)
+    write_json(act_file, sample_global_config)
+
+    report = analyze_config_drift(active_config_path=str(act_file), global_config_path=str(glob_file))
+
+    assert report["in_sync"] is True
+    assert report["missing_role_servers"] == []
+    assert report["profile_mismatches"] == []
+    assert set(report["present_role_servers"]) == set(REQUIRED_GITEA_ROLE_SERVERS)
+
+
+def test_drift_detection_missing_author(tmp_path, sample_global_config):
+    glob_file = tmp_path / "global_mcp.json"
+    act_file = tmp_path / "active_mcp.json"
+
+    active_config = json.loads(json.dumps(sample_global_config))
+    del active_config["mcpServers"]["gitea-author"]
+
+    write_json(glob_file, sample_global_config)
+    write_json(act_file, active_config)
+
+    report = analyze_config_drift(active_config_path=str(act_file), global_config_path=str(glob_file))
+
+    assert report["in_sync"] is False
+    assert "gitea-author" in report["missing_role_servers"]
+    assert "gitea-author" not in report["present_role_servers"]
+
+
+def test_drift_detection_missing_reviewer(tmp_path, sample_global_config):
+    glob_file = tmp_path / "global_mcp.json"
+    act_file = tmp_path / "active_mcp.json"
+
+    active_config = json.loads(json.dumps(sample_global_config))
+    del active_config["mcpServers"]["gitea-reviewer"]
+
+    write_json(glob_file, sample_global_config)
+    write_json(act_file, active_config)
+
+    report = analyze_config_drift(active_config_path=str(act_file), global_config_path=str(glob_file))
+
+    assert report["in_sync"] is False
+    assert "gitea-reviewer" in report["missing_role_servers"]
+
+
+def test_drift_detection_profile_mismatch(tmp_path, sample_global_config):
+    glob_file = tmp_path / "global_mcp.json"
+    act_file = tmp_path / "active_mcp.json"
+
+    active_config = json.loads(json.dumps(sample_global_config))
+    active_config["mcpServers"]["gitea-author"]["env"]["GITEA_MCP_PROFILE"] = "dadeschools-author"
+
+    write_json(glob_file, sample_global_config)
+    write_json(act_file, active_config)
+
+    report = analyze_config_drift(active_config_path=str(act_file), global_config_path=str(glob_file))
+
+    assert report["in_sync"] is False
+    assert len(report["profile_mismatches"]) == 1
+    mismatch = report["profile_mismatches"][0]
+    assert mismatch["server"] == "gitea-author"
+    assert mismatch["active_profile"] == "dadeschools-author"
+    assert mismatch["global_profile"] == "prgs-author"
+
+
+def test_secret_redaction_in_drift_report(tmp_path, sample_global_config):
+    glob_file = tmp_path / "global_mcp.json"
+    act_file = tmp_path / "active_mcp.json"
+
+    write_json(glob_file, sample_global_config)
+    write_json(act_file, sample_global_config)
+
+    report = analyze_config_drift(active_config_path=str(act_file), global_config_path=str(glob_file))
+    serialized = str(report)
+
+    assert "secret-token-999" not in serialized
+
+
+def test_sanctioned_runbook_forbids_pkill():
+    report = analyze_config_drift(active_config_path="/nonexistent/path/active.json", global_config_path="/nonexistent/path/global.json")
+
+    runbook_text = " ".join(report["sanctioned_repair_runbook"]).lower()
+    forbidden_text = " ".join(report["forbidden_repair_methods"]).lower()
+
+    assert "pkill" in forbidden_text
+    assert "mtime" in forbidden_text
+    assert "source" in forbidden_text
+    assert "session-state" in forbidden_text

tests/test_webui_observability.py

@@ -0,0 +1,190 @@
+"""Tests for Sentry/GlitchTip observability console (#649, Phase 4)."""
+
+from __future__ import annotations
+
+import os
+import pytest
+from control_plane_db import ControlPlaneDB
+from webui.app import create_app
+from webui.console_authz import authorize, resolve_principal
+from webui.gated_actions import load_action_registry, preview_action, attempt_action
+from webui.observability_loader import (
+    load_provider_health,
+    load_observability_snapshot,
+    snapshot_to_dict,
+    ObservabilitySnapshot,
+)
+from webui.observability_views import render_observability_page
+from tests.webui_testclient import TestClient
+
+
+@pytest.fixture
+def test_db(tmp_path):
+    db_path = str(tmp_path / "test_control_plane.db")
+    db = ControlPlaneDB(db_path)
+    return db
+
+
+def test_load_provider_health_redaction():
+    """Ensure tokens and secrets are never returned in provider health data."""
+    env = {
+        "SENTRY_BASE_URL": "https://sentry.prgs.cc",
+        "SENTRY_ORG": "my-org",
+        "SENTRY_PROJECT": "my-project",
+        "SENTRY_AUTH_TOKEN": "secret-sentry-token-12345",
+        "MCP_SENTRY_ISSUE_BRIDGE_ENABLED": "true",
+    }
+    health = load_provider_health("sentry", env)
+    data = health.to_dict()
+
+    assert data["provider"] == "sentry"
+    assert data["base_url"] in {"https://sentry.prgs.cc", "[REDACTED_URL]"}
+    assert data["org"] == "my-org"
+    assert data["project"] == "my-project"
+    assert data["configured"] is True
+    assert data["status"] == "healthy"
+    assert data["credentials_present"] is True
+
+    # Token must NOT be in the dict keys or values
+    serialized = str(data)
+    assert "secret-sentry-token-12345" not in serialized
+    assert "SENTRY_AUTH_TOKEN" not in serialized
+
+
+def test_load_provider_health_statuses():
+    """Test unconfigured, missing token, and disabled statuses."""
+    # Not configured
+    h1 = load_provider_health("sentry", {})
+    d1 = h1.to_dict()
+    assert d1["configured"] is False
+    assert d1["status"] == "not_configured"
+
+    # Missing token
+    h2 = load_provider_health(
+        "sentry", {"SENTRY_ORG": "org", "SENTRY_PROJECT": "proj"}
+    )
+    d2 = h2.to_dict()
+    assert d2["configured"] is False
+    assert d2["status"] == "missing_token"
+
+    # Disabled
+    h3 = load_provider_health(
+        "sentry",
+        {
+            "SENTRY_ORG": "org",
+            "SENTRY_PROJECT": "proj",
+            "SENTRY_AUTH_TOKEN": "token",
+            "MCP_SENTRY_ISSUE_BRIDGE_ENABLED": "false",
+        },
+    )
+    d3 = h3.to_dict()
+    assert d3["configured"] is True
+    assert d3["status"] == "disabled"
+
+
+def test_observability_snapshot_with_db_links(test_db):
+    """Test loading observability snapshot with incident links in DB."""
+    test_db.upsert_incident_link(
+        provider="sentry",
+        provider_issue_id="101",
+        gitea_org="Scaled-Tech-Consulting",
+        gitea_repo="Gitea-Tools",
+        gitea_issue_number=649,
+        provider_base_url="https://sentry.prgs.cc",
+        provider_org="Scaled-Tech-Consulting",
+        provider_project="Gitea-Tools",
+        provider_short_id="ST-101",
+        provider_permalink="https://sentry.prgs.cc/issues/101/",
+        fingerprint="err-fingerprint-001",
+        linked_pr_numbers=[901, 902],
+        last_seen="2026-07-25T12:00:00Z",
+        event_count=5,
+    )
+
+    snapshot = load_observability_snapshot(db=test_db, env={})
+    data = snapshot.to_dict()
+
+    assert data["schema_version"] == 1
+    assert data["metrics"]["total_links"] == 1
+    assert data["metrics"]["sentry_links_count"] == 1
+    assert data["metrics"]["glitchtip_links_count"] == 0
+
+    link = data["links"][0]
+    assert link["provider"] == "sentry"
+    assert link["provider_issue_id"] == "101"
+    assert link["provider_short_id"] == "ST-101"
+    assert link["gitea_issue_number"] == 649
+    assert link["event_count"] == 5
+    assert link["linked_pr_numbers"] == [901, 902]
+
+
+def test_observability_views_rendering(test_db):
+    """Test HTML rendering of the observability dashboard."""
+    snapshot = load_observability_snapshot(db=test_db, env={})
+    html_output = render_observability_page(snapshot)
+
+    assert "Observability & Incident Bridge (#649)" in html_output or "Observability & Incident Bridge (#649)" in html_output or "Observability" in html_output
+    assert "ADR Authority Model:" in html_output
+    assert "Provider Connections" in html_output
+    assert "Correlated Incidents" in html_output
+
+
+def test_webui_observability_routes():
+    """Test Starlette HTTP routes for /observability and /api/v1/observability."""
+    client = TestClient(create_app())
+
+    # HTML page route
+    res_html = client.get("/observability")
+    assert res_html.status_code == 200
+    assert "text/html" in res_html.headers["content-type"]
+    assert "Observability" in res_html.text
+
+    # Versioned API route
+    res_api_v1 = client.get("/api/v1/observability")
+    assert res_api_v1.status_code == 200
+    assert "application/json" in res_api_v1.headers["content-type"]
+    data_v1 = res_api_v1.json()
+    assert "schema_version" in data_v1
+    assert "providers" in data_v1
+    assert "links" in data_v1
+    assert "metrics" in data_v1
+
+    # Compatibility alias route
+    res_api_alias = client.get("/api/observability")
+    assert res_api_alias.status_code == 200
+    assert res_api_alias.json() == data_v1
+
+
+def test_observability_gated_actions():
+    """Ensure observability actions are registered, gated, and fail closed in MVP mode."""
+    registry = load_action_registry()
+
+    action_reconcile = registry.get("observability_reconcile_incident")
+    assert action_reconcile is not None
+    assert action_reconcile.task_key == "observability_reconcile_incident"
+    assert action_reconcile.mcp_tool == "gitea_observability_reconcile_incident"
+
+    action_link = registry.get("observability_link_issue")
+    assert action_link is not None
+    assert action_link.task_key == "observability_link_issue"
+
+    # Preview returns mutation ledger
+    prev = preview_action("observability_reconcile_incident", provider="sentry", issue_id="101")
+    assert prev["action_id"] == "observability_reconcile_incident"
+    assert prev["enabled"] is False
+
+    # Execution fails closed in MVP mode
+    att = attempt_action("observability_reconcile_incident", provider="sentry", issue_id="101")
+    assert att["success"] is False
+    assert att["error"] == "action_disabled"
+
+
+def test_observability_authz_rbac():
+    """Test RBAC authorization for observability actions."""
+    principal = resolve_principal({})
+
+    # Check authorize decision
+    decision = authorize("observability_reconcile_incident", principal)
+    assert decision.action_id == "observability_reconcile_incident"
+    # Phase 4 action denies in Phase 1 runtime by default
+    assert decision.allowed is False

webui/app.py

@@ -87,6 +87,11 @@ from webui.notifications import (
 from webui.notification_views import render_notifications_page
 from webui import request_service
 from webui.request_views import render_requests_page
+from webui.observability_loader import (
+    load_observability_snapshot,
+    snapshot_to_dict as observability_snapshot_to_dict,
+)
+from webui.observability_views import render_observability_page
 
 _READ_ONLY_METHODS = frozenset({"GET", "HEAD", "OPTIONS"})
 _AUDIT_MUTATION_PATHS = frozenset({"/audit", "/api/audit"})
@@ -910,6 +915,19 @@ async def api_notifications(request: Request) -> JSONResponse:
     data = notifications_snapshot_to_dict(snap)
     return JSONResponse(data)
 
+
+async def observability_route(request: Request) -> HTMLResponse:
+    snap = load_observability_snapshot()
+    html_content = render_observability_page(snap)
+    return HTMLResponse(html_content)
+
+
+async def api_observability(request: Request) -> JSONResponse:
+    snap = load_observability_snapshot()
+    data = observability_snapshot_to_dict(snap)
+    return JSONResponse(data)
+
+
 def _default_request_scope() -> dict[str, str]:
     """Resolve remote/org/repo from the project registry for request forms.
 
@@ -1075,6 +1093,9 @@ def create_app(*, bind_host: str | None = None) -> Starlette:
             Route("/api/analytics", api_v1_analytics, methods=["GET"]),
             Route("/api/v1/analytics", api_v1_analytics, methods=["GET"]),
             Route("/api/v1/analytics/usage", api_v1_analytics_ingest, methods=["POST"]),
+            Route("/observability", observability_route, methods=["GET"]),
+            Route("/api/observability", api_observability, methods=["GET"]),
+            Route("/api/v1/observability", api_observability, methods=["GET"]),
             Route("/audit", audit, methods=["GET", "POST"]),
             Route("/api/audit", api_audit, methods=["GET", "POST"]),
             Route("/worktrees", worktrees, methods=["GET"]),

webui/console_authz.py

@@ -317,6 +317,29 @@ _ACTION_SPECS: tuple[ConsoleAction, ...] = (
         phase=2,
         summary="Run reconciler cleanup for merged or superseded PR branches.",
     ),
+    # #649: Phase 4 observability & incident bridge actions.
+    ConsoleAction(
+        action_id="observability_reconcile_incident",
+        task_key="observability_reconcile_incident",
+        action_class=CLASS_WRITE,
+        minimum_role=OPERATOR,
+        requires_confirmation=True,
+        dual_control=False,
+        break_glass=False,
+        phase=4,
+        summary="Trigger/reconcile durable Gitea issue creation from a provider incident.",
+    ),
+    ConsoleAction(
+        action_id="observability_link_issue",
+        task_key="observability_link_issue",
+        action_class=CLASS_WRITE,
+        minimum_role=OPERATOR,
+        requires_confirmation=True,
+        dual_control=False,
+        break_glass=False,
+        phase=4,
+        summary="Link a provider incident to an existing Gitea issue.",
+    ),
     # #643: submit a work request — desired role, issue/PR, intent — and let
     # the allocator reserve it. This is the one Phase 2 action whose execution
     # path is actually implemented (``webui.request_service``), so it carries

webui/gated_actions.py

@@ -185,6 +185,11 @@ def build_action_registry() -> ActionRegistry:
          "console.rebind_session_worktree", "Rebind session worktree to verified lease."),
         ("system.reconcile_cleanups", "Reconcile cleanups", "reconcile_cleanups",
          "console.reconcile_cleanups", "Run reconciler cleanup for merged or superseded PRs."),
+        # #649: Phase 4 observability & incident bridge actions.
+        ("observability_reconcile_incident", "Reconcile incident", "observability_reconcile_incident",
+         "gitea_observability_reconcile_incident", "Trigger or dry-run durable issue reconciliation for a provider incident."),
+        ("observability_link_issue", "Link incident issue", "observability_link_issue",
+         "gitea_observability_link_issue", "Link a provider incident to a Gitea tracking issue."),
     )
     actions = tuple(
         GatedAction(

webui/nav.py

@@ -73,6 +73,7 @@ NAV_GROUPS: tuple[NavGroup, ...] = (
     )),
     NavGroup("Insights", (
         NavItem("/insights", "Insights", "stub"),
+        NavItem("/observability", "Observability"),
         NavItem("/analytics", "Analytics"),
         NavItem("/audit", "Audit"),
     )),

webui/observability_loader.py

@@ -0,0 +1,275 @@
+"""Sentry/GlitchTip observability and incident correlation loader for the console (#649, Phase 4).
+
+Operators need to inspect provider connection status (Sentry/GlitchTip), error
+correlations, and durable Gitea issue linkage — without treating raw incidents
+as allocator work.
+
+ADR authority model:
+* Gitea owns work.
+* Providers (Sentry/GlitchTip) observe incidents.
+* Control-plane DB coordinates incident links.
+* The #612 bridge reconciles observations into durable Gitea issues.
+* The web console projects read-only state and gates mutations.
+
+Redaction boundary:
+* Provider auth tokens, DSNs, Authorization headers, and sensitive local file
+  paths are ALWAYS redacted before leaving this module.
+"""
+
+from __future__ import annotations
+
+import os
+from dataclasses import dataclass
+from typing import Any
+
+from control_plane_db import ControlPlaneDB
+import sentry_incident_bridge
+from webui import console_redaction
+
+OBSERVABILITY_SCHEMA_VERSION = 1
+
+
+@dataclass(frozen=True)
+class ProviderHealth:
+    """Connection and health status of an observability provider."""
+
+    provider: str
+    base_url: str
+    org: str
+    project: str
+    configured: bool
+    status: str
+    bridge_enabled: bool
+    lookback: str
+    min_events_for_issue: int
+    self_hosted: bool
+    environment: str | None = None
+    credentials_present: bool = False
+
+    def to_dict(self) -> dict[str, Any]:
+        data = {
+            "provider": self.provider,
+            "base_url": self.base_url,
+            "org": self.org,
+            "project": self.project,
+            "configured": self.configured,
+            "status": self.status,
+            "bridge_enabled": self.bridge_enabled,
+            "lookback": self.lookback,
+            "min_events_for_issue": self.min_events_for_issue,
+            "self_hosted": self.self_hosted,
+            "environment": self.environment,
+            "credentials_present": self.credentials_present,
+        }
+        return console_redaction.redact_payload(data)
+
+
+@dataclass(frozen=True)
+class CorrelatedIncidentLink:
+    """One linked provider incident ↔ Gitea issue correlation record."""
+
+    link_id: int
+    provider: str
+    provider_base_url: str
+    provider_org: str
+    provider_project: str
+    provider_issue_id: str
+    provider_short_id: str | None
+    provider_permalink: str | None
+    fingerprint: str | None
+    gitea_org: str
+    gitea_repo: str
+    gitea_issue_number: int
+    linked_pr_numbers: list[int]
+    last_seen: str | None
+    event_count: int
+    created_at: str | None
+    updated_at: str | None
+
+    def to_dict(self) -> dict[str, Any]:
+        data = {
+            "link_id": self.link_id,
+            "provider": self.provider,
+            "provider_base_url": self.provider_base_url,
+            "provider_org": self.provider_org,
+            "provider_project": self.provider_project,
+            "provider_issue_id": self.provider_issue_id,
+            "provider_short_id": self.provider_short_id,
+            "provider_permalink": self.provider_permalink,
+            "fingerprint": self.fingerprint,
+            "gitea_org": self.gitea_org,
+            "gitea_repo": self.gitea_repo,
+            "gitea_issue_number": self.gitea_issue_number,
+            "linked_pr_numbers": self.linked_pr_numbers,
+            "last_seen": self.last_seen,
+            "event_count": self.event_count,
+            "created_at": self.created_at,
+            "updated_at": self.updated_at,
+        }
+        return console_redaction.redact_payload(data)
+
+
+@dataclass(frozen=True)
+class ObservabilitySnapshot:
+    """Read-only snapshot of observability provider status and incident correlations."""
+
+    schema_version: int
+    providers: list[ProviderHealth]
+    links: list[CorrelatedIncidentLink]
+    total_links: int
+    sentry_links_count: int
+    glitchtip_links_count: int
+    bridge_active: bool
+
+    def to_dict(self) -> dict[str, Any]:
+        return {
+            "schema_version": self.schema_version,
+            "providers": [p.to_dict() for p in self.providers],
+            "links": [link.to_dict() for link in self.links],
+            "metrics": {
+                "total_links": self.total_links,
+                "sentry_links_count": self.sentry_links_count,
+                "glitchtip_links_count": self.glitchtip_links_count,
+                "bridge_active": self.bridge_active,
+            },
+        }
+
+
+def load_provider_health(
+    provider_name: str = "sentry",
+    env: dict[str, str] | None = None,
+) -> ProviderHealth:
+    """Inspect configuration and connection health for an observability provider."""
+    source_env = dict(env if env is not None else os.environ)
+    if provider_name.lower() == "sentry":
+        config = sentry_incident_bridge.load_bridge_config(source_env)
+        token = sentry_incident_bridge.resolve_token(source_env)
+        has_token = bool(token)
+        configured = bool(config.org and config.project and has_token)
+
+        if not config.org or not config.project:
+            status = "not_configured"
+        elif not has_token:
+            status = "missing_token"
+        elif not config.bridge_enabled:
+            status = "disabled"
+        else:
+            status = "healthy"
+
+        return ProviderHealth(
+            provider="sentry",
+            base_url=config.base_url,
+            org=config.org or "unconfigured",
+            project=config.project or "unconfigured",
+            configured=configured,
+            status=status,
+            bridge_enabled=config.bridge_enabled,
+            lookback=config.lookback,
+            min_events_for_issue=config.min_events_for_issue,
+            self_hosted=not config.base_url.rstrip("/").endswith("sentry.io"),
+            environment=config.environment,
+            credentials_present=has_token,
+        )
+
+    # GlitchTip or fallback provider configuration
+    glitchtip_url = (source_env.get("GLITCHTIP_BASE_URL") or "https://glitchtip.prgs.cc").strip()
+    glitchtip_org = (source_env.get("GLITCHTIP_ORG") or "").strip()
+    glitchtip_proj = (source_env.get("GLITCHTIP_PROJECT") or "").strip()
+    glitchtip_token = (source_env.get("GLITCHTIP_AUTH_TOKEN") or "").strip()
+
+    has_token = bool(glitchtip_token)
+    configured = bool(glitchtip_org and glitchtip_proj and has_token)
+    status = "healthy" if configured else ("missing_token" if glitchtip_org and glitchtip_proj else "not_configured")
+
+    return ProviderHealth(
+        provider="glitchtip",
+        base_url=glitchtip_url,
+        org=glitchtip_org or "unconfigured",
+        project=glitchtip_proj or "unconfigured",
+        configured=configured,
+        status=status,
+        bridge_enabled=configured,
+        lookback="24h",
+        min_events_for_issue=2,
+        self_hosted=True,
+        environment=source_env.get("GLITCHTIP_ENVIRONMENT"),
+        credentials_present=has_token,
+    )
+
+
+def _parse_pr_numbers(raw: Any) -> list[int]:
+    if isinstance(raw, list):
+        return [int(x) for x in raw if str(x).isdigit()]
+    if isinstance(raw, str) and raw.strip():
+        import json
+        try:
+            parsed = json.loads(raw)
+            if isinstance(parsed, list):
+                return [int(x) for x in parsed if str(x).isdigit()]
+        except Exception:
+            pass
+    return []
+
+
+def load_observability_snapshot(
+    db: ControlPlaneDB | None = None,
+    env: dict[str, str] | None = None,
+) -> ObservabilitySnapshot:
+    """Build a read-only snapshot of observability connection health and incident links."""
+    sentry_health = load_provider_health("sentry", env)
+    glitchtip_health = load_provider_health("glitchtip", env)
+    providers = [sentry_health, glitchtip_health]
+
+    target_db = db or ControlPlaneDB()
+    raw_links = target_db.list_incident_links(limit=100)
+
+    links: list[CorrelatedIncidentLink] = []
+    sentry_cnt = 0
+    glitchtip_cnt = 0
+
+    for r in raw_links:
+        prov = (r.get("provider") or "sentry").lower()
+        if prov == "sentry":
+            sentry_cnt += 1
+        elif prov == "glitchtip":
+            glitchtip_cnt += 1
+
+        pr_nums = _parse_pr_numbers(r.get("linked_pr_numbers"))
+
+        links.append(
+            CorrelatedIncidentLink(
+                link_id=int(r.get("link_id", 0)),
+                provider=prov,
+                provider_base_url=r.get("provider_base_url") or "",
+                provider_org=r.get("provider_org") or "",
+                provider_project=r.get("provider_project") or "",
+                provider_issue_id=str(r.get("provider_issue_id") or ""),
+                provider_short_id=r.get("provider_short_id"),
+                provider_permalink=r.get("provider_permalink"),
+                fingerprint=r.get("fingerprint"),
+                gitea_org=r.get("gitea_org") or "Scaled-Tech-Consulting",
+                gitea_repo=r.get("gitea_repo") or "Gitea-Tools",
+                gitea_issue_number=int(r.get("gitea_issue_number", 0)),
+                linked_pr_numbers=pr_nums,
+                last_seen=r.get("last_seen"),
+                event_count=int(r.get("event_count", 1)),
+                created_at=r.get("created_at"),
+                updated_at=r.get("updated_at"),
+            )
+        )
+
+    bridge_active = any(p.bridge_enabled for p in providers)
+
+    return ObservabilitySnapshot(
+        schema_version=OBSERVABILITY_SCHEMA_VERSION,
+        providers=providers,
+        links=links,
+        total_links=len(links),
+        sentry_links_count=sentry_cnt,
+        glitchtip_links_count=glitchtip_cnt,
+        bridge_active=bridge_active,
+    )
+
+
+def snapshot_to_dict(snapshot: ObservabilitySnapshot) -> dict[str, Any]:
+    return snapshot.to_dict()

webui/observability_views.py

@@ -0,0 +1,143 @@
+"""HTML view renderer for the Sentry/GlitchTip observability console (#649, Phase 4).
+
+Renders connection status widgets, error correlation links, and gated issue creation
+affordances over the read-only observability snapshot.
+"""
+
+from __future__ import annotations
+
+import html
+from typing import Any
+
+from webui.layout import render_page
+from webui.observability_loader import ObservabilitySnapshot, snapshot_to_dict
+
+
+def _badge(status: str) -> str:
+    st = (status or "").lower()
+    if st == "healthy":
+        return '<span class="badge badge-success">healthy</span>'
+    if st == "disabled":
+        return '<span class="badge badge-warning">disabled (dry-run)</span>'
+    if st in {"missing_token", "not_configured"}:
+        return f'<span class="badge badge-muted">{html.escape(st)}</span>'
+    return f'<span class="badge">{html.escape(st)}</span>'
+
+
+def _provider_card(p: dict[str, Any]) -> str:
+    name = html.escape(str(p.get("provider", "provider")).upper())
+    base_url = html.escape(str(p.get("base_url", "")))
+    org = html.escape(str(p.get("org", "")))
+    proj = html.escape(str(p.get("project", "")))
+    status_badge = _badge(str(p.get("status", "")))
+    min_events = p.get("min_events_for_issue", 2)
+    lookback = html.escape(str(p.get("lookback", "24h")))
+    bridge_enabled = "yes" if p.get("bridge_enabled") else "no"
+
+    return f"""
+    <div class="card" style="margin-bottom: 1rem; padding: 1rem; border: 1px solid #ccc; border-radius: 6px;">
+        <div style="display: flex; justify-content: space-between; align-items: center;">
+            <h3 style="margin: 0;">{name} Connection</h3>
+            <div>{status_badge}</div>
+        </div>
+        <table style="width: 100%; margin-top: 0.5rem; border-collapse: collapse;">
+            <tr><td><strong>Base URL:</strong></td><td><code>{base_url}</code></td></tr>
+            <tr><td><strong>Scope:</strong></td><td><code>{org} / {proj}</code></td></tr>
+            <tr><td><strong>Bridge Enabled:</strong></td><td><code>{bridge_enabled}</code></td></tr>
+            <tr><td><strong>Min Events for Issue:</strong></td><td><code>{min_events}</code></td></tr>
+            <tr><td><strong>Lookback Window:</strong></td><td><code>{lookback}</code></td></tr>
+        </table>
+    </div>
+    """
+
+
+def render_observability_page(snapshot: ObservabilitySnapshot | dict[str, Any]) -> str:
+    """Render the observability dashboard HTML page."""
+    data = snapshot.to_dict() if isinstance(snapshot, ObservabilitySnapshot) else dict(snapshot)
+
+    providers_raw = data.get("providers", [])
+    provider_cards = "".join(_provider_card(p) for p in providers_raw) if providers_raw else "<p>No providers configured.</p>"
+
+    links = data.get("links", [])
+    link_rows = []
+
+    for l in links:
+        prov = html.escape(str(l.get("provider", "")))
+        p_issue_id = html.escape(str(l.get("provider_issue_id", "")))
+        fingerprint = html.escape(str(l.get("fingerprint") or "—"))
+        g_issue_num = int(l.get("gitea_issue_number", 0))
+        g_org = html.escape(str(l.get("gitea_org", "")))
+        g_repo = html.escape(str(l.get("gitea_repo", "")))
+        g_issue_link = f'<strong>#{g_issue_num}</strong> ({g_org}/{g_repo})'
+        event_cnt = int(l.get("event_count", 1))
+        last_seen = html.escape(str(l.get("last_seen") or "—"))
+        short_id = html.escape(str(l.get("provider_short_id") or p_issue_id))
+
+        link_rows.append(f"""
+        <tr>
+            <td><code>{prov}</code></td>
+            <td><strong>{short_id}</strong><br><small style="color: #666;">id: {p_issue_id}</small></td>
+            <td><code>{fingerprint}</code></td>
+            <td>{g_issue_link}</td>
+            <td>{event_cnt}</td>
+            <td><small>{last_seen}</small></td>
+        </tr>
+        """)
+
+    table_body = "".join(link_rows) if link_rows else '<tr><td colspan="6" style="text-align: center; padding: 1.5rem; color: #666;">No correlated incident links stored. Bridge operates under dry-run default.</td></tr>'
+
+    metrics = data.get("metrics", {})
+    total_links = metrics.get("total_links", 0)
+    sentry_cnt = metrics.get("sentry_links_count", 0)
+    glitchtip_cnt = metrics.get("glitchtip_links_count", 0)
+
+    body_html = f"""
+    <h2>Observability & Incident Bridge (#649)</h2>
+    <p>Read-only console surface for Sentry/GlitchTip provider connections, error correlation,
+    and durable Gitea issue linkage.</p>
+
+    <div class="alert alert-info" style="background: #f0f4f8; padding: 1rem; border-left: 4px solid #0052cc; margin-bottom: 1.5rem;">
+        <strong>ADR Authority Model:</strong> Gitea records durable issue history. Control-plane DB coordinates incident links.
+        Sentry/GlitchTip observe errors. Raw monitoring incidents are <em>never</em> assignable control-plane work items.
+        Durable issue creation is gated and dry-runable via the <code>#612</code> bridge APIs.
+    </div>
+
+    <h3>Provider Connections</h3>
+    <div style="display: grid; grid-template-columns: repeat(auto-fit, minmax(300px, 1fr)); gap: 1rem; margin-bottom: 2rem;">
+        {provider_cards}
+    </div>
+
+    <div style="display: flex; justify-content: space-between; align-items: center; margin-bottom: 1rem;">
+        <h3 style="margin: 0;">Correlated Incidents ({total_links})</h3>
+        <div>
+            <span class="badge" style="margin-right: 0.5rem;">Sentry: {sentry_cnt}</span>
+            <span class="badge">GlitchTip: {glitchtip_cnt}</span>
+        </div>
+    </div>
+
+    <table class="table" style="width: 100%; border-collapse: collapse; border: 1px solid #ddd;">
+        <thead>
+            <tr style="background: #f9f9f9; text-align: left;">
+                <th style="padding: 0.5rem; border-bottom: 2px solid #ddd;">Provider</th>
+                <th style="padding: 0.5rem; border-bottom: 2px solid #ddd;">Incident ID</th>
+                <th style="padding: 0.5rem; border-bottom: 2px solid #ddd;">Fingerprint</th>
+                <th style="padding: 0.5rem; border-bottom: 2px solid #ddd;">Gitea Issue Link</th>
+                <th style="padding: 0.5rem; border-bottom: 2px solid #ddd;">Events</th>
+                <th style="padding: 0.5rem; border-bottom: 2px solid #ddd;">Last Seen</th>
+            </tr>
+        </thead>
+        <tbody>
+            {table_body}
+        </tbody>
+    </table>
+
+    <div style="margin-top: 2rem; padding: 1rem; background: #fafafa; border: 1px solid #eee; border-radius: 4px;">
+        <h4 style="margin-top: 0;">Reconcile & Link Controls (Gated)</h4>
+        <p style="margin-bottom: 0.5rem; color: #555;">
+            Create or reconcile durable Gitea issues from provider observations using the <code>#612</code> incident bridge:
+        </p>
+        <code>mcp call gitea_observability_reconcile_incident --provider sentry --apply false</code>
+    </div>
+    """
+
+    return render_page(title="Observability", body_html=body_html)