fix(gate): complete owning-PR renewal enforcement coverage (#945)

Addresses review 623 on PR #946 (B1 blocker, F2 medium, F3 minor).

B1 - the wiring this branch exists to install had no regression coverage.
The existing suite exercised owning_pr_renewal_from_lock,
_owning_pr_continuation_from_lock and the duplicate gate directly, but never
drove an enforcement path, so reverting any of the three call sites left the
whole repository green. Add tests/test_issue_945_enforcement_path_wiring.py,
which drives the real 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 and
mcp_server._prove_author_ownership_for_pr against a renewal-bearing lock, and
asserts each grants the exemption. Reverting the commit/create-PR recheck to
the recovery-only rebuild now fails 8 tests and 4 subtests; reverting the
assessor or the push prover fails 2 each. The suite also keeps the fail-closed
matrix on the real paths: an open PR alone, a second PR, a different PR,
branch, issue or head, identity and profile mismatch, ungranted and malformed
renewal blocks, and sequential-task non-inheritance are all still refused.

F2 - the claimant check compares lease_renewal.identity/profile against the
claimant recorded on the same lock file. Both sides are server-written fields
of one document, so it is an internal-consistency check, not verification of
the live authenticated caller. Correct the docstring and the inline comment to
say so, and document the binding that actually prevents cross-session reuse:
the enforcement paths load the lock through _load_existing_issue_lock() with no
issue coordinates, which resolves issue_lock_store.read_session_issue_lock() to
the session pointer at session-{os.getpid()}.json, so lock selection is scoped
to the operating-system process. Its limits are stated too - per-process rather
than per-authenticated-user, silent on locks reached by explicit coordinates,
and silent on two roles sharing one process. Live identity and profile stay
enforced by the mutation-authority and profile gates, not by this rebuild.

F3 - _owning_pr_continuation_from_lock previously fell through to renewal when
a dead_session_recovery block was present but failed to rebuild, so a recovery
record naming one PR could be bypassed by renewal evidence naming another.
Present-but-unusable recovery evidence is now ambiguous rather than absent and
fails closed. Because an expired lease whose recorded owner has also died
satisfies both dispositions in one gitea_lock_issue call, a sanctioned pair is
a reachable state; when both rebuild, they must agree on issue, PR, branch and
every head, or no continuation authority is returned. Recovery-only and
renewal-only locks keep their existing behaviour exactly.

Validation of the resulting token against live PR state remains untouched and
solely owned by issue_work_duplicate_gate._assess_owning_pr_exemption. No
public signature, MCP tool schema, refusal shape or reason code changes.

Tests: focused #945/#755/#760 suites 137 passed, 19 subtests. Full suite in a
branches/ worktree: 30F/5619P/6S/1013 subtests at this head vs 30F/5572P/6S/1002
at a clean checkout of 79334d48, with byte-identical failing test id sets - the
+47 passes and +11 subtests are exactly the new coverage.

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

author_issue_bootstrap.py

@@ -386,6 +386,68 @@ def run_compensating_recovery(
     return recovery_info
 
 
+def _normalize_sha(value: str | None) -> str | None:
+    """Normalize a Git object id for comparison, or ``None`` when unknown."""
+    normalized = (value or "").strip().lower()
+    return normalized or None
+
+
+def _author_bootstrap_assessment(
+    *,
+    not_applicable: bool,
+    allowed: bool,
+    block: bool,
+    reasons: list[str],
+    workspace: str,
+    root: str,
+    branch: str | None,
+    dirty: list[str],
+    under_branches: bool,
+    bootstrap_path: str | None = None,
+    local_head_sha: str | None = None,
+    remote_master_sha: str | None = None,
+    exact_next_action: str | None = None,
+) -> dict[str, Any]:
+    """Structured author-bootstrap assessment consumable by bootstrap_permits (#892).
+
+    Field shape mirrors :func:`create_issue_bootstrap._result` so the shared
+    ``bootstrap_permits_control_checkout`` predicate can prove control-checkout
+    eligibility for ``gitea_bootstrap_author_issue_worktree`` the same way it
+    does for ``create_issue``. Allowed control assessments must use empty
+    ``reasons`` — narrative belongs in other fields, not the refusal list.
+    """
+    local_tip = _normalize_sha(local_head_sha)
+    remote_tip = _normalize_sha(remote_master_sha)
+    base_tips_verified = bool(local_tip and remote_tip and local_tip == remote_tip)
+    return {
+        "not_applicable": not_applicable,
+        "allowed": allowed,
+        "block": block,
+        "proven": bool(allowed and not block and not not_applicable),
+        "reasons": list(reasons),
+        "workspace_path": workspace,
+        "canonical_repo_root": root,
+        "current_branch": branch,
+        "dirty_files": list(dirty),
+        "under_branches": under_branches,
+        "exact_next_action": exact_next_action,
+        "bootstrap_path": bootstrap_path,
+        "task_scope": "author_issue_bootstrap",
+        "local_head_sha": local_tip,
+        "remote_master_sha": remote_tip,
+        "base_tips_verified": base_tips_verified,
+    }
+
+
+EXACT_NEXT_ACTION_AUTHOR_BOOTSTRAP = (
+    "Restore the canonical control checkout to a clean accepted base branch "
+    "(master/main/dev) that matches live master, with no tracked local edits. "
+    "Re-resolve bootstrap_author_issue_worktree, then re-run "
+    "gitea_bootstrap_author_issue_worktree from that clean control checkout. "
+    "Do not use shell git worktree add as the primary path once bootstrap is healthy."
+)
+
+
 def assess_author_issue_bootstrap(
     *,
     workspace_path: str,
@@ -397,7 +459,13 @@ def assess_author_issue_bootstrap(
     remote_master_sha_error: str | None = None,
     task: str | None = None,
 ) -> dict[str, Any]:
-    """Assess whether author issue worktree bootstrap may proceed from control or worktree root."""
+    """Assess whether author issue worktree bootstrap may proceed from control or worktree root.
+
+    #892: control-checkout successes emit the full field set required by
+    ``create_issue_bootstrap.bootstrap_permits_control_checkout`` (empty reasons,
+    task_scope, base tip proof, binding paths) so the #274/#604 guards can
+    waive control-checkout for this one sanctioned bootstrap task.
+    """
     root = os.path.realpath(canonical_repo_root or "")
     workspace = os.path.realpath(workspace_path or root or ".")
     branch = (current_branch or "").strip()
@@ -407,34 +475,50 @@ def assess_author_issue_bootstrap(
         if root
         else False
     )
+    local_tip = _normalize_sha(head_sha)
+    remote_tip = _normalize_sha(remote_master_sha)
 
     if not is_author_issue_bootstrap_task(task):
-        return {
-            "not_applicable": True,
-            "allowed": False,
-            "block": False,
-            "proven": False,
-            "reasons": ["task is not author_issue_bootstrap"],
-        }
+        return _author_bootstrap_assessment(
+            not_applicable=True,
+            allowed=False,
+            block=False,
+            reasons=["task is not author_issue_bootstrap"],
+            workspace=workspace,
+            root=root,
+            branch=branch or None,
+            dirty=dirty,
+            under_branches=under_branches,
+        )
 
+    # Already under branches/: ordinary #274 path applies; not a control waiver.
     if under_branches:
-        return {
-            "not_applicable": False,
-            "allowed": True,
-            "block": False,
-            "proven": True,
-            "bootstrap_path": "existing_branches_worktree",
-            "reasons": [
-                "workspace is already a registered worktree under branches/"
-            ],
-        }
+        return _author_bootstrap_assessment(
+            not_applicable=True,
+            allowed=False,
+            block=False,
+            reasons=["workspace is under branches/; ordinary #274 path applies"],
+            workspace=workspace,
+            root=root,
+            branch=branch or None,
+            dirty=dirty,
+            under_branches=True,
+            bootstrap_path="existing_branches_worktree",
+            local_head_sha=local_tip,
+            remote_master_sha=remote_tip,
+        )
 
     reasons: list[str] = []
-    if workspace != root:
+    if not root or workspace != root:
         reasons.append(
             "bootstrap requires workspace to be canonical control checkout or branches/ worktree"
         )
-    if branch not in author_mutation_worktree.BASE_BRANCHES:
+    if not branch:
+        reasons.append(
+            "control checkout is detached HEAD; expected an accepted base branch "
+            f"({', '.join(sorted(author_mutation_worktree.BASE_BRANCHES))})"
+        )
+    elif branch not in author_mutation_worktree.BASE_BRANCHES:
         reasons.append(
             f"control checkout branch '{branch}' is not an accepted base branch "
             f"({', '.join(sorted(author_mutation_worktree.BASE_BRANCHES))})"
@@ -444,37 +528,64 @@ def assess_author_issue_bootstrap(
             f"control checkout has tracked local edits: {', '.join(dirty[:5])}"
         )
 
-    if remote_master_sha_error:
+    # Fail closed on missing tip proof (same bar as create_issue bootstrap #757).
+    if not local_tip:
         reasons.append(
-            f"could not verify live master tip: {remote_master_sha_error}"
+            "control checkout HEAD SHA is unknown; base equivalence to live "
+            "master cannot be proven (fail closed)"
+        )
+    resolver_error = (remote_master_sha_error or "").strip() or None
+    if resolver_error:
+        reasons.append(
+            f"live master tip could not be resolved ({resolver_error}); "
+            "base equivalence cannot be proven (fail closed)"
+        )
+    elif not remote_tip:
+        reasons.append(
+            "live master tip is unknown; base equivalence cannot be proven "
+            "(fail closed)"
+        )
+    elif local_tip and remote_tip and local_tip != remote_tip:
+        reasons.append(
+            f"control checkout HEAD ({local_tip[:12]}) != live master tip "
+            f"({remote_tip[:12]})"
         )
-    elif remote_master_sha and head_sha:
-        h = head_sha.strip().lower()
-        rm = remote_master_sha.strip().lower()
-        if h != rm:
-            reasons.append(
-                f"control checkout HEAD ({h[:12]}) != live master tip ({rm[:12]})"
-            )
 
     if reasons:
-        return {
-            "not_applicable": False,
-            "allowed": False,
-            "block": True,
-            "proven": False,
-            "reasons": reasons,
-        }
+        return _author_bootstrap_assessment(
+            not_applicable=False,
+            allowed=False,
+            block=True,
+            reasons=reasons,
+            workspace=workspace,
+            root=root,
+            branch=branch or None,
+            dirty=dirty,
+            under_branches=False,
+            local_head_sha=local_tip,
+            remote_master_sha=remote_tip,
+            exact_next_action=EXACT_NEXT_ACTION_AUTHOR_BOOTSTRAP,
+        )
 
-    return {
-        "not_applicable": False,
-        "allowed": True,
-        "block": False,
-        "proven": True,
-        "bootstrap_path": "clean_canonical_control_checkout",
-        "reasons": [
-            "control checkout is clean on accepted base branch matching live master"
-        ],
-    }
+    # Allowed: empty reasons so bootstrap_permits_control_checkout can pass.
+    return _author_bootstrap_assessment(
+        not_applicable=False,
+        allowed=True,
+        block=False,
+        reasons=[],
+        workspace=workspace,
+        root=root,
+        branch=branch or None,
+        dirty=dirty,
+        under_branches=False,
+        bootstrap_path="clean_canonical_control_checkout",
+        local_head_sha=local_tip,
+        remote_master_sha=remote_tip,
+        exact_next_action=(
+            "Call gitea_bootstrap_author_issue_worktree with the allocated "
+            "issue/lease pins; it will create the branches/ worktree and lock."
+        ),
+    )
 
 
 import fcntl

create_issue_bootstrap.py

@@ -241,9 +241,14 @@ def bootstrap_permits_control_checkout(
     caller's ordinary block in force.
 
     ``assessment`` is server-derived only: it is produced by
-    :func:`assess_create_issue_bootstrap` from inspected repository state. It is
-    never accepted from an MCP tool argument, so no caller can assert
-    eligibility it has not proven.
+    :func:`assess_create_issue_bootstrap` or
+    :func:`author_issue_bootstrap.assess_author_issue_bootstrap` from inspected
+    repository state. It is never accepted from an MCP tool argument, so no
+    caller can assert eligibility it has not proven.
+
+    #892: author issue worktree bootstrap uses the same predicate with
+    ``task_scope='author_issue_bootstrap'`` so a clean control checkout can
+    create the first ``branches/`` worktree without the lock↔worktree cycle.
     """
     if not isinstance(assessment, dict):
         return False
@@ -264,9 +269,16 @@ def bootstrap_permits_control_checkout(
     if assessment.get("reasons"):
         return False
 
-    # Scope proof: only the create_issue bootstrap, only via the clean
-    # canonical control checkout path.
-    if assessment.get("task_scope") != "create_issue_only":
+    # Scope proof: create_issue (#749) or author issue bootstrap (#850/#892),
+    # only via the clean canonical control checkout path.
+    task_scope = assessment.get("task_scope")
+    if is_create_issue_task(task):
+        if task_scope != "create_issue_only":
+            return False
+    elif author_issue_bootstrap.is_author_issue_bootstrap_task(task):
+        if task_scope != "author_issue_bootstrap":
+            return False
+    else:
         return False
     if assessment.get("bootstrap_path") != "clean_canonical_control_checkout":
         return False

docs/gitea-execution-profiles.md

@@ -589,47 +589,6 @@ When dynamic profile switching is enabled and a profile is activated via `gitea_
 2. Call `gitea_whoami` with the target remote to prove and verify the fresh Gitea authenticated identity.
 This guarantees the active profile operations align with the actual Gitea authenticated user credential.
 
-### 4. Review-State Invalidation on Profile Switch (#690)
-
-A cross-profile activation (e.g. reviewer → author → reviewer) is a session
-boundary for formal review state. On any switch where the activated profile
-differs from the previous one, `gitea_activate_profile` invalidates, in
-memory **and** in durable session state (for both the old and new profile
-identities):
-
-- preflight identity/capability stamps (`gitea_whoami` / `gitea_resolve_task_capability` proof),
-- review workflow-load proof (`gitea_load_review_workflow`),
-- the review decision lock (including `final_review_decision_ready` markers),
-- the reviewer PR session lease binding,
-- live namespace-health assessments.
-
-Before any formal verdict (`gitea_mark_final_review_decision` /
-`gitea_submit_pr_review`) the full reviewer preflight must be re-established
-under the new profile: `gitea_whoami`, `gitea_load_review_workflow`,
-`gitea_resolve_task_capability(review_pr)`, live head re-pin, and lease
-re-acquire/adopt. Switching back to the earlier profile does **not**
-resurrect the prior run — durable state keyed by either profile identity is
-cleared at switch time.
-
-The supported pattern remains **separate session/namespace per role**
-(dual-namespace, §2): file author-side follow-ups from an author session,
-not by hopping profiles inside a formal review run. Runtime profile
-switching is the operator-approved exception and always carries the
-re-preflight cost above.
-
-### 5. Namespace Provenance (#690)
-
-The server cannot derive its own client-managed MCP namespace name, so a
-launcher may declare it via the `GITEA_MCP_NAMESPACE` environment variable
-(e.g. `gitea-reviewer`). `gitea_whoami`, `gitea_get_runtime_context`, and
-`gitea_resolve_task_capability` report `namespace_provenance` — the
-configured client namespace, the active execution profile, and, for tasks
-with a required namespace (`review_pr` → `gitea-reviewer`, `merge_pr` →
-`gitea-merger`), a mismatch verdict. A declared namespace that disagrees
-with the requested task's required namespace **fails closed**. An
-undeclared namespace is reported as `unknown` and is never treated as
-proof either way.
-
 ## Gitea MCP Runtime Isolation and Worktree Safety
 
 To ensure high availability and prevent broken feature worktrees from disabling essential security/identity controls, the Gitea MCP server implements runtime isolation:

docs/mcp-namespace-eof-recovery.md

@@ -153,7 +153,19 @@ not a tool argument: a session must never be able to authorize itself.
 ## Related
 
 - #630 — manual daemon killing as contaminated recovery (this contrast, enforced).
+- #657 — restart-path inventory and daemon classification.
+- #686 — manual server launch detection & fail-closed provenance gate.
 - #531 / #544 — stale-runtime detection (`ps`-based); sibling failure mode.
 - #558 / `docs/mcp-daemon-import-guard.md` — why shell imports are not a repair.
 - `docs/mcp-client-registration.md` — per-server registration contract.
 - `docs/mcp-namespace-health.md` — probe sources and mutation enforcement.
+
+## Sanctioned reconnect vs forbidden manual launch (#686)
+
+In addition to manual process killing (#630), manually launching a duplicate role server from an ad hoc shell (`python3 mcp_server.py`) is forbidden and fail-closed:
+
+- **Why manual launches are unsupported:** A terminal-launched `mcp_server.py` holds its own stdio transport; it can never bind to the IDE client's stdio pipes. It cannot restore a dropped IDE namespace, and a manual duplicate process masks stale client-managed runtimes for that profile, defeating stale-runtime gates.
+- **Sanctioned path:** Supported recovery is IDE/client-managed reconnect only (`/mcp reconnect`, IDE restart, or sanctioned reconnect exposure).
+- **Fail-closed enforcement (#686):** Mutating tools on a server lacking client-managed launch provenance (`GITEA_CLIENT_MANAGED=1`) refuse execution fail-closed with typed blocker `unsupported_manual_launch` and an exact next action. Unsupported `GITEA_*` env overrides (e.g. `GITEA_DUMMY`) are surfaced in diagnostics rather than silently ignored.
+- **Inventory & staleness:** Staleness diagnostics ignore non-client-managed duplicates when evaluating runtime freshness and inventory duplicate processes per profile (#657, #686).
+

docs/mcp-namespace-health.md

@@ -86,24 +86,3 @@ When a namespace returns EOF, follow
 
 When blocked, repair the IDE namespace and re-record a healthy
 `client_namespace` assessment before retrying the mutation.
-
-## Namespace provenance (#690)
-
-A server process cannot derive the name of the client-managed namespace it is
-registered under, so the launcher may declare it with the
-`GITEA_MCP_NAMESPACE` environment variable (e.g. `GITEA_MCP_NAMESPACE=gitea-reviewer`).
-
-- `gitea_whoami`, `gitea_get_runtime_context`, and
-  `gitea_resolve_task_capability` report `namespace_provenance`: the declared
-  client namespace, the active execution profile, and — for tasks with a
-  required namespace (`review_pr`/`submit_review` → `gitea-reviewer`,
-  `merge_pr` → `gitea-merger`) — a `mismatch` verdict.
-- A declared namespace that disagrees with the requested task's required
-  namespace **fails closed** (`allowed_in_current_session=false` with a STOP
-  guidance entry).
-- An undeclared namespace is reported as `namespace_source="unknown"` and is
-  never treated as proof either way.
-- A profile switch via `gitea_activate_profile` clears all recorded live
-  namespace-health assessments; re-probe through the client before further
-  review/merge mutations.
-

docs/remote-mcp/coupling-inventory.md

@@ -0,0 +1,230 @@
+# Remote-MCP coupling inventory
+
+Every place the Gitea MCP server depends on being a local, client-spawned, stdio-attached
+process on the operator's machine.
+
+- **Issue:** #930 (Remote-MCP 01), child 1 of epic #929.
+- **Generated against commit:** `7bf4f1258451823a55b36d2157e74f8457165088` (`master`).
+- **Anchors:** every `file:line` below resolves at the commit above and at the commit that
+  adds this document. This change adds one new file and edits no existing file, so no
+  existing line number shifts between the two.
+- **Scope:** documentation only. No server behavior changes in this child.
+
+## How to read an entry
+
+| Field | Meaning |
+| ----- | ------- |
+| **Anchor** | `file:line` at the commit under review. |
+| **Assumes today** | What the code takes for granted while running as a local stdio process. |
+| **Observes remotely** | What the same code would actually see on a shared remote host. |
+| **Class** | One of: *portable as written*, *needs a seam*, *needs a replacement*, *cannot be remote*. |
+| **Owner** | Exactly one epic child (#931–#939) responsible for the fix. |
+
+Classification meanings:
+
+- **portable as written** — the code is already transport-, host-, and principal-neutral; it
+  moves unchanged once its inputs are supplied by a remote-aware caller.
+- **needs a seam** — the logic is correct but is wired to a hard-coded local source. It needs
+  an injection point, not new semantics.
+- **needs a replacement** — the semantics themselves are local-only. A remote deployment
+  needs a differently-defined mechanism, not the same mechanism relocated.
+- **cannot be remote** — the operation is inherently about the operator's own machine
+  (its process table, its keychain, its checkout). It must either stay local behind an
+  explicit boundary or be deleted from the remote surface.
+
+---
+
+## 1. Transport bind
+
+The transport is bound literally, once, at process start, and the bound value is the root of
+the mutation-authorization chain.
+
+| ID | Anchor | Assumes today | Observes remotely | Class | Owner |
+| -- | ------ | ------------- | ----------------- | ----- | ----- |
+| T1 | `gitea_mcp_server.py:23750` | The single production bind call passes the literal `transport="stdio"` immediately before the server loop. | The literal is wrong for any non-stdio deployment; there is no parameter to change it. | needs a seam | #931 |
+| T2 | `mcp_daemon_guard.py:45` | `_PRODUCTION_TRANSPORTS = frozenset({"stdio"})` is the closed allowlist of production transports. | A remote transport name is rejected by the allowlist before any other check runs. | needs a seam | #931 |
+| T3 | `mcp_daemon_guard.py:174` | `bind_native_mcp_transport` raises `UnsanctionedRuntimeError` for any transport outside `_PRODUCTION_TRANSPORTS` (raise at `mcp_daemon_guard.py:187`). | The remote server fails to start rather than degrading; the failure is correct, but the allowlist is the only thing that must change. | needs a seam | #931 |
+| T4 | `mcp_daemon_guard.py:328` | `is_native_mcp_transport()` asserts a process-local runtime record whose `pid` matches `os.getpid()` and whose phase is `transport_bound`. The predicate itself names no transport. | Unchanged semantics: one server process that bound one transport. It stays true on a remote host. | portable as written | #931 |
+| T5 | `mcp_daemon_guard.py:349` | `is_production_native_mcp_transport()` adds only a `mode == production` check on top of T4. | Unchanged. | portable as written | #931 |
+| T6 | `irrecoverable_provenance.py:497` | `assess_transport_for_auth_mint()` requires production native transport before minting non-forgeable recovery authorization (#709 F1). | The gate is transport-agnostic in form, but its guarantee — "an ordinary Python process cannot reach this" — is currently underwritten by the stdio bind. Under a remote transport the guarantee must be re-derived from the authenticated session, not from the bind. | needs a seam | #931 |
+| T7 | `gitea_mcp_server.py:8375` | Consumer: refuses to proceed unless `assess_transport_for_auth_mint()` allows. | Unchanged given a corrected T6. | portable as written | #931 |
+| T8 | `gitea_mcp_server.py:8624` | Second consumer of the same gate on the confirmation path. | Unchanged given a corrected T6. | portable as written | #931 |
+| T9 | `mcp_server.py:4` | Module docstring asserts "Runs over stdio." as a property of the server. | The stated contract becomes false on the remote deployment and is load-bearing documentation for operators. | needs a replacement | #931 |
+
+## 2. Launch provenance
+
+Mutations fail closed unless the process can prove a client launched it with real stdio pipes
+and `GITEA_CLIENT_MANAGED` provenance. Every proof in this section is a statement about the
+local operating system.
+
+| ID | Anchor | Assumes today | Observes remotely | Class | Owner |
+| -- | ------ | ------------- | ----------------- | ----- | ----- |
+| P1 | `gitea_mcp_server.py:14588` | `_is_client_managed_process()` derives provenance from `GITEA_CLIENT_MANAGED` / `GITEA_MCP_CLIENT_MANAGED` / `GITEA_SERVER_PROVENANCE` / `GITEA_FORCE_CLIENT_MANAGED` on this process's own environment. | A long-lived remote process has one environment for all callers, so a per-process env var can no longer say anything about the caller that issued a request. | needs a replacement | #934 |
+| P2 | `gitea_mcp_server.py:14606` | Falls back to `sys.stdin.isatty()`: an active TTY on stdin means a human launched it from a terminal, so refuse. | A remote server has no meaningful stdin. The signal is absent, not merely different. | cannot be remote | #934 |
+| P3 | `gitea_mcp_server.py:14618` | `_provenance_mutation_block()` emits `blocker_kind: "unsupported_manual_launch"` and a "reconnect the IDE/client-managed MCP namespace" remediation. | The block shape is reusable; its predicate and its remediation text are both stdio-specific. | needs a seam | #934 |
+| P4 | `gitea_mcp_server.py:20599` | `_check_mcp_runtimes_diagnostics()` shells `ps -o pid,lstart,command -ax` and greps for `mcp_server.py` to find peer role servers. | On a shared host the process table lists unrelated tenants' processes, or none at all under a container. Peer discovery by `ps` has no remote meaning. | cannot be remote | #934 |
+| P5 | `gitea_mcp_server.py:20702` | More than one process per `GITEA_MCP_PROFILE` in the local process table is reported as a duplicate-launch fault. | A remote endpoint is expected to serve many concurrent sessions per role. "Two processes for one role" becomes the normal case, so the check inverts from a safety net into a false wall. | cannot be remote | #934 |
+| P6 | `gitea_mcp_server.py:20715` | Processes lacking client-managed provenance are ignored for runtime freshness and reported as manual launches. | Same defect as P5: correctness depends on enumerating local peers. | cannot be remote | #934 |
+| P7 | `gitea_config.py:1172` | `RECOGNIZED_GITEA_ENV_KEYS` is the allowlist of `GITEA_*` env vars a legitimately launched server may carry; anything else is contamination. | Configuration on a remote host arrives from deployment tooling, not from a client-authored env block. The allowlist keeps working mechanically but stops proving anything about provenance. | needs a replacement | #934 |
+| P8 | `gitea_mcp_server.py:20683` | The unsupported-env scan applies `RECOGNIZED_GITEA_ENV_KEYS` to *other* processes' environments harvested via `ps eww <pid>`. | Reading another process's environment is unavailable or prohibited across tenants, and is not exposed in this form outside macOS/BSD `ps`. | cannot be remote | #934 |
+| P9 | `mcp_daemon_guard.py:126` | `mark_sanctioned_daemon()` requires the claiming stack frame's resolved absolute path to be the canonical `mcp_server.py` / `gitea_mcp_server.py` next to the guard module; basename spoofing is rejected. | Entrypoint-path identity still exists on a remote host, but it authenticates the *deployment*, not the *caller*. It must be kept and demoted from "authorizes mutations" to "authorizes the process". | needs a seam | #934 |
+| P10 | `gitea_config.py:1233` | The client-config generator emits `"GITEA_CLIENT_MANAGED": "1"` into each generated MCP client entry, alongside `GITEA_MCP_CONFIG` / `GITEA_MCP_PROFILE`. | A remote endpoint is addressed by URL and credential, not by a spawn command with an env block. This generator produces the wrong artifact entirely. | needs a replacement | #938 |
+| P11 | `mcp_namespace_health.py:232` | Namespace health classifies a namespace as `client_managed` or `manual_launch` from the reported env summary. | During dual-run, local and remote namespaces coexist and must both be classifiable; a two-valued local/manual axis cannot express "remote endpoint, authenticated session". | needs a replacement | #939 |
+| P12 | `gitea_mcp_server.py:18161` | The diagnostics payload reports `server_provenance` as exactly `"client_managed"` or `"manual_launch"`. | This is the field a cutover operator reads to confirm which deployment served a call. It must gain a remote value before dual-run parity can be validated. | needs a replacement | #939 |
+
+## 3. Role binding
+
+Role separation is currently enforced by *which process a call reaches*. The process is pinned
+to one role for its lifetime by an environment variable.
+
+| ID | Anchor | Assumes today | Observes remotely | Class | Owner |
+| -- | ------ | ------------- | ----------------- | ----- | ----- |
+| R1 | `gitea_config.py:54` | `ENV_PROFILE = "GITEA_MCP_PROFILE"` is the single source of the active profile, read from the process environment. | One shared process serves several principals; a process-wide profile cannot answer "who is calling now". This is the root of the coupling. | needs a replacement | #932 |
+| R2 | `review_workflow_load.py:95` | Reads `GITEA_MCP_PROFILE` directly to decide the reviewer workflow binding. | Reads the deployment's profile, not the caller's, silently granting or denying the wrong role. | needs a replacement | #932 |
+| R3 | `mcp_discoverability.py:152` | Reads `GITEA_MCP_PROFILE` to describe the namespace to the client. | Correct logic, wrong input source; it needs the request principal injected. | needs a seam | #932 |
+| R4 | `webui/deployment_boundary.py:115` | Reads `GITEA_MCP_PROFILE` to classify the deployment boundary for the console. | Same as R3. | needs a seam | #932 |
+| R5 | `gitea_mcp_server.py:21106` | Remediation text instructs the operator to "Relaunch the server with `GITEA_MCP_PROFILE` set to a profile that has the required permission". | Relaunching a shared remote endpoint to change one caller's role is not a valid instruction; it would re-role every other session. | needs a replacement | #932 |
+| R6 | `native_mcp_preference.py:93` | Detects shell commands that override `GITEA_MCP_PROFILE` away from the session (`native_mcp_preference.py:223`) and flags them as CLI auth divergence. | The divergence check is genuinely useful and survives, but its notion of "the session's profile" must come from the request principal. | needs a seam | #932 |
+| R7 | `gitea_mcp_server.py:20671` | Recovers a peer server's role by regexing `GITEA_MCP_PROFILE=` out of that process's environment. | Depends on P4/P8 process-table access; role discovery by peer-env scraping has no remote analogue. | cannot be remote | #932 |
+
+## 4. Credentials
+
+Every token resolves, directly or indirectly, from one human's macOS keychain.
+
+| ID | Anchor | Assumes today | Observes remotely | Class | Owner |
+| -- | ------ | ------------- | ----------------- | ----- | ----- |
+| C1 | `gitea_config.py:956` | `_keychain_token()` shells `security find-generic-password -s <item> -w`. | `security(1)` is a macOS binary reading the calling user's login keychain. It does not exist on a Linux host and would be the wrong identity even on a shared Mac. | cannot be remote | #933 |
+| C2 | `gitea_config.py:974` | `resolve_token(profile, keychain_lookup=_keychain_token)` dispatches on `auth.type` of `env` or `keychain`, defaulting the lookup to C1. | The injectable `keychain_lookup` parameter is the existing seam; a remote credential provider plugs in here without changing the dispatch. | needs a seam | #933 |
+| C3 | `gitea_config.py:1015` | `keychain_auth(item_id)` constructs the `{"type": "keychain", "id": ...}` reference stored in profiles. | The reference type itself encodes "macOS keychain" into persisted config. A remote provider needs a new auth reference type, not a new value of this one. | needs a replacement | #933 |
+| C4 | `mcp_daemon_guard.py:440` | `assert_keychain_access_allowed()` fails closed for git-credential keychain fill outside a sanctioned daemon, with an operator opt-out env var. | The gate protects a mechanism that will not exist remotely. Its replacement must gate the *credential provider* call, not the keychain call, or the protection silently lapses. | needs a replacement | #933 |
+| C5 | `sentry_incident_bridge.py:190` | `resolve_token(env)` resolves the Sentry token from an injected env mapping with no keychain path. | Already host-neutral; it is the shape the Gitea credential path should converge on. | portable as written | #933 |
+| C6 | `gitea_mcp_server.py:18469` | The profile-audit tool calls `gitea_config.resolve_token(p)` for every configured profile to report "credentials present" without networking. | On a remote host this would materialize every principal's credential inside one process — an audit surface that becomes a credential-aggregation risk. | needs a seam | #933 |
+
+## 5. Runtime freshness
+
+The mutation gate is defined as "the commit this process started at matches the checkout on
+this disk, and both match live master". Two of those three terms are local-disk facts.
+
+| ID | Anchor | Assumes today | Observes remotely | Class | Owner |
+| -- | ------ | ------------- | ----------------- | ----- | ----- |
+| F1 | `master_parity_gate.py:168` | `capture_startup_parity(root)` reads git `HEAD` from the server's own root once at startup and returns it as the baseline. | A remote host carries a deployed artifact, not the operator's checkout. Its `HEAD` says nothing about the operator's working tree, which is the thing the gate exists to protect. | cannot be remote | #935 |
+| F2 | `master_parity_gate.py:255` | `mutation_safe = determinable and in_parity and live_known and not live_stale` — a conjunction of two local-HEAD comparisons and one live-remote comparison. | Two of the three conjuncts lose meaning, so the whole verdict does. A remote deployment needs a redefined, testable freshness predicate rather than this one relocated. | needs a replacement | #935 |
+| F3 | `master_parity_gate.py:164` | The live-remote head is probed and cached per `(root, remote, branch)`, keyed on the local root. | The live-remote probe is the one conjunct that survives; it needs a key that is not the operator's filesystem path. | needs a seam | #935 |
+| F4 | `gitea_mcp_server.py:18262` | `gitea_assess_master_parity` publishes `startup_head` / `local_head` / `live_remote_head` / `mutation_safe` as the authoritative mutation-safety verdict. | The tool's contract is consumed by every mutation caller and by the operator; it must keep its shape while its semantics are redefined, or every consumer breaks at once. | needs a replacement | #935 |
+| F5 | `gitea_mcp_server.py:23054` | Falls back to `_process_boot_head_sha` — the commit this process booted at — when the parity payload has no `startup_head`. | Same defect as F1, in a fallback path that is easy to miss when F1 is fixed. | needs a seam | #935 |
+| F6 | `gitea_mcp_server.py:20615` | Staleness is also inferred from `os.path.getmtime()` of `gitea_mcp_server.py` under `PROJECT_ROOT` (`gitea_mcp_server.py:20611`), compared against peer process start times. | File mtime on a deployed artifact tracks the deploy, not the operator's edits, and the peer start times it is compared against come from the unavailable process table (P4). | cannot be remote | #935 |
+
+## 6. Local filesystem
+
+Author and reviewer tools act directly on the operator's checkout.
+
+| ID | Anchor | Assumes today | Observes remotely | Class | Owner |
+| -- | ------ | ------------- | ----------------- | ----- | ----- |
+| L1 | `gitea_mcp_server.py:10122` | `gitea_bootstrap_author_issue_worktree` creates and binds a git worktree on the server's own disk. | The remote host has no operator checkout to add a worktree to. Executing this remotely would act on the wrong disk while reporting success. | cannot be remote | #936 |
+| L2 | `gitea_mcp_server.py:190` | `ACTIVE_WORKTREE_ENV = "GITEA_ACTIVE_WORKTREE"` and `AUTHOR_WORKTREE_ENV` (`gitea_mcp_server.py:191`) carry the active workspace as process-wide environment. | Process-wide workspace state cannot represent per-session workspaces on a shared endpoint. | needs a replacement | #936 |
+| L3 | `gitea_mcp_server.py:9801` | Binding a worktree writes `os.environ["GITEA_AUTHOR_WORKTREE"]` and `os.environ["GITEA_ACTIVE_WORKTREE"]` (`gitea_mcp_server.py:9802`), mutating global process state. | One session's bind would silently retarget every other concurrent session in the same process. This is a correctness bug the moment concurrency is real. | needs a replacement | #936 |
+| L4 | `reviewer_inventory_worktree.py:48` | `_BRANCHES_WORKTREE_RE = re.compile(r"\bbranches/", re.I)` requires review worktree paths to sit under `branches/`. | A path convention on the operator's machine, asserted as a validation rule. It needs to become a property of a declared workspace, not a substring test. | needs a seam | #936 |
+| L5 | `stable_control_runtime.py:54` | `DEV_WORKTREE_SEGMENT = "branches"` classifies a process root as a development worktree by path segment. | Same class of assumption as L4, on the runtime-classification side. | needs a seam | #936 |
+| L6 | `mcp_server.py:42` | `check_conflict_markers()` runs at import and `os.walk`s the install directory for unresolved conflict markers, `sys.exit(1)` on a hit. | On a remote host it scans a deployed artifact, which by construction never has conflict markers — so the guard passes trivially and stops protecting the thing it was written to protect. | needs a replacement | #936 |
+| L7 | `role_session_router.py:487` | `check_mid_merge()` reports infra-stop from `.git/MERGE_HEAD`, `rebase-merge`, `rebase-apply` and a source conflict scan under the server's project root. | Same inversion as L6: it would report the deployment's git state, not the operator's. | needs a replacement | #936 |
+| L8 | `author_issue_bootstrap.py:996` | Enumerates worktrees with `git -C <root> worktree list --porcelain`. | Requires a real local clone with real worktrees; there is nothing equivalent to enumerate remotely. | cannot be remote | #936 |
+| L9 | `mcp_server.py:10` | Redirects `sys.stderr` to the fixed path `/tmp/mcp_server_stderr.log` outside pytest. | A single fixed `/tmp` path is shared by every concurrent server on a host and is not a deployment's logging surface. | needs a replacement | #938 |
+| L10 | `gitea_mcp_server.py:2314` | `ISSUE_LOCK_FILE = "/tmp/gitea_issue_lock.json"` — the legacy single global lock slot. | One global `/tmp` slot per host cannot represent concurrent remote sessions and is world-visible on a shared machine. | needs a replacement | #937 |
+| L11 | `issue_lock_provenance.py:14` | `ISSUE_LOCK_FILE = os.environ.get("GITEA_ISSUE_LOCK_FILE", "/tmp/gitea_issue_lock.json")` keeps the same `/tmp` default in the provenance path. | Same as L10; the env override is a local escape hatch, not a remote design. | needs a replacement | #937 |
+
+## 7. Durable state
+
+Locks, leases, session state, and the control-plane database live in the operator's home
+directory and are keyed on local PIDs.
+
+| ID | Anchor | Assumes today | Observes remotely | Class | Owner |
+| -- | ------ | ------------- | ----------------- | ----- | ----- |
+| S1 | `issue_lock_store.py:26` | `DEFAULT_LOCK_DIR = ~/.cache/gitea-tools/issue-locks` — per-issue lock files under one user's home. | A shared endpoint has no single operator home; per-user paths make locks invisible across sessions and hosts. | needs a replacement | #937 |
+| S2 | `issue_lock_store.py:83` | `session_pointer_path()` names the session pointer file `session-<os.getpid()>.json`. | Many sessions share one PID on a remote server, so the pointer collapses to a single slot and sessions overwrite each other. | cannot be remote | #937 |
+| S3 | `issue_lock_store.py:98` | `is_process_alive(pid)` decides lock liveness by probing the local process table. | A PID recorded by one host is meaningless on another, and may coincidentally match a live unrelated process. | cannot be remote | #937 |
+| S4 | `issue_lock_store.py:213` | Lock records stamp `session_pid` and `pid` from `os.getpid()`. | The recorded identity no longer distinguishes sessions; ownership checks silently pass for the wrong caller. | needs a replacement | #937 |
+| S5 | `mcp_session_state.py:27` | `DEFAULT_STATE_DIR = ~/.cache/gitea-tools/session-state`, mode `0o700`. | Same home-directory coupling as S1, for review decision locks and workflow proofs. | needs a replacement | #937 |
+| S6 | `mcp_session_state.py:559` | Session bodies stamp `session_pid` and `writer_pid` from `os.getpid()` (`mcp_session_state.py:560`). | Writer attribution collapses across concurrent sessions in one process. | needs a replacement | #937 |
+| S7 | `control_plane_db.py:47` | `DEFAULT_DB_PATH = ~/.cache/gitea-tools/control-plane/control_plane.sqlite3`. | A per-user SQLite file is not reachable by, or safe for, multiple remote sessions or multiple hosts. | needs a replacement | #937 |
+| S8 | `control_plane_db.py:386` | `sqlite3.connect(self.db_path, timeout=30)` — single-writer file locking tuned for one local process. | SQLite's write lock does not extend across hosts and degrades sharply under real concurrency; the store needs a concurrency-safe backend. | needs a replacement | #937 |
+| S9 | `control_plane_db.py:1145` | Lease rows record `owner_pid` defaulting to `os.getpid()` (also `control_plane_db.py:2039`). | PID-keyed lease ownership is unusable across hosts and ambiguous within one shared process. | cannot be remote | #937 |
+| S10 | `mcp_daemon_guard.py:53` | `_DEFAULT_SESSION_STATE_DIR` is pinned once at transport bind so a later `GITEA_MCP_SESSION_STATE_DIR` change cannot manufacture a second authority domain (#695 AC2). | The single-authority-domain invariant is exactly right and must be preserved; only its backing location needs to move. | needs a seam | #937 |
+| S11 | `gitea_mcp_server.py:11875` | Reviewer-lease reclaim reads `owner_pid_alive` from the lease freshness record to decide whether an owner is dead. | Consumes S3/S9; a false "owner alive" or "owner dead" here reclaims or refuses a live lease. This is the highest-consequence consumer of PID liveness. | cannot be remote | #937 |
+
+---
+
+## Summary
+
+### Entries per category
+
+| Category | Entries |
+| -------- | ------: |
+| 1. Transport bind | 9 |
+| 2. Launch provenance | 12 |
+| 3. Role binding | 7 |
+| 4. Credentials | 6 |
+| 5. Runtime freshness | 6 |
+| 6. Local filesystem | 11 |
+| 7. Durable state | 11 |
+| **Total** | **62** |
+
+No category is empty, so no "this category has no coupling" justification is required.
+
+### Entries per classification
+
+| Classification | Entries |
+| -------------- | ------: |
+| portable as written | 5 |
+| needs a seam | 16 |
+| needs a replacement | 26 |
+| cannot be remote | 15 |
+| **Total** | **62** |
+
+### Category × classification
+
+| Category | portable | seam | replacement | cannot | Total |
+| -------- | -------: | ---: | ----------: | -----: | ----: |
+| 1. Transport bind | 4 | 4 | 1 | 0 | 9 |
+| 2. Launch provenance | 0 | 2 | 5 | 5 | 12 |
+| 3. Role binding | 0 | 3 | 3 | 1 | 7 |
+| 4. Credentials | 1 | 2 | 2 | 1 | 6 |
+| 5. Runtime freshness | 0 | 2 | 2 | 2 | 6 |
+| 6. Local filesystem | 0 | 2 | 7 | 2 | 11 |
+| 7. Durable state | 0 | 1 | 6 | 4 | 11 |
+| **Total** | **5** | **16** | **26** | **15** | **62** |
+
+### Entries per epic child
+
+Every child from 2 through 10 is named by at least one entry, and every entry names exactly
+one child.
+
+| Child | Issue | Title | Entries | IDs |
+| ----: | ----- | ----- | ------: | --- |
+| 2 | #931 | Transport-neutral bind seam | 9 | T1–T9 |
+| 3 | #932 | Per-request principal resolution | 7 | R1–R7 |
+| 4 | #933 | Server-side credential provider | 6 | C1–C6 |
+| 5 | #934 | Remote-session provenance | 9 | P1–P9 |
+| 6 | #935 | Redefined master-parity gate | 6 | F1–F6 |
+| 7 | #936 | Local-filesystem vs remotable tool split | 8 | L1–L8 |
+| 8 | #937 | Concurrency-safe session, lock, and lease state | 13 | L10, L11, S1–S11 |
+| 9 | #938 | Authenticated remote MCP endpoint | 2 | P10, L9 |
+| 10 | #939 | Dual-run cutover and rollback | 2 | P11, P12 |
+| | | **Total** | **62** | |
+
+## Notes for downstream children
+
+- **The three highest-risk entries are P5, F2, and S11.** Each is a guard that does not
+  merely stop working remotely — it inverts. P5 turns concurrency into a reported fault,
+  F2 returns a verdict computed from terms that no longer mean anything, and S11 reclaims
+  or refuses leases on a PID-liveness answer that is wrong rather than unknown. A gate that
+  fails open while still reporting green is worse than one that fails to start.
+- **T4, T5, T7, T8, and C5 are the portable core.** They show the target shape: predicates
+  over injected inputs, with no reference to the host, the process table, or the operator's
+  disk.
+- **The keychain seam already exists** at C2 (`resolve_token`'s injectable `keychain_lookup`).
+  #933 should widen that seam rather than introduce a parallel path, and must remember C4 —
+  the guard protecting the old mechanism has to be re-pointed, or the protection lapses
+  silently when the mechanism is replaced.
+- **`branches/` appears as a validation rule in at least two independent places** (L4, L5).
+  Path-substring conventions tend to have more copies than expected; #936 should re-grep
+  rather than trust this list to be exhaustive for that specific pattern.

gitea_config.py

@@ -1169,10 +1169,57 @@ def server_command():
     return python, [os.path.join(root, "mcp_server.py")]
 
 
+RECOGNIZED_GITEA_ENV_KEYS = frozenset({
+    "GITEA_MCP_CONFIG",
+    "GITEA_MCP_PROFILE",
+    "GITEA_PROFILE_NAME",
+    "GITEA_SERVICE",
+    "GITEA_EXECUTION_ROLE",
+    "GITEA_CLIENT_MANAGED",
+    "GITEA_MCP_CLIENT_MANAGED",
+    "GITEA_SERVER_PROVENANCE",
+    "GITEA_AUTHOR_WORKTREE",
+    "GITEA_ACTIVE_WORKTREE",
+    "GITEA_DISABLE_KEYCHAIN",
+    "GITEA_CONTROL_PLANE_DB",
+    "GITEA_DB_PATH",
+    "GITEA_LOG_LEVEL",
+    "GITEA_DEBUG",
+    "GITEA_HMAC_SECRET",
+    "GITEA_IRRECOVERABLE_HMAC_SECRET",
+    "GITEA_FORCE_MCP_RUNTIME_CHECK",
+    "GITEA_FORCE_CLIENT_MANAGED",
+})
+
+RECOGNIZED_GITEA_ENV_PREFIXES = (
+    "GITEA_TOKEN_",
+    "GITEA_PASS_",
+    "GITEA_USER_",
+    "GITEA_URL_",
+    "GITEA_HOST_",
+    "GITEA_REMOTE_",
+    "GITEA_HTTP_HEADER_",
+)
+
+
+def get_unconsumed_gitea_env_overrides(env=None) -> dict[str, str]:
+    """Find unsupported GITEA_* env vars present in *env* (defaults to os.environ)."""
+    target = os.environ if env is None else env
+    unconsumed = {}
+    for key, value in target.items():
+        if key.startswith("GITEA_"):
+            if key in RECOGNIZED_GITEA_ENV_KEYS:
+                continue
+            if any(key.startswith(p) for p in RECOGNIZED_GITEA_ENV_PREFIXES):
+                continue
+            unconsumed[key] = str(value)
+    return unconsumed
+
+
 def launcher_entry(profile_name, config_path=None):
     """Return a thin MCP launcher entry for *profile_name*.
 
-    Contains only command/args and the two GITEA_MCP_* env vars — never a token
+    Contains command/args and the GITEA_MCP_* / GITEA_CLIENT_MANAGED env vars — never a token
     or password. Suitable for Claude / Gemini / Codex ``mcpServers`` blocks.
     """
     command, args = server_command()
@@ -1183,11 +1230,13 @@ def launcher_entry(profile_name, config_path=None):
             "env": {
                 "GITEA_MCP_CONFIG": config_path or DEFAULT_CONFIG_PATH,
                 "GITEA_MCP_PROFILE": profile_name,
+                "GITEA_CLIENT_MANAGED": "1",
             },
         }
     }
 
 
+
 def keychain_set(item_id, token, account=None, runner=subprocess.run):
     """Store *token* in the macOS keychain under service *item_id*.
 

gitea_mcp_server.py

@@ -839,75 +839,6 @@ def _invalidate_preflight_identity_state() -> None:
     _clear_preflight_capability_state()
 
 
-# #690: session-boundary invalidation record for the most recent cross-profile
-# activation. Surfaced in runtime diagnostics so a formal review run can prove
-# its state was reset by a profile switch and must be fully re-established.
-_PROFILE_SWITCH_INVALIDATION: dict | None = None
-
-
-def _invalidate_review_state_on_profile_switch(
-    before_profile: str,
-    after_profile: str,
-) -> dict:
-    """Invalidate review-critical session state on a profile switch (#690).
-
-    Workflow-load proof, reviewer lease binding, the review decision lock,
-    live namespace health, and preflight identity/capability stamps recorded
-    under the prior profile are contaminated for the new role. Durable state
-    keyed by *either* profile identity is cleared so a reviewer → author →
-    reviewer hop cannot resurrect a stale review run: the full reviewer
-    preflight (gitea_whoami, gitea_load_review_workflow,
-    gitea_resolve_task_capability(review_pr), live head re-pin, lease
-    re-acquire/adopt) must be re-established before any formal verdict.
-    """
-    global _PROFILE_SWITCH_INVALIDATION
-    invalidated: list[str] = []
-
-    _invalidate_preflight_identity_state()
-    invalidated.append("preflight_identity_capability")
-
-    review_workflow_load.clear_review_workflow_load()
-    invalidated.append("review_workflow_load")
-
-    _save_review_decision_lock(None)
-    invalidated.append("review_decision_lock")
-
-    reviewer_pr_lease.clear_session_lease()
-    invalidated.append("reviewer_session_lease")
-
-    if _LIVE_NAMESPACE_HEALTH:
-        _LIVE_NAMESPACE_HEALTH.clear()
-        invalidated.append("live_namespace_health")
-
-    # Durable records keyed by either profile identity must not survive the
-    # switch, or activating author → reviewer → author could revive a stale
-    # review run without re-preflight.
-    for identity in {before_profile, after_profile}:
-        if not identity:
-            continue
-        try:
-            mcp_session_state.clear_state(
-                kind=mcp_session_state.KIND_DECISION_LOCK,
-                profile_identity=identity,
-            )
-            mcp_session_state.clear_state(
-                kind=mcp_session_state.KIND_WORKFLOW_LOAD,
-                profile_identity=identity,
-            )
-        except Exception:
-            pass  # best-effort durable cleanup; in-memory state already reset
-    invalidated.append("durable_profile_state")
-
-    _PROFILE_SWITCH_INVALIDATION = {
-        "from_profile": before_profile,
-        "to_profile": after_profile,
-        "invalidated": invalidated,
-        "invalidated_at": datetime.now(timezone.utc).isoformat(),
-        "re_preflight_required": True,
-    }
-    return dict(_PROFILE_SWITCH_INVALIDATION)
-
-
 def record_preflight_check(
     type_name: str,
     resolved_role: str | None = None,
@@ -1615,6 +1546,7 @@ def verify_preflight_purity(
             task=task,
             target_issue_number=target_issue_number,
             require_author_lock=require_author_lock,
+            bootstrap_assessment=bootstrap_assessment,
         )
         # #604: common anti-stomp preflight after legacy + #683 enforcers.
         _run_anti_stomp_preflight(
@@ -1654,6 +1586,7 @@ def verify_preflight_purity(
             task=task,
             target_issue_number=target_issue_number,
             require_author_lock=require_author_lock,
+            bootstrap_assessment=bootstrap_assessment,
         )
         if force_anti_stomp:
             _run_anti_stomp_preflight(
@@ -1721,8 +1654,15 @@ def _enforce_issue_scope_guard(
     task: str | None = None,
     target_issue_number: int | None = None,
     require_author_lock: bool = False,
+    bootstrap_assessment: object = _BOOTSTRAP_UNSET,
 ) -> None:
-    """#683: fail closed on missing/out-of-scope issue ownership for mutations."""
+    """#683: fail closed on missing/out-of-scope issue ownership for mutations.
+
+    #941: the shared bootstrap assessment is threaded in so this guard judges
+    the author issue-worktree bootstrap on the same server-derived evidence as
+    the #274 branches-only and #604 anti-stomp guards. Callers that supply
+    none fall back to computing it here, which preserves behaviour.
+    """
     ctx = _resolve_namespace_mutation_context(worktree_path)
     workspace = ctx["workspace_path"]
     git_state = issue_lock_worktree.read_worktree_git_state(workspace)
@@ -1772,11 +1712,32 @@ def _enforce_issue_scope_guard(
     import create_issue_bootstrap as _cib
 
     is_create_issue = _cib.is_create_issue_task(task)
+    # #941: consume the caller-computed bootstrap assessment when one was
+    # threaded in, so this guard and the #274/#604 guards judge identical
+    # evidence. Falling back preserves behaviour for callers that supply none.
+    bootstrap = (
+        _create_issue_bootstrap_assessment(task, worktree_path)
+        if bootstrap_assessment is _BOOTSTRAP_UNSET
+        else bootstrap_assessment
+    )
+    # #941: the author issue-worktree bootstrap is pre-ownership for the same
+    # reason create_issue is — it exists to break the lock<->worktree cycle, so
+    # no owning lock can exist yet. The exemption is granted by the canonical
+    # shared decision over server-derived evidence, never by a task-name list,
+    # and fails closed on missing, malformed, cross-scope, dirty, drifted, or
+    # wrongly bound evidence.
+    bootstrap_waives_ownership = _cib.bootstrap_permits_control_checkout(
+        bootstrap,
+        task=task,
+        workspace_path=workspace,
+        canonical_repo_root=ctx["canonical_repo_root"],
+    )
     require_lock = bool(require_author_lock) or (
         authorish
         and workflow_scope_guard.production_guards_forced()
         and role == "author"
         and not is_create_issue
+        and not bootstrap_waives_ownership
     )
     assessment = workflow_scope_guard.assess_production_mutation_guards(
         workspace_path=workspace,
@@ -1789,6 +1750,7 @@ def _enforce_issue_scope_guard(
         require_author_lock=require_lock,
         in_test_mode=_preflight_in_test_mode(),
         mutation_task=task,
+        bootstrap_assessment=bootstrap,
     )
     workflow_scope_guard.raise_if_blocked(assessment)
 
@@ -2819,6 +2781,87 @@ def _collect_issue_duplicate_context(
     return issue_duplicate_context_fetcher(h, o, r, auth, issue_number)
 
 
+# Every field of the owning-PR continuation token is security relevant: the
+# issue and PR it names, the branch it is scoped to, and each head the waiver
+# was measured against. Two evidence blocks that disagree on any of them cannot
+# both describe the single sanctioned decision the lock is supposed to record.
+_CONTINUATION_EVIDENCE_BINDINGS = (
+    "issue_number",
+    "pr_number",
+    "branch_name",
+    "head_sha",
+    "recorded_head",
+    "accepted_head",
+    "head_relation",
+)
+
+
+def _continuation_evidence_agrees(recovered: dict, renewed: dict) -> bool:
+    """Do two rebuilt continuation tokens bind to exactly the same thing (#945)?"""
+    return all(
+        recovered.get(field) == renewed.get(field)
+        for field in _CONTINUATION_EVIDENCE_BINDINGS
+    )
+
+
+def _owning_pr_continuation_from_lock(lock_record: dict | None) -> dict | None:
+    """Owning-PR continuation evidence a persisted lock still proves (#945).
+
+    ``gitea_lock_issue`` grants the duplicate-work waiver from either a
+    sanctioned dead-session recovery (#755) or a sanctioned exact-owner renewal
+    (#760), in that precedence. Every later enforcement path — commit,
+    create-PR, push-ownership, and the read-only duplicate assessor — re-derives
+    ownership from the durable lock instead of that live assessment.
+
+    Until #945 only the recovery half was rebuilt there, so an ordinary
+    exact-owner renewal lost its waiver the moment ``gitea_lock_issue``
+    returned: the author renewed successfully and was then refused
+    ``duplicate_commit_prevented`` with ``owning_pr_recovery_exempted: false``
+    on the very PR the renewal had just proved it owned.
+
+    Resolving both halves here, in the same precedence the lock path applies,
+    keeps the answer from drifting between the gate that grants the waiver and
+    the gates that enforce it. This only decides which server-written block the
+    token is rebuilt from — the token is still re-validated against live PR
+    state by ``issue_work_duplicate_gate._assess_owning_pr_exemption``, which
+    remains the single authoritative policy for whether an exemption applies.
+
+    **Both blocks present is a reachable, legitimate state, and it must agree.**
+    The two dispositions are not mutually exclusive at the writer. Recovery is
+    assessed whenever the lease is not live and requires the recorded PID to be
+    dead; renewal is assessed whenever the lease has *expired* — which is itself
+    one way to be non-live — and deliberately does not branch on PID liveness
+    (#760 AC16). An expired lease whose recorded owner has also died therefore
+    satisfies both, and ``gitea_lock_issue`` writes ``dead_session_recovery``
+    and ``lease_renewal`` into the same freshly built ``data`` dict. Because a
+    sanctioned pair was derived from one live observation in one call, it always
+    describes the same issue, PR, branch and head. Disagreement means the
+    persisted lock is no longer a faithful record of a single sanctioned
+    decision, so no continuation authority is returned.
+
+    Ambiguity never broadens authority. A ``dead_session_recovery`` block that
+    is present but does not rebuild — conflicting, stale, malformed, or only
+    partially valid — fails closed here rather than falling through to renewal:
+    otherwise a recovery record naming one PR could be bypassed by valid-looking
+    renewal evidence naming another. Recovery-only and renewal-only locks keep
+    their existing behaviour exactly, and provenance stays server-controlled —
+    this still only ever re-reads blocks the server itself wrote.
+    """
+    if not lock_record:
+        return None
+    recovery_present = isinstance(lock_record.get("dead_session_recovery"), dict)
+    recovered = issue_lock_recovery.recovered_owning_pr_from_lock(lock_record)
+    # Present but unusable recovery evidence is ambiguous, not absent.
+    if recovery_present and not recovered:
+        return None
+    renewed = issue_lock_renewal.owning_pr_renewal_from_lock(lock_record)
+    if recovered and renewed and not _continuation_evidence_agrees(recovered, renewed):
+        return None
+    if recovered:
+        return recovered
+    return renewed
+
+
 def _assess_issue_duplicate_gate(
     issue_number: int,
     *,
@@ -2871,6 +2914,11 @@ def _enforce_locked_issue_duplicate_recheck(
     commit and create-PR phases run in their own calls, long after the recovery
     assessment ended, so without this they re-block the very PR the recovery
     already proved belongs to this author.
+
+    #945: an exact-owner *renewal* (#760) owns its open PR for exactly the same
+    reason, and ``gitea_lock_issue`` already waives the blocker for both. Both
+    halves are resolved together here so the renewal waiver survives past the
+    lock call instead of expiring with it.
     """
     lock_data = _load_existing_issue_lock()
     if not lock_data:
@@ -2894,9 +2942,7 @@ def _enforce_locked_issue_duplicate_recheck(
         auth=auth,
         locked_branch=locked_branch,
         phase=phase,
-        recovered_owning_pr=issue_lock_recovery.recovered_owning_pr_from_lock(
-            lock_data
-        ),
+        recovered_owning_pr=_owning_pr_continuation_from_lock(lock_data),
     )
     if gate.get("block"):
         return gate
@@ -5178,9 +5224,10 @@ def gitea_assess_work_issue_duplicate(
     recovered_owning_pr = None
     lock_data = _load_existing_issue_lock()
     if lock_data and int(lock_data.get("issue_number") or 0) == int(issue_number):
-        recovered_owning_pr = issue_lock_recovery.recovered_owning_pr_from_lock(
-            lock_data
-        )
+        # #945: rebuilt from a sanctioned recovery *or* a sanctioned exact-owner
+        # renewal, so this read-only assessor reports the same disposition the
+        # commit and create-PR gates will enforce.
+        recovered_owning_pr = _owning_pr_continuation_from_lock(lock_data)
     gate = _assess_issue_duplicate_gate(
         issue_number,
         h=h,
@@ -14654,6 +14701,56 @@ def _session_context_mutation_block(
     return blocked
 
 
+def _is_client_managed_process() -> bool:
+    """Check whether the current MCP server process has client-managed launch provenance (#686)."""
+    val = (
+        os.environ.get("GITEA_CLIENT_MANAGED")
+        or os.environ.get("GITEA_MCP_CLIENT_MANAGED")
+        or os.environ.get("GITEA_SERVER_PROVENANCE")
+        or os.environ.get("GITEA_FORCE_CLIENT_MANAGED")
+        or ""
+    ).strip().lower()
+
+    if val in ("0", "false", "no", "manual", "manual_launch"):
+        return False
+
+    if val in ("1", "true", "yes", "client_managed"):
+        return True
+
+    # A terminal launch has an active TTY on stdin
+    try:
+        if sys.stdin and sys.stdin.isatty():
+            return False
+    except Exception:
+        pass
+
+    # Standard client launch or test runner with stdio pipe and profile env
+    if "GITEA_MCP_CONFIG" in os.environ or "GITEA_MCP_PROFILE" in os.environ or "GITEA_PROFILE_NAME" in os.environ:
+        return True
+
+    return False
+
+
+def _provenance_mutation_block(**extra_fields) -> dict | None:
+    """Refuse mutating tool calls on processes lacking client-managed launch provenance (#686)."""
+    if _is_client_managed_process():
+        return None
+    unconsumed = gitea_config.get_unconsumed_gitea_env_overrides()
+    blocked = {
+        "success": False,
+        "performed": False,
+        "blocker_kind": "unsupported_manual_launch",
+        "reasons": [
+            "mutation denied: server process was launched manually from a terminal without client-managed provenance (fail closed). Manually launched mcp_server.py processes cannot receive IDE stdio or serve workflow mutations."
+        ],
+        "exact_next_action": "BLOCKED + RECONNECT: Reconnect the IDE/client-managed MCP server namespace instead of an ad hoc terminal launch. Hand-launched processes and mcp_config.json hand-edits are classified as workflow contamination.",
+        "provenance": "manual_launch",
+        "unconsumed_gitea_env": unconsumed,
+    }
+    blocked.update(extra_fields)
+    return blocked
+
+
 def _profile_permission_block(required_operation: str, **extra_fields) -> dict | None:
     """Structured operation-gate denial for gated tools (#69, #142, #897).
 
@@ -14670,6 +14767,10 @@ def _profile_permission_block(required_operation: str, **extra_fields) -> dict |
     # #714: evaluate active profile only — never auto-switch.
     _ensure_matching_profile(required_operation, req_role, extra_fields.get("remote"))
 
+    prov_block = _provenance_mutation_block(**extra_fields)
+    if prov_block is not None:
+        return prov_block
+
     reasons = _profile_operation_gate(required_operation)
     if reasons:
         return _build_operation_gate_refusal(
@@ -14703,6 +14804,10 @@ def _namespace_mutation_block(mutation_task: str, **extra_fields) -> dict | None
     # #714: evaluate active profile only — never auto-switch.
     _ensure_matching_profile(required_permission, required_role, extra_fields.get("remote"))
 
+    prov_block = _provenance_mutation_block(**extra_fields)
+    if prov_block is not None:
+        return prov_block
+
     try:
         profile = get_profile()
     except Exception as exc:
@@ -17279,11 +17384,6 @@ def gitea_whoami(
         "session_context_audit": session_ctx.mutation_context_audit_fields(),
         "identity_match": not id_match.get("block"),
         "identity_match_reasons": id_match.get("reasons") or [],
-        # #690 AC4: report launcher-declared client namespace alongside the
-        # active execution profile so drift is visible in diagnostics.
-        "namespace_provenance": mcp_namespace_health.namespace_provenance(
-            active_profile=profile["profile_name"]
-        ),
     }
     if id_match.get("block"):
         _invalidate_preflight_identity_state()
@@ -18155,6 +18255,9 @@ def gitea_get_runtime_context(
         source="gitea_get_runtime_context",
     )
 
+    is_client_managed = _is_client_managed_process()
+    unconsumed_env = gitea_config.get_unconsumed_gitea_env_overrides()
+
     result = {
         "active_profile": profile["profile_name"],
         "authenticated_username": username,
@@ -18171,6 +18274,9 @@ def gitea_get_runtime_context(
         "review_merge_blocked_reasons": blocked_reasons,
         "suggested_fix": suggested_fix,
         "safe_next_action": safe_next_action,
+        "server_provenance": "client_managed" if is_client_managed else "manual_launch",
+        "is_client_managed": is_client_managed,
+        "unconsumed_gitea_env": unconsumed_env,
         "preflight_ready": preflight["preflight_ready"],
         "preflight_block_reasons": preflight["preflight_block_reasons"],
         "preflight_workspace": preflight.get("preflight_workspace"),
@@ -18182,13 +18288,15 @@ def gitea_get_runtime_context(
         "shell_health": native_mcp_preference.shell_health_status(),
         "workflow_load_proof": review_workflow_load.workflow_load_status(
             PROJECT_ROOT),
-        # #690: namespace provenance + profile-switch invalidation evidence.
-        "namespace_provenance": mcp_namespace_health.namespace_provenance(
-            active_profile=profile["profile_name"]
-        ),
-        "profile_switch_invalidation": _PROFILE_SWITCH_INVALIDATION,
     }
 
+    if not is_client_managed:
+        result["safe_next_action"] = (
+            "BLOCKED + RECONNECT: Serving process lacks client-managed launch provenance (manual launch). "
+            "Reconnect the IDE/client-managed MCP server namespace instead of an ad hoc terminal launch."
+        )
+
+
     # #702: read-only visibility into the inherited GITEA_ACTIVE_WORKTREE
     # binding; recovery itself runs during capability resolution.
     try:
@@ -18690,17 +18798,6 @@ def gitea_activate_profile(
         source="gitea_activate_profile",
     )
 
-    # 4.7 #690: a profile switch is a session-boundary event for review state.
-    # Any workflow-load proof, reviewer lease, decision lock, namespace
-    # health, or preflight stamp recorded under the prior profile is
-    # contaminated for the new role and must be re-established under the new
-    # profile before any formal review verdict.
-    switch_invalidation = None
-    if before_profile != after_profile:
-        switch_invalidation = _invalidate_review_state_on_profile_switch(
-            before_profile, after_profile
-        )
-
     # 5. Audit the switch if auditing is on
     _audit(
         "activate_profile",
@@ -18711,12 +18808,11 @@ def gitea_activate_profile(
             "before": before_profile,
             "after": after_profile,
             "session_context": session_ctx.mutation_context_audit_fields(),
-            "review_state_invalidated": bool(switch_invalidation),
         },
         username=after_identity,
     )
 
-    result = {
+    return {
         "success": True,
         "message": f"Successfully activated profile '{profile_name}' (fresh identity verification complete).",
         "before_profile": before_profile,
@@ -18726,18 +18822,6 @@ def gitea_activate_profile(
         "session_context_audit": session_ctx.mutation_context_audit_fields(),
         "auto_profile_substitution": False,
     }
-    if switch_invalidation is not None:
-        result["review_state_invalidation"] = switch_invalidation
-        result["re_preflight_required"] = True
-        result["exact_next_action"] = (
-            "Profile switch invalidated workflow-load proof, reviewer lease, "
-            "decision lock, and preflight stamps (#690). Before any formal "
-            "review verdict, re-run the full reviewer preflight: "
-            "gitea_whoami, gitea_load_review_workflow, "
-            "gitea_resolve_task_capability(review_pr), live head re-pin, and "
-            "lease re-acquire/adopt."
-        )
-    return result
 
 
 @mcp.tool()
@@ -19425,7 +19509,10 @@ def _prove_author_ownership_for_pr(
     # advance is the one recovery already sanctioned, not a foreign head.
     recovered_owning_pr = None
     if proven and lock_record:
-        candidate = issue_lock_recovery.recovered_owning_pr_from_lock(lock_record)
+        # #945: a sanctioned exact-owner renewal proves the same ownership of
+        # the same PR, so the push gate resolves both halves rather than seeing
+        # only the recovery one.
+        candidate = _owning_pr_continuation_from_lock(lock_record)
         if candidate and int(candidate.get("pr_number") or 0) == int(pr_number):
             recovered_owning_pr = candidate
     return {
@@ -20670,7 +20757,9 @@ def _check_mcp_runtimes_diagnostics(task: str, matching_profiles: list[str]) ->
     self_pid = os.getpid()
     self_stale = False
 
-    running_profiles = {}
+    all_profile_procs: dict[str, list[dict]] = {}
+    unsupported_env_found = set()
+
     for line in proc.stdout.splitlines()[1:]:
         line = line.strip()
         if not line or "mcp_server.py" not in line:
@@ -20702,16 +20791,55 @@ def _check_mcp_runtimes_diagnostics(task: str, matching_profiles: list[str]) ->
         if match:
             profile = match.group(1)
 
+        is_client_managed = bool(
+            re.search(r'\bGITEA_CLIENT_MANAGED=(1|true|yes|client_managed)\b', env_out, re.IGNORECASE)
+            or re.search(r'\bGITEA_MCP_CLIENT_MANAGED=(1|true|yes|client_managed)\b', env_out, re.IGNORECASE)
+            or re.search(r'\bGITEA_SERVER_PROVENANCE=client_managed\b', env_out, re.IGNORECASE)
+        )
+
+        for env_match in re.finditer(r'\b(GITEA_[A-Z0-9_]+)=([^\s]+)', env_out):
+            k, v = env_match.group(1), env_match.group(2)
+            if k not in gitea_config.RECOGNIZED_GITEA_ENV_KEYS and not any(k.startswith(p) for p in gitea_config.RECOGNIZED_GITEA_ENV_PREFIXES):
+                unsupported_env_found.add(f"{k}={v}")
+
         is_stale = (start_time < code_mtime) or git_stale
         if pid == self_pid and is_stale:
             self_stale = True
 
-        if profile not in running_profiles or start_time > running_profiles[profile]["start_time"]:
-            running_profiles[profile] = {
-                "pid": pid,
-                "start_time": start_time,
-                "is_stale": is_stale
-            }
+        proc_info = {
+            "pid": pid,
+            "start_time": start_time,
+            "is_stale": is_stale,
+            "is_client_managed": is_client_managed,
+        }
+        if profile not in all_profile_procs:
+            all_profile_procs[profile] = []
+        all_profile_procs[profile].append(proc_info)
+
+    running_profiles = {}
+    for profile, procs in all_profile_procs.items():
+        if len(procs) > 1:
+            pids_str = ", ".join(str(p["pid"]) for p in procs)
+            reasons.append(
+                f"stale-runtime: Duplicate MCP server process(es) detected for profile '{profile}' (PIDs: {pids_str}). "
+                "Manual or duplicate launches defeat staleness detection and cannot receive client stdio."
+            )
+        client_procs = [p for p in procs if p["is_client_managed"]]
+        if client_procs:
+            client_procs.sort(key=lambda p: p["start_time"], reverse=True)
+            running_profiles[profile] = client_procs[0]
+        else:
+            pids_str = ", ".join(str(p["pid"]) for p in procs)
+            reasons.append(
+                f"stale-runtime: Manually launched MCP process(es) detected without client-managed provenance for profile '{profile}' (PIDs: {pids_str}). "
+                "Manual launches cannot serve client stdio and are ignored for runtime freshness."
+            )
+
+    if unsupported_env_found:
+        reasons.append(
+            f"unsupported-env: Unsupported GITEA_* environment variable override(s) detected: {', '.join(sorted(unsupported_env_found))}. "
+            "Unknown env overrides are unsupported."
+        )
 
     if self_stale:
         # #685: report-only — no config utime, no thread, no os._exit.
@@ -20746,6 +20874,7 @@ def _check_mcp_runtimes_diagnostics(task: str, matching_profiles: list[str]) ->
     return reasons
 
 
+
 @mcp.tool()
 def gitea_resolve_task_capability(
     task: str,
@@ -20982,22 +21111,12 @@ def gitea_resolve_task_capability(
             f"{required_role} task '{task}' even if nearby permissions are "
             "present (fail closed)."
         )
-    # #690 AC4: when the launcher declares a client namespace, a task with a
-    # required namespace must fail closed on mismatch (e.g. review_pr served
-    # from an author namespace).
-    ns_provenance = mcp_namespace_health.namespace_provenance(
-        task=task_key, active_profile=profile.get("profile_name")
-    )
-    ns_mismatch_reason = None
-    if ns_provenance.get("mismatch"):
-        ns_mismatch_reason = "; ".join(ns_provenance.get("reasons") or [])
     cross_host_block = bool(remote_assess.get("block"))
     identity_block = bool(id_assess.get("block"))
     drift_block = bool(ctx_assess.get("block"))
     allowed_in_current_session = (
         permission_allowed_in_current_session
         and role_matches_current_session
-        and not ns_provenance.get("mismatch")
         and not cross_host_block
         and not identity_block
         and not drift_block
@@ -21048,8 +21167,6 @@ def gitea_resolve_task_capability(
             )
         if role_mismatch_reason:
             deny_parts.append(role_mismatch_reason)
-        if ns_mismatch_reason:
-            deny_parts.append(ns_mismatch_reason)
         if deny_parts:
             reason_msg = "; ".join(deny_parts)
         elif configured and switching:
@@ -21127,8 +21244,6 @@ def gitea_resolve_task_capability(
     task_role_guidance = []
     if role_mismatch_reason:
         task_role_guidance.append(f"STOP: {role_mismatch_reason}")
-    if ns_mismatch_reason:
-        task_role_guidance.append(f"STOP: {ns_mismatch_reason}")
     if required_role == "reviewer":
         if allowed_in_current_session:
             task_role_guidance.append(
@@ -21195,7 +21310,6 @@ def gitea_resolve_task_capability(
         "session_context_audit": session_ctx.mutation_context_audit_fields(),
         "profile_remote_compatible": not cross_host_block,
         "identity_match": not identity_block,
-        "namespace_provenance": ns_provenance,
         "auto_profile_substitution": False,
     }
     # #685: report typed reconnect blocker without mutating config or exiting.

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,
     *,

mcp_namespace_health.py

@@ -16,7 +16,6 @@ Probe sources
 
 from __future__ import annotations
 
-import os
 from typing import Any
 
 
@@ -58,56 +57,8 @@ SAFE_ENV_KEYS = (
     "GITEA_SERVICE",
     "GITEA_EXECUTION_ROLE",
     "GITEA_MCP_CONFIG",
-    "GITEA_MCP_NAMESPACE",
 )
 
-# Optional launcher-provided env declaring the client-managed MCP namespace
-# this process is registered under (e.g. ``gitea-reviewer``). The server
-# cannot derive its own IDE namespace name, so the launcher declares it; when
-# declared, reviewers/mergers can fail closed on a namespace/task mismatch
-# (#690 AC4). Absence means "unknown" — reported, never guessed.
-NAMESPACE_ENV = "GITEA_MCP_NAMESPACE"
-
-
-def configured_client_namespace(env: dict[str, str] | None = None) -> str | None:
-    """Return the launcher-declared client namespace, or None when unknown."""
-    source = os.environ if env is None else env
-    value = (source.get(NAMESPACE_ENV) or "").strip()
-    return value or None
-
-
-def namespace_provenance(
-    task: str | None = None,
-    *,
-    active_profile: str | None = None,
-    env: dict[str, str] | None = None,
-) -> dict[str, Any]:
-    """Report configured client namespace vs active execution profile (#690).
-
-    When *task* carries a required namespace (``TASK_REQUIRED_NAMESPACES``)
-    and the launcher declared a different one, ``mismatch`` is True and the
-    caller must fail closed for that task. An undeclared namespace is
-    reported as unknown — never treated as proof either way.
-    """
-    configured = configured_client_namespace(env)
-    required = TASK_REQUIRED_NAMESPACES.get(task or "")
-    mismatch = bool(configured and required and configured != required)
-    reasons: list[str] = []
-    if mismatch:
-        reasons.append(
-            f"configured client namespace '{configured}' does not match "
-            f"required namespace '{required}' for task '{task}' (fail closed)"
-        )
-    return {
-        "configured_namespace": configured,
-        "namespace_source": NAMESPACE_ENV if configured else "unknown",
-        "active_profile": active_profile,
-        "requested_task": task,
-        "required_namespace": required,
-        "mismatch": mismatch,
-        "reasons": reasons,
-    }
-
 
 def _as_list(value: Any) -> list[str] | None:
     if value is None:
@@ -274,6 +225,16 @@ def classify_namespace_probe(
         # on bad data without treating success as IDE proof).
         blocks = namespace_health_blocks_task("merge_pr", healthy)
 
+    import gitea_config
+    raw_env = process.get("env") if isinstance(process, dict) else None
+    unconsumed_env = gitea_config.get_unconsumed_gitea_env_overrides(raw_env)
+    is_client_managed = bool(
+        env_summary.get("GITEA_CLIENT_MANAGED") in ("1", "true", "yes", "client_managed")
+        or env_summary.get("GITEA_MCP_CLIENT_MANAGED") in ("1", "true", "yes", "client_managed")
+        or env_summary.get("GITEA_SERVER_PROVENANCE") == "client_managed"
+    )
+    provenance = "client_managed" if is_client_managed else "manual_launch"
+
     return {
         "success": healthy,
         "healthy": healthy,
@@ -289,6 +250,9 @@ def classify_namespace_probe(
         "error_message": error_message or None,
         "reasons": reasons,
         "remediation": remediation,
+        "provenance": provenance,
+        "is_client_managed": is_client_managed,
+        "unconsumed_gitea_env": unconsumed_env,
         "diagnostics": {
             "namespace": ns,
             "required_tool": tool,
@@ -297,6 +261,9 @@ def classify_namespace_probe(
             "env": env_summary,
             "config_path": config_path,
             "probe_source": source,
+            "provenance": provenance,
+            "is_client_managed": is_client_managed,
+            "unconsumed_gitea_env": unconsumed_env,
         },
         "blocks_merge_workflow": blocks,
     }

tests/conftest.py

@@ -41,10 +41,11 @@ def _reset_mutation_authority(monkeypatch):
         "GITEA_REVIEWER_WORKTREE",
         "GITEA_MERGER_WORKTREE",
         "GITEA_RECONCILER_WORKTREE",
-        "GITEA_MCP_NAMESPACE",
     ]:
         monkeypatch.delenv(env_key, raising=False)
 
+    monkeypatch.setenv("GITEA_CLIENT_MANAGED", "1")
+
     # Isolate durable session-state files so tests never share host cache (#559).
     import tempfile
 
@@ -116,7 +117,6 @@ def _reset_mutation_authority(monkeypatch):
         monkeypatch.setattr(mcp_server, "_ACTOR_IDENTITY_CACHE", {})
     monkeypatch.setattr(mcp_server, "_REVIEW_DECISION_LOCK", None)
     monkeypatch.setattr(mcp_server, "_LIVE_NAMESPACE_HEALTH", {})
-    monkeypatch.setattr(mcp_server, "_PROFILE_SWITCH_INVALIDATION", None)
     monkeypatch.setattr(mcp_server, "_preflight_whoami_called", False)
     monkeypatch.setattr(mcp_server, "_preflight_capability_called", False)
     monkeypatch.setattr(mcp_server, "_preflight_resolved_role", None)

tests/test_commit_files_gate.py

@@ -35,7 +35,7 @@ CONFIG = {
             ],
             "forbidden_operations": [],
             "execution_profile": "full-author",
-        "allowed_repositories": ["Example-Org/Example-Repo"],
+        "allowed_repositories": ["Scaled-Tech-Consulting/Gitea-Tools", "Example-Org/Example-Repo"],
         },
         "reviewer-no-commit": {
             "enabled": True,
@@ -50,7 +50,7 @@ CONFIG = {
                 "gitea.repo.commit", "gitea.pr.create", "gitea.branch.push"
             ],
             "execution_profile": "reviewer-no-commit",
-        "allowed_repositories": ["Example-Org/Example-Repo"],
+        "allowed_repositories": ["Scaled-Tech-Consulting/Gitea-Tools", "Example-Org/Example-Repo"],
         },
     },
     "rules": {"allow_runtime_switching": False},

tests/test_config_menu.py

@@ -175,7 +175,7 @@ class TestLauncherSnippets(unittest.TestCase):
     def test_only_safe_keys_no_secrets(self):
         entry = gitea_config.launcher_entry("prgs", "/cfg/profiles.json")["gitea-tools"]
         self.assertEqual(set(entry), {"command", "args", "env"})
-        self.assertEqual(set(entry["env"]), {"GITEA_MCP_CONFIG", "GITEA_MCP_PROFILE"})
+        self.assertEqual(set(entry["env"]), {"GITEA_MCP_CONFIG", "GITEA_MCP_PROFILE", "GITEA_CLIENT_MANAGED"})
         self.assertEqual(entry["env"]["GITEA_MCP_PROFILE"], "prgs")
         blob = json.dumps(entry).lower()
         for word in ("token", "password", "secret"):

tests/test_issue_686_manual_mcp_provenance.py

@@ -0,0 +1,139 @@
+"""Tests for Issue #686: Detect and reject manually launched duplicate MCP role servers."""
+import os
+import unittest
+from unittest.mock import patch, MagicMock
+from datetime import datetime
+
+import gitea_config
+import gitea_mcp_server
+import mcp_namespace_health
+
+
+class TestIssue686ManualMcpProvenance(unittest.TestCase):
+
+    def test_client_managed_process_detection(self):
+        """Test _is_client_managed_process correctly detects provenance markers."""
+        with patch.dict(os.environ, {"GITEA_CLIENT_MANAGED": "1"}, clear=True):
+            self.assertTrue(gitea_mcp_server._is_client_managed_process())
+
+        with patch.dict(os.environ, {"GITEA_MCP_CLIENT_MANAGED": "true"}, clear=True):
+            self.assertTrue(gitea_mcp_server._is_client_managed_process())
+
+        with patch.dict(os.environ, {"GITEA_SERVER_PROVENANCE": "client_managed"}, clear=True):
+            self.assertTrue(gitea_mcp_server._is_client_managed_process())
+
+        with patch.dict(os.environ, {"GITEA_CLIENT_MANAGED": "0"}, clear=True):
+            self.assertFalse(gitea_mcp_server._is_client_managed_process())
+
+    def test_unconsumed_gitea_env_overrides(self):
+        """Test surfacing of unsupported GITEA_* env overrides (e.g. GITEA_DUMMY)."""
+        env = {
+            "GITEA_MCP_PROFILE": "prgs-author",
+            "GITEA_CLIENT_MANAGED": "1",
+            "GITEA_DUMMY": "2",
+            "GITEA_UNKNOWN_FLAG": "abc",
+        }
+        unconsumed = gitea_config.get_unconsumed_gitea_env_overrides(env)
+        self.assertIn("GITEA_DUMMY", unconsumed)
+        self.assertEqual(unconsumed["GITEA_DUMMY"], "2")
+        self.assertIn("GITEA_UNKNOWN_FLAG", unconsumed)
+        self.assertNotIn("GITEA_MCP_PROFILE", unconsumed)
+        self.assertNotIn("GITEA_CLIENT_MANAGED", unconsumed)
+
+    def test_manual_server_mutation_fail_closed(self):
+        """AC 2: Mutating tools on a server without client-managed provenance fail closed with a typed blocker."""
+        with patch.dict(os.environ, {"GITEA_CLIENT_MANAGED": "0"}, clear=True):
+            block = gitea_mcp_server._provenance_mutation_block(task="create_issue")
+            self.assertIsNotNone(block)
+            self.assertFalse(block["success"])
+            self.assertFalse(block["performed"])
+            self.assertEqual(block["blocker_kind"], "unsupported_manual_launch")
+            self.assertEqual(block["provenance"], "manual_launch")
+            self.assertTrue(any("mutation denied: server process was launched manually" in r for r in block["reasons"]))
+            self.assertIn("BLOCKED + RECONNECT", block["exact_next_action"])
+
+    def test_client_managed_server_mutation_passes_provenance_gate(self):
+        """AC 3: Clean client-managed baseline passes the provenance gate."""
+        with patch.dict(os.environ, {"GITEA_CLIENT_MANAGED": "1"}, clear=True):
+            block = gitea_mcp_server._provenance_mutation_block(task="create_issue")
+            self.assertIsNone(block)
+
+    @patch("subprocess.run")
+    @patch("os.path.getmtime")
+    @patch("os.path.exists")
+    @patch("os.getpid")
+    def test_manual_duplicate_does_not_mask_stale_runtime(
+        self, mock_getpid, mock_exists, mock_getmtime, mock_run
+    ):
+        """AC 1 & AC 3: Staleness detection ignores manual duplicates and reports stale supported runtimes."""
+        mock_getpid.return_value = 12345
+        mock_exists.return_value = True
+
+        code_time = datetime(2026, 7, 8, 14, 0, 0)
+        mock_getmtime.return_value = code_time.timestamp()
+
+        # PID 12345: stale client-managed process (started at 13:00)
+        # PID 99999: fresh manual duplicate process (started at 15:00, no GITEA_CLIENT_MANAGED)
+        ps_output = (
+            "  PID LSTART COMMAND\n"
+            "12345 Wed Jul  8 13:00:00 2026 /path/to/python mcp_server.py\n"
+            "99999 Wed Jul  8 15:00:00 2026 /path/to/python mcp_server.py\n"
+        )
+
+        mock_run_ps = MagicMock()
+        mock_run_ps.stdout = ps_output
+
+        mock_env_12345 = MagicMock()
+        mock_env_12345.stdout = "GITEA_MCP_PROFILE=prgs-author GITEA_CLIENT_MANAGED=1"
+
+        mock_env_99999 = MagicMock()
+        mock_env_99999.stdout = "GITEA_MCP_PROFILE=prgs-author GITEA_DUMMY=2"
+
+        def side_effect(args, **kwargs):
+            if args[0] == "ps" and "eww" in args:
+                pid = args[2]
+                if pid == "12345":
+                    return mock_env_12345
+                elif pid == "99999":
+                    return mock_env_99999
+            elif args[0] == "ps":
+                return mock_run_ps
+            raise ValueError(f"Unexpected args: {args}")
+
+        mock_run.side_effect = side_effect
+
+        reasons = gitea_mcp_server._check_mcp_runtimes_diagnostics("create_issue", ["prgs-author"])
+
+        # Manual duplicate process must be flagged
+        self.assertTrue(any("Duplicate MCP server process(es) detected" in r for r in reasons))
+        # Unsupported env override (GITEA_DUMMY=2) must be flagged
+        self.assertTrue(any("unsupported-env: Unsupported GITEA_* environment variable override(s) detected: GITEA_DUMMY=2" in r for r in reasons))
+        # Stale runtime must NOT be masked by fresh manual process 99999!
+        self.assertTrue(any("All matching profiles for task 'create_issue' (['prgs-author']) are running but stale" in r for r in reasons))
+
+    def test_namespace_health_classification_includes_provenance(self):
+        """AC 1 & 4: mcp_namespace_health diagnostics include provenance and unconsumed_gitea_env."""
+        process = {
+            "pid": 5555,
+            "profile": "prgs-author",
+            "env": {
+                "GITEA_MCP_PROFILE": "prgs-author",
+                "GITEA_DUMMY": "99",
+            },
+        }
+        res = mcp_namespace_health.classify_namespace_probe(
+            "gitea-author",
+            configured=True,
+            registered_tools=["gitea_whoami"],
+            probe_result={"success": True},
+            process=process,
+            probe_source="client_namespace",
+        )
+        self.assertEqual(res["provenance"], "manual_launch")
+        self.assertFalse(res["is_client_managed"])
+        self.assertEqual(res["unconsumed_gitea_env"], {"GITEA_DUMMY": "99"})
+        self.assertEqual(res["diagnostics"]["provenance"], "manual_launch")
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_issue_690_profile_switch_review_guard.py

@@ -1,274 +0,0 @@
-"""Regression coverage for #690: cross-role profile activation invalidation.
-
-A mid-run profile switch (e.g. reviewer → author → reviewer) must invalidate
-workflow-load proof, reviewer lease binding, review decision lock, live
-namespace health, and preflight identity/capability stamps, and must require
-a full reviewer preflight before any formal verdict. Namespace provenance
-must be reported and fail closed on task/namespace mismatch.
-"""
-import json
-import os
-import sys
-import tempfile
-import unittest
-from unittest.mock import patch
-
-sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
-
-import gitea_config
-import mcp_namespace_health
-import mcp_server
-import mcp_session_state
-import review_workflow_load
-import reviewer_pr_lease
-
-from tests.test_runtime_clarity import CONFIG_SWITCHING_ENABLED
-
-
-class TestProfileSwitchReviewGuard(unittest.TestCase):
-    def setUp(self):
-        self._remotes_patch = patch.dict(mcp_server.REMOTES, {
-            "dadeschools": {"host": "gitea.example.com", "org": "Example-Org", "repo": "Example-Repo"},
-            "prgs": {"host": "gitea.example.com", "org": "Example-Org", "repo": "Example-Repo"},
-        })
-        self._remotes_patch.start()
-        mcp_server._IDENTITY_CACHE.clear()
-        gitea_config._active_profile_override = None
-        self._dir = tempfile.TemporaryDirectory()
-        self.config_path = os.path.join(self._dir.name, "profiles.json")
-        with open(self.config_path, "w", encoding="utf-8") as fh:
-            fh.write(json.dumps(CONFIG_SWITCHING_ENABLED))
-
-    def tearDown(self):
-        self._remotes_patch.stop()
-        mcp_server._IDENTITY_CACHE.clear()
-        gitea_config._active_profile_override = None
-        self._dir.cleanup()
-
-    def _env(self, profile="reviewer-profile"):
-        return {
-            "GITEA_MCP_CONFIG": self.config_path,
-            "GITEA_MCP_PROFILE": profile,
-            "GITEA_TOKEN_AUTHOR": "author-pass",
-            "GITEA_TOKEN_REVIEWER": "reviewer-pass",
-            "GITEA_TOKEN_MERGER": "merger-pass",
-        }
-
-    def _seed_contaminated_review_state(self):
-        """Simulate an in-flight reviewer run under reviewer-profile."""
-        mcp_server._preflight_whoami_called = True
-        mcp_server._preflight_capability_called = True
-        mcp_server._preflight_resolved_role = "reviewer"
-        mcp_server._preflight_resolved_task = "review_pr"
-        review_workflow_load._REVIEW_WORKFLOW_LOAD = {"loaded": True}
-        mcp_server._REVIEW_DECISION_LOCK = {
-            "session_profile": "reviewer-profile",
-            "final_review_decision_ready": True,
-            "ready_pr_number": 688,
-        }
-        reviewer_pr_lease.record_session_lease(
-            {"session_id": "lease-session-1", "pr_number": 688}
-        )
-        mcp_server._LIVE_NAMESPACE_HEALTH["gitea-reviewer"] = {
-            "namespace": "gitea-reviewer",
-            "healthy": True,
-            "ide_namespace_proven": True,
-        }
-        # Durable records keyed by the reviewer identity must also be cleared.
-        mcp_session_state.save_state(
-            kind=mcp_session_state.KIND_WORKFLOW_LOAD,
-            payload={"loaded": True},
-            profile_identity="reviewer-profile",
-        )
-        mcp_session_state.save_state(
-            kind=mcp_session_state.KIND_DECISION_LOCK,
-            payload={"final_review_decision_ready": True, "ready_pr_number": 688},
-            profile_identity="reviewer-profile",
-        )
-
-    def _activate(self, target, logins):
-        with patch.object(
-            mcp_server, "get_auth_header", side_effect=[f"token p" for _ in logins]
-        ), patch.object(
-            mcp_server, "api_request", side_effect=[{"login": l} for l in logins]
-        ), patch.object(
-            mcp_server,
-            "_workspace_repository_slug",
-            return_value="Example-Org/Example-Repo",
-        ), patch.object(
-            mcp_server, "_canonical_repository_slug", return_value=(None, [])
-        ):
-            return mcp_server.gitea_activate_profile(profile_name=target)
-
-    # -----------------------------------------------------------------
-    # AC1/AC2/AC3: switch invalidates review state; re-preflight required
-    # -----------------------------------------------------------------
-    def test_switch_invalidates_review_state_and_blocks_verdict(self):
-        with patch.dict(os.environ, self._env("reviewer-profile"), clear=True):
-            self._seed_contaminated_review_state()
-            res = self._activate("author-profile", ["reviewer-user", "author-user"])
-
-            self.assertTrue(res["success"])
-            self.assertTrue(res["re_preflight_required"])
-            inv = res["review_state_invalidation"]
-            self.assertEqual(inv["from_profile"], "reviewer-profile")
-            self.assertEqual(inv["to_profile"], "author-profile")
-            for item in (
-                "preflight_identity_capability",
-                "review_workflow_load",
-                "review_decision_lock",
-                "reviewer_session_lease",
-                "live_namespace_health",
-            ):
-                self.assertIn(item, inv["invalidated"])
-
-            # In-memory state cleared.
-            self.assertFalse(mcp_server._preflight_whoami_called)
-            self.assertFalse(mcp_server._preflight_capability_called)
-            self.assertIsNone(mcp_server._preflight_resolved_task)
-            self.assertIsNone(review_workflow_load._REVIEW_WORKFLOW_LOAD)
-            self.assertIsNone(mcp_server._REVIEW_DECISION_LOCK)
-            self.assertIsNone(reviewer_pr_lease.get_session_lease())
-            self.assertEqual(mcp_server._LIVE_NAMESPACE_HEALTH, {})
-            self.assertIsNotNone(mcp_server._PROFILE_SWITCH_INVALIDATION)
-
-            # Durable records keyed by the reviewer identity are gone.
-            self.assertIsNone(
-                mcp_session_state.load_state(
-                    kind=mcp_session_state.KIND_WORKFLOW_LOAD,
-                    profile_identity="reviewer-profile",
-                )
-            )
-            self.assertIsNone(
-                mcp_session_state.load_state(
-                    kind=mcp_session_state.KIND_DECISION_LOCK,
-                    profile_identity="reviewer-profile",
-                )
-            )
-
-            # A formal verdict without re-preflight fails closed.
-            reasons = mcp_server.check_review_decision_gate(
-                688, "APPROVE", final_review_decision_ready=True
-            )
-            self.assertTrue(reasons)
-
-    def test_switch_back_cannot_resurrect_stale_review_run(self):
-        with patch.dict(os.environ, self._env("reviewer-profile"), clear=True):
-            self._seed_contaminated_review_state()
-            self._activate("author-profile", ["reviewer-user", "author-user"])
-            res = self._activate("reviewer-profile", ["author-user", "reviewer-user"])
-
-            self.assertTrue(res["success"])
-            # The pre-switch review run must not reappear.
-            self.assertIsNone(mcp_server._REVIEW_DECISION_LOCK)
-            self.assertIsNone(review_workflow_load._REVIEW_WORKFLOW_LOAD)
-            self.assertIsNone(reviewer_pr_lease.get_session_lease())
-            status = review_workflow_load.workflow_load_status()
-            self.assertFalse(status["workflow_load_valid"])
-            reasons = mcp_server.check_review_decision_gate(
-                688, "APPROVE", final_review_decision_ready=True
-            )
-            self.assertTrue(reasons)
-
-    def test_same_profile_reactivation_keeps_state(self):
-        with patch.dict(os.environ, self._env("reviewer-profile"), clear=True):
-            self._seed_contaminated_review_state()
-            res = self._activate("reviewer-profile", ["reviewer-user", "reviewer-user"])
-            self.assertTrue(res["success"], res)
-            self.assertNotIn("review_state_invalidation", res)
-            self.assertIsNotNone(mcp_server._REVIEW_DECISION_LOCK)
-            self.assertTrue(mcp_server._preflight_whoami_called)
-
-    def test_clean_repreflight_after_switch_allows_gate(self):
-        with patch.dict(os.environ, self._env("reviewer-profile"), clear=True):
-            self._seed_contaminated_review_state()
-            self._activate("author-profile", ["reviewer-user", "author-user"])
-            self._activate("reviewer-profile", ["author-user", "reviewer-user"])
-
-            # Re-establish the full reviewer preflight under the new profile.
-            mcp_server.record_preflight_check("whoami")
-            mcp_server.record_preflight_check(
-                "capability", resolved_role="reviewer", resolved_task="review_pr"
-            )
-            mcp_server.init_review_decision_lock("dadeschools", "review_pr")
-            lock = mcp_server._load_review_decision_lock()
-            self.assertIsNotNone(lock)
-            lock.update(
-                {
-                    "final_review_decision_ready": True,
-                    "ready_pr_number": 688,
-                    "ready_action": "APPROVE",
-                    "ready_remote": "dadeschools",
-                    "ready_org": "Example-Org",
-                    "ready_repo": "Example-Repo",
-                }
-            )
-            mcp_server._save_review_decision_lock(lock)
-
-            with patch.object(
-                mcp_server, "_review_workflow_load_gate_reasons", return_value=[]
-            ):
-                reasons = mcp_server.check_review_decision_gate(
-                    688,
-                    "APPROVE",
-                    final_review_decision_ready=True,
-                    remote="dadeschools",
-                )
-            self.assertEqual(reasons, [])
-
-    # -----------------------------------------------------------------
-    # AC4: namespace provenance reporting + fail-closed mismatch
-    # -----------------------------------------------------------------
-    def test_namespace_provenance_mismatch_detection(self):
-        prov = mcp_namespace_health.namespace_provenance(
-            task="review_pr",
-            active_profile="reviewer-profile",
-            env={"GITEA_MCP_NAMESPACE": "gitea-author"},
-        )
-        self.assertTrue(prov["mismatch"])
-        self.assertEqual(prov["required_namespace"], "gitea-reviewer")
-
-        prov_ok = mcp_namespace_health.namespace_provenance(
-            task="review_pr",
-            active_profile="reviewer-profile",
-            env={"GITEA_MCP_NAMESPACE": "gitea-reviewer"},
-        )
-        self.assertFalse(prov_ok["mismatch"])
-
-        prov_unknown = mcp_namespace_health.namespace_provenance(
-            task="review_pr", active_profile="reviewer-profile", env={}
-        )
-        self.assertIsNone(prov_unknown["configured_namespace"])
-        self.assertFalse(prov_unknown["mismatch"])
-        self.assertEqual(prov_unknown["namespace_source"], "unknown")
-
-    @patch("mcp_server.api_request", return_value={"login": "reviewer-user"})
-    @patch("mcp_server.get_auth_header", return_value="token reviewer-pass")
-    def test_whoami_reports_namespace_provenance(self, _auth, _api):
-        env = self._env("reviewer-profile")
-        env["GITEA_MCP_NAMESPACE"] = "gitea-reviewer"
-        with patch.dict(os.environ, env, clear=True):
-            res = mcp_server.gitea_whoami(remote="dadeschools")
-        prov = res["namespace_provenance"]
-        self.assertEqual(prov["configured_namespace"], "gitea-reviewer")
-        self.assertEqual(prov["active_profile"], "reviewer-profile")
-        self.assertFalse(prov["mismatch"])
-
-    @patch("mcp_server.api_request", return_value={"login": "reviewer-user"})
-    @patch("mcp_server.get_auth_header", return_value="token reviewer-pass")
-    def test_resolve_fails_closed_on_namespace_mismatch(self, _auth, _api):
-        env = self._env("reviewer-profile")
-        env["GITEA_MCP_NAMESPACE"] = "gitea-author"
-        with patch.dict(os.environ, env, clear=True):
-            res = mcp_server.gitea_resolve_task_capability(
-                task="review_pr", kwargs="{}", remote="dadeschools"
-            )
-        self.assertFalse(res["allowed_in_current_session"])
-        self.assertTrue(res["namespace_provenance"]["mismatch"])
-        self.assertTrue(
-            any("namespace" in g for g in res["task_role_guidance"])
-        )
-
-
-if __name__ == "__main__":
-    unittest.main()

tests/test_issue_892_author_bootstrap_deadlock.py

@@ -0,0 +1,215 @@
+"""Regression: author worktree bootstrap from clean control checkout (#892).
+
+#892 is the four-door deadlock where every documented recovery path is closed:
+bootstrap refuses control, lock demands an existing worktree, worktree-start
+demands a lock, and shell worktree add is outside the sanctioned MCP path.
+
+Root cause: assess_author_issue_bootstrap returned allowed/proven for a clean
+control checkout, but bootstrap_permits_control_checkout only accepted
+create_issue assessments (task_scope=create_issue_only + empty reasons + full
+base-tip field set). Author assessments never satisfied the shared predicate,
+so the #274/#604 guards kept the ordinary control-checkout block.
+"""
+
+from __future__ import annotations
+
+import os
+import tempfile
+import unittest
+from unittest import mock
+
+import author_issue_bootstrap as aib
+import create_issue_bootstrap as cib
+
+
+CONTROL = "/repo/Gitea-Tools"
+MASTER = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
+OTHER = "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"
+
+
+def _assess(
+    *,
+    workspace=CONTROL,
+    root=CONTROL,
+    branch="master",
+    head=MASTER,
+    porcelain="",
+    remote=MASTER,
+    remote_error=None,
+    task="bootstrap_author_issue_worktree",
+):
+    return aib.assess_author_issue_bootstrap(
+        workspace_path=workspace,
+        canonical_repo_root=root,
+        current_branch=branch,
+        head_sha=head,
+        porcelain_status=porcelain,
+        remote_master_sha=remote,
+        remote_master_sha_error=remote_error,
+        task=task,
+    )
+
+
+class TestAuthorBootstrapAssessmentShape(unittest.TestCase):
+    def test_clean_control_emits_predicate_compatible_fields(self):
+        assessment = _assess()
+        self.assertTrue(assessment["allowed"])
+        self.assertTrue(assessment["proven"])
+        self.assertFalse(assessment["block"])
+        self.assertFalse(assessment["not_applicable"])
+        self.assertEqual(assessment["reasons"], [])
+        self.assertEqual(assessment["task_scope"], "author_issue_bootstrap")
+        self.assertEqual(
+            assessment["bootstrap_path"], "clean_canonical_control_checkout"
+        )
+        self.assertEqual(assessment["dirty_files"], [])
+        self.assertIs(assessment["under_branches"], False)
+        self.assertTrue(assessment["base_tips_verified"])
+        self.assertEqual(assessment["local_head_sha"], MASTER)
+        self.assertEqual(assessment["remote_master_sha"], MASTER)
+        self.assertEqual(assessment["workspace_path"], os.path.realpath(CONTROL))
+        self.assertEqual(
+            assessment["canonical_repo_root"], os.path.realpath(CONTROL)
+        )
+
+    def test_wrong_task_not_applicable(self):
+        assessment = _assess(task="lock_issue")
+        self.assertTrue(assessment["not_applicable"])
+        self.assertFalse(assessment["allowed"])
+
+    def test_branches_worktree_not_applicable_for_control_waiver(self):
+        branches = os.path.join(CONTROL, "branches", "fix-issue-1")
+        assessment = _assess(workspace=branches)
+        self.assertTrue(assessment["not_applicable"])
+        self.assertFalse(assessment["allowed"])
+        self.assertEqual(assessment["bootstrap_path"], "existing_branches_worktree")
+
+    def test_dirty_control_blocks(self):
+        assessment = _assess(porcelain=" M gitea_mcp_server.py\n")
+        self.assertTrue(assessment["block"])
+        self.assertFalse(assessment["allowed"])
+        self.assertTrue(any("tracked local edits" in r for r in assessment["reasons"]))
+
+    def test_head_remote_mismatch_blocks(self):
+        assessment = _assess(head=MASTER, remote=OTHER)
+        self.assertTrue(assessment["block"])
+        self.assertFalse(assessment["allowed"])
+
+    def test_missing_remote_tip_blocks(self):
+        assessment = _assess(remote=None)
+        self.assertTrue(assessment["block"])
+        self.assertFalse(assessment["allowed"])
+
+
+class TestAuthorBootstrapPredicate(unittest.TestCase):
+    def _permits(self, assessment, task="bootstrap_author_issue_worktree"):
+        return cib.bootstrap_permits_control_checkout(
+            assessment,
+            task=task,
+            workspace_path=os.path.realpath(CONTROL),
+            canonical_repo_root=os.path.realpath(CONTROL),
+        )
+
+    def test_clean_author_bootstrap_permits(self):
+        self.assertTrue(self._permits(_assess()))
+
+    def test_tool_alias_permits(self):
+        assessment = _assess(task="gitea_bootstrap_author_issue_worktree")
+        self.assertTrue(
+            self._permits(assessment, task="gitea_bootstrap_author_issue_worktree")
+        )
+
+    def test_create_issue_scope_cannot_license_author_bootstrap(self):
+        # Cross-scope smuggling: a create_issue-shaped assessment must not
+        # authorize the author bootstrap task.
+        create_shaped = dict(_assess())
+        create_shaped["task_scope"] = "create_issue_only"
+        self.assertFalse(self._permits(create_shaped))
+
+    def test_author_scope_cannot_license_create_issue(self):
+        assessment = _assess()
+        self.assertFalse(
+            cib.bootstrap_permits_control_checkout(
+                assessment,
+                task="create_issue",
+                workspace_path=os.path.realpath(CONTROL),
+                canonical_repo_root=os.path.realpath(CONTROL),
+            )
+        )
+
+    def test_nonempty_reasons_fail_closed(self):
+        bad = dict(_assess(), reasons=["informational text must not be here"])
+        self.assertFalse(self._permits(bad))
+
+    def test_dirty_fails_closed(self):
+        self.assertFalse(self._permits(_assess(porcelain=" M x.py\n")))
+
+    def test_mismatch_fails_closed(self):
+        self.assertFalse(self._permits(_assess(remote=OTHER)))
+
+
+class TestAuthorBootstrapPreflightIntegration(unittest.TestCase):
+    """Server preflight path: clean control + author bootstrap task must not raise."""
+
+    def test_enforce_branches_only_allows_clean_control_for_bootstrap(self):
+        # Exercise the real enforcer wiring with a temporary clean repo.
+        import gitea_mcp_server as srv
+
+        with tempfile.TemporaryDirectory() as tmp:
+            repo = os.path.join(tmp, "repo")
+            os.makedirs(os.path.join(repo, "branches"))
+            # Minimal git repo on master at a known tip.
+            import subprocess
+
+            subprocess.check_call(["git", "init", "-b", "master", repo])
+            subprocess.check_call(
+                ["git", "-C", repo, "commit", "--allow-empty", "-m", "init"]
+            )
+            head = subprocess.check_output(
+                ["git", "-C", repo, "rev-parse", "HEAD"], text=True
+            ).strip()
+
+            assessment = aib.assess_author_issue_bootstrap(
+                workspace_path=repo,
+                canonical_repo_root=repo,
+                current_branch="master",
+                head_sha=head,
+                porcelain_status="",
+                remote_master_sha=head,
+                task="bootstrap_author_issue_worktree",
+            )
+            self.assertTrue(
+                cib.bootstrap_permits_control_checkout(
+                    assessment,
+                    task="bootstrap_author_issue_worktree",
+                    workspace_path=repo,
+                    canonical_repo_root=repo,
+                )
+            )
+
+            # Simulate what _enforce_branches_only_author_mutation does when
+            # durable resolution blocks control: the shared predicate must waive.
+            durable_block = {
+                "block": True,
+                "workspace_path": repo,
+                "workspace_binding_source": "process_project_root",
+                "reasons": [
+                    "author mutation blocked: workspace is the stable control checkout"
+                ],
+            }
+            if cib.bootstrap_permits_control_checkout(
+                assessment,
+                task="bootstrap_author_issue_worktree",
+                workspace_path=repo,
+                canonical_repo_root=repo,
+            ):
+                waived = True
+            else:
+                waived = False
+            self.assertTrue(waived)
+            # Keep durable_block referenced so the scenario is explicit.
+            self.assertTrue(durable_block["block"])
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_issue_941_scope_guard_bootstrap_wiring.py

@@ -0,0 +1,346 @@
+"""Regression: author bootstrap scope reaches workflow_scope_guard (#941).
+
+PR #926 (#892) made ``bootstrap_permits_control_checkout`` accept
+``task_scope=author_issue_bootstrap`` and wired that canonical decision into
+the #274 branches-only enforcer and the #604 anti-stomp preflight. A third
+enforcement path was left unwired.
+
+``workflow_scope_guard.assess_root_source_mutation`` kept its own copy of the
+clean-root author decision, gated on ``create_issue_bootstrap.is_create_issue_task``
+— a task-name allowlist that never contained ``bootstrap_author_issue_worktree``.
+So the real call path
+
+    gitea_bootstrap_author_issue_worktree
+      -> verify_preflight_purity
+        -> _enforce_issue_scope_guard
+          -> workflow_scope_guard.assess_production_mutation_guards
+
+raised ProductionGuardError(missing_issue_worktree) before
+``assess_author_issue_bootstrap`` was ever consulted.
+
+These tests drive the real enforcer, not the authorization helper in
+isolation. A helper-only test cannot observe this defect: #892's own predicate
+tests all passed while the live bootstrap stayed blocked.
+"""
+
+from __future__ import annotations
+
+import os
+import subprocess
+import tempfile
+import unittest
+from unittest import mock
+
+import author_issue_bootstrap as aib
+import create_issue_bootstrap as cib
+import workflow_scope_guard
+
+BOOTSTRAP_TASK = "bootstrap_author_issue_worktree"
+BOOTSTRAP_TOOL = "gitea_bootstrap_author_issue_worktree"
+
+
+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 _assessment(
+    repo: str,
+    head: str,
+    *,
+    task: str = BOOTSTRAP_TASK,
+    porcelain: str = "",
+    remote: str | None = None,
+) -> dict:
+    return aib.assess_author_issue_bootstrap(
+        workspace_path=repo,
+        canonical_repo_root=repo,
+        current_branch="master",
+        head_sha=head,
+        porcelain_status=porcelain,
+        remote_master_sha=head if remote is None else remote,
+        task=task,
+    )
+
+
+class _ControlCheckoutHarness(unittest.TestCase):
+    """Drive the real server guard against a temporary clean control checkout."""
+
+    def setUp(self):
+        self._tmp = tempfile.TemporaryDirectory()
+        self.addCleanup(self._tmp.cleanup)
+        self.repo, self.head = _make_control_repo(self._tmp.name)
+
+        # #683 force-on: production guards must execute under pytest.
+        patcher = mock.patch.dict(
+            os.environ,
+            {workflow_scope_guard.FORCE_PRODUCTION_GUARDS_ENV: "1"},
+        )
+        patcher.start()
+        self.addCleanup(patcher.stop)
+
+    def _enforce(
+        self,
+        task: str,
+        *,
+        porcelain: str = "",
+        assessment: object = "auto",
+        role_kind: str = "author",
+    ):
+        """Call the real _enforce_issue_scope_guard for *task*."""
+        import gitea_mcp_server as srv
+
+        if assessment == "auto":
+            assessment = _assessment(
+                self.repo, self.head, task=task, porcelain=porcelain
+            )
+
+        ctx = {
+            "workspace_path": self.repo,
+            "canonical_repo_root": self.repo,
+            "workspace_role_kind": role_kind,
+            "workspace_binding_source": "process_project_root",
+        }
+        git_state = {
+            "current_branch": "master",
+            "head_sha": self.head,
+            "porcelain_status": porcelain,
+        }
+
+        with mock.patch.object(
+            srv, "_resolve_namespace_mutation_context", return_value=ctx
+        ), mock.patch.object(
+            srv.issue_lock_worktree,
+            "read_worktree_git_state",
+            return_value=git_state,
+        ), mock.patch.object(
+            srv,
+            "_session_issue_lock_snapshot",
+            return_value={
+                "locked_issue_number": None,
+                "lock_branch_name": None,
+                "worktrees_match": False,
+            },
+        ), mock.patch.object(
+            srv, "_actual_profile_role", return_value=role_kind
+        ), mock.patch.object(
+            srv, "_effective_workspace_role", return_value=role_kind
+        ), mock.patch.object(
+            srv, "_create_issue_bootstrap_assessment", return_value=assessment
+        ):
+            srv._enforce_issue_scope_guard(None, task=task)
+
+
+class TestRealPathBootstrapReachesGuard(_ControlCheckoutHarness):
+    """The defect and its fix, observed through the real enforcer."""
+
+    def test_bootstrap_task_passes_scope_guard_from_clean_control(self):
+        # Pre-fix this raises ProductionGuardError(missing_issue_worktree)
+        # because the guard consulted a task-name allowlist instead of the
+        # canonical authorization decision.
+        self._enforce(BOOTSTRAP_TASK)
+
+    def test_bootstrap_tool_alias_passes_scope_guard(self):
+        self._enforce(BOOTSTRAP_TOOL)
+
+    def test_guard_consults_canonical_predicate(self):
+        """The guard must reach bootstrap_permits_control_checkout, not a name list."""
+        real = cib.bootstrap_permits_control_checkout
+        seen: list[str | None] = []
+
+        def _spy(assessment, *, task, workspace_path, canonical_repo_root):
+            seen.append(task)
+            return real(
+                assessment,
+                task=task,
+                workspace_path=workspace_path,
+                canonical_repo_root=canonical_repo_root,
+            )
+
+        with mock.patch.object(
+            cib, "bootstrap_permits_control_checkout", side_effect=_spy
+        ):
+            self._enforce(BOOTSTRAP_TASK)
+
+        self.assertIn(
+            BOOTSTRAP_TASK,
+            seen,
+            "workflow_scope_guard did not consult the canonical bootstrap "
+            "authorization decision",
+        )
+
+
+class TestFailClosedOnBadEvidence(_ControlCheckoutHarness):
+    """Missing, malformed, or mismatched scope evidence must still block."""
+
+    def _assert_blocked(self, **kwargs):
+        with self.assertRaises(workflow_scope_guard.ProductionGuardError):
+            self._enforce(BOOTSTRAP_TASK, **kwargs)
+
+    def test_missing_assessment_fails_closed(self):
+        self._assert_blocked(assessment=None)
+
+    def test_malformed_assessment_fails_closed(self):
+        self._assert_blocked(assessment={"allowed": True})
+
+    def test_non_dict_assessment_fails_closed(self):
+        self._assert_blocked(assessment="allowed")
+
+    def test_wrong_task_scope_fails_closed(self):
+        bad = dict(_assessment(self.repo, self.head))
+        bad["task_scope"] = "create_issue_only"
+        self._assert_blocked(assessment=bad)
+
+    def test_nonempty_reasons_fail_closed(self):
+        bad = dict(_assessment(self.repo, self.head), reasons=["note"])
+        self._assert_blocked(assessment=bad)
+
+    def test_mismatched_base_tips_fail_closed(self):
+        bad = dict(_assessment(self.repo, self.head))
+        bad["remote_master_sha"] = "b" * 40
+        self._assert_blocked(assessment=bad)
+
+    def test_unverified_base_tips_fail_closed(self):
+        bad = dict(_assessment(self.repo, self.head), base_tips_verified=False)
+        self._assert_blocked(assessment=bad)
+
+    def test_mismatched_workspace_binding_fails_closed(self):
+        bad = dict(_assessment(self.repo, self.head))
+        bad["workspace_path"] = os.path.join(self.repo, "elsewhere")
+        self._assert_blocked(assessment=bad)
+
+    def test_mismatched_repo_root_binding_fails_closed(self):
+        bad = dict(_assessment(self.repo, self.head))
+        bad["canonical_repo_root"] = os.path.join(self.repo, "other-root")
+        self._assert_blocked(assessment=bad)
+
+    def test_blocked_assessment_fails_closed(self):
+        bad = dict(_assessment(self.repo, self.head), block=True, allowed=False)
+        self._assert_blocked(assessment=bad)
+
+
+class TestOrdinaryControlCheckoutMutationStillForbidden(_ControlCheckoutHarness):
+    """The waiver must not leak to ordinary author work."""
+
+    def test_ordinary_author_task_still_blocked(self):
+        with self.assertRaises(workflow_scope_guard.ProductionGuardError):
+            self._enforce("commit_files", assessment=None)
+
+    def test_lock_issue_still_blocked_from_control(self):
+        with self.assertRaises(workflow_scope_guard.ProductionGuardError):
+            self._enforce("lock_issue", assessment=None)
+
+    def test_bootstrap_assessment_cannot_license_other_task(self):
+        # Cross-task smuggling: valid bootstrap evidence must not waive a
+        # different author mutation.
+        good = _assessment(self.repo, self.head)
+        with self.assertRaises(workflow_scope_guard.ProductionGuardError):
+            self._enforce("commit_files", assessment=good)
+
+    def test_dirty_control_checkout_still_blocked_for_bootstrap(self):
+        with self.assertRaises(workflow_scope_guard.ProductionGuardError):
+            self._enforce(BOOTSTRAP_TASK, porcelain=" M gitea_mcp_server.py\n")
+
+
+class TestCreateIssueBehaviorUnchanged(_ControlCheckoutHarness):
+    """#749 create_issue keeps its own sanctioned path."""
+
+    def test_create_issue_still_allowed_from_clean_control(self):
+        self._enforce("create_issue", assessment=None)
+
+    def test_create_issue_tool_alias_still_allowed(self):
+        self._enforce("gitea_create_issue", assessment=None)
+
+    def test_create_issue_blocked_when_control_dirty(self):
+        with self.assertRaises(workflow_scope_guard.ProductionGuardError):
+            self._enforce(
+                "create_issue",
+                porcelain=" M gitea_mcp_server.py\n",
+                assessment=None,
+            )
+
+
+class TestGuardUnitLevelWiring(unittest.TestCase):
+    """assess_root_source_mutation itself must accept and honour the evidence."""
+
+    def setUp(self):
+        self._tmp = tempfile.TemporaryDirectory()
+        self.addCleanup(self._tmp.cleanup)
+        self.repo, self.head = _make_control_repo(self._tmp.name)
+        patcher = mock.patch.dict(
+            os.environ,
+            {workflow_scope_guard.FORCE_PRODUCTION_GUARDS_ENV: "1"},
+        )
+        patcher.start()
+        self.addCleanup(patcher.stop)
+
+    def _assess(self, *, task=BOOTSTRAP_TASK, bootstrap_assessment="auto"):
+        if bootstrap_assessment == "auto":
+            bootstrap_assessment = _assessment(self.repo, self.head, task=task)
+        return workflow_scope_guard.assess_root_source_mutation(
+            workspace_path=self.repo,
+            canonical_repo_root=self.repo,
+            porcelain_status="",
+            current_branch="master",
+            role_kind="author",
+            mutation_task=task,
+            bootstrap_assessment=bootstrap_assessment,
+        )
+
+    def test_valid_evidence_unblocks(self):
+        result = self._assess()
+        self.assertFalse(result["block"])
+        self.assertIsNone(result["blocker_kind"])
+
+    def test_absent_evidence_blocks(self):
+        result = self._assess(bootstrap_assessment=None)
+        self.assertTrue(result["block"])
+        self.assertEqual(
+            result["blocker_kind"], workflow_scope_guard.BLOCKER_MISSING_WORKTREE
+        )
+
+    def test_reconciler_exemption_preserved(self):
+        result = workflow_scope_guard.assess_root_source_mutation(
+            workspace_path=self.repo,
+            canonical_repo_root=self.repo,
+            porcelain_status="",
+            current_branch="master",
+            role_kind="reconciler",
+            mutation_task=BOOTSTRAP_TASK,
+        )
+        self.assertFalse(result["block"])
+
+    def test_signature_accepts_evidence_without_it_being_required(self):
+        # Callers that supply no evidence keep the pre-existing behaviour.
+        result = workflow_scope_guard.assess_root_source_mutation(
+            workspace_path=self.repo,
+            canonical_repo_root=self.repo,
+            porcelain_status="",
+            current_branch="master",
+            role_kind="author",
+            mutation_task="create_issue",
+        )
+        self.assertFalse(result["block"])
+
+
+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_stale_runtime.py

@@ -35,10 +35,10 @@ class TestMcpStaleRuntime(unittest.TestCase):
         
         # Mock env output for ps eww
         mock_run_env12345 = MagicMock()
-        mock_run_env12345.stdout = "GITEA_MCP_PROFILE=prgs-reconciler"
+        mock_run_env12345.stdout = "GITEA_MCP_PROFILE=prgs-reconciler GITEA_CLIENT_MANAGED=1"
         
         mock_run_env54321 = MagicMock()
-        mock_run_env54321.stdout = "GITEA_MCP_PROFILE=prgs-author"
+        mock_run_env54321.stdout = "GITEA_MCP_PROFILE=prgs-author GITEA_CLIENT_MANAGED=1"
         
         def side_effect(args, **kwargs):
             if args[0] == "ps" and "eww" in args:
@@ -91,7 +91,7 @@ class TestMcpStaleRuntime(unittest.TestCase):
         mock_run_ps.stdout = ps_output
 
         mock_run_env = MagicMock()
-        mock_run_env.stdout = "GITEA_MCP_PROFILE=prgs-author"
+        mock_run_env.stdout = "GITEA_MCP_PROFILE=prgs-author GITEA_CLIENT_MANAGED=1"
 
         mock_run_git = MagicMock()
         mock_run_git.stdout = "FAKE2" # different SHA

tests/test_runtime_clarity.py

@@ -243,9 +243,10 @@ class TestRuntimeClarity(unittest.TestCase):
         self.assertIn("switching is disabled", res["message"].lower())
         self.assertIsNone(gitea_config._active_profile_override)
 
+    @patch("mcp_server._trusted_session_repository", return_value={"repository": "Example-Org/Example-Repo", "org": "Example-Org", "repo": "Example-Repo", "reasons": []})
     @patch("mcp_server.api_request")
     @patch("mcp_server.get_auth_header")
-    def test_activate_profile_succeeds_when_enabled(self, mock_auth, mock_api):
+    def test_activate_profile_succeeds_when_enabled(self, mock_auth, mock_api, mock_trusted):
         self._write_config(CONFIG_SWITCHING_ENABLED)
 
         # Setup mock responses for whoami checks

workflow_scope_guard.py

@@ -279,6 +279,7 @@ def assess_root_source_mutation(
     locked_issue_number: int | None = None,
     role_kind: str | None = None,
     mutation_task: str | None = None,
+    bootstrap_assessment: Any | None = None,
 ) -> dict[str, Any]:
     """Fail closed for diagnostic/source edits on the control/root checkout.
 
@@ -286,6 +287,13 @@ def assess_root_source_mutation(
     tracked source/test files on the control checkout always block, including
     temporary/diagnostic/test-only intent.
 
+    #941: ``bootstrap_author_issue_worktree`` is judged by the canonical
+    ``create_issue_bootstrap.bootstrap_permits_control_checkout`` decision over
+    *bootstrap_assessment* — the same server-derived evidence the #274 and
+    #604 guards consume — instead of a task-name allowlist local to this
+    module. Evidence that is absent, malformed, wrongly scoped, or bound to
+    another workspace leaves the ordinary block in force.
+
     #749: ``create_issue`` is a pure remote mutation with no local tree write.
     When *mutation_task* is create_issue and the control checkout has no dirty
     source/test files, the missing-worktree signal is suppressed so the
@@ -336,6 +344,19 @@ def assess_root_source_mutation(
         if _cib is not None and _cib.is_create_issue_task(mutation_task):
             # #749: clean-root create_issue is the sanctioned bootstrap path.
             create_issue_bootstrap = True
+        elif _cib is not None and _cib.bootstrap_permits_control_checkout(
+            bootstrap_assessment,
+            task=mutation_task,
+            workspace_path=workspace,
+            canonical_repo_root=root,
+        ):
+            # #941: the author issue-worktree bootstrap is authorized by the
+            # canonical shared decision over server-derived task-scope
+            # evidence, never by a task-name allowlist kept in this module.
+            # The predicate fails closed on missing, malformed, cross-scope,
+            # dirty, drifted, or wrongly bound evidence, so this arm cannot
+            # widen the waiver beyond the one sanctioned bootstrap task.
+            create_issue_bootstrap = True
         else:
             # Explicit missing-worktree signal for force-on author entrypoints.
             reasons.append(
@@ -393,8 +414,15 @@ def assess_production_mutation_guards(
     require_author_lock: bool = False,
     in_test_mode: bool = False,
     mutation_task: str | None = None,
+    bootstrap_assessment: Any | None = None,
 ) -> dict[str, Any]:
-    """Compose root + scope production guards when they must be active (#683)."""
+    """Compose root + scope production guards when they must be active (#683).
+
+    #941: *bootstrap_assessment* is the server-derived author-bootstrap
+    evidence, forwarded unchanged to :func:`assess_root_source_mutation` so
+    this guard reaches the same canonical decision as the #274 and #604
+    guards. Omitting it preserves the pre-existing behaviour.
+    """
     if not production_guards_active(in_test_mode=in_test_mode):
         return {
             "proven": True,
@@ -414,6 +442,7 @@ def assess_production_mutation_guards(
         locked_issue_number=locked_issue_number,
         role_kind=role_kind,
         mutation_task=mutation_task,
+        bootstrap_assessment=bootstrap_assessment,
     )
     if root_assess["block"]:
         return {**root_assess, "skipped": False}