fix(workflow): fail-closed key version, strict incident evidence, archive-gated clear (#709)

Address formal review 438 REQUEST_CHANGES on PR #710 (F6/F7/F8).

F6 — HMAC key-version validation fails closed:
- verify_authorization_artifact validates key_version BEFORE any MAC work, so an
  attacker-chosen version can never select the signing key.
- Require exactly one nonempty, well-formed key-version field; missing, empty,
  unknown, malformed, duplicated (including identical-valued and nested
  aliases), and mismatched versions all fail. No versionless legacy fallback.
- Artifact version must equal the configured active version; production now
  requires GITEA_IRRECOVERABLE_AUTH_HMAC_KEY_VERSION explicitly (an implicit
  default made rotation ambiguous). Version stays inside the signed material.

F7 — strictly canonical incident evidence:
- Replace substring/first-match parsing with an exact schema: marker on line 1,
  every field once, fixed order, no duplicate/unknown/empty/conflicting fields
  in or outside the signed block. The parsed body is re-rendered and compared
  for exact equality before acceptance.
- content_digest now binds the full recovery scope: repository identity, PR,
  decision-lock identity, destroyed subject, recovery action, recorded and
  expected head, incident issue, evidence author, minting actor, key version,
  nonce and issued_at.
- Actor identity is the immutable user id with login consistency; conflicting
  ids/logins and display-name-only identities fail closed. Edited comments are
  rejected. The independent-author rule is preserved and enforced by stable id.
- build_canonical_incident_body is the single source of the accepted format and
  refuses to emit ambiguous evidence.

F8 — archival is a prerequisite for clearing terminal evidence:
- _clear_decision_lock_for_profile no longer swallows archive failures. It
  requires a successful write plus a durable read-back matching the PR/head,
  and otherwise returns a structured, retry-safe failure that retains the lock
  and records actionable recovery evidence.
- Fixes a latent bug the read-back exposed: the archive payload inherited the
  source lock's session_profile_lock, so save_state keyed the archive under the
  reviewer profile instead of the archive identity and it never read back.

Adversarial regressions added for every listed case: key-version missing/empty/
unknown/malformed/duplicate/rotation/wrong-key-after-restart, reordered fields,
duplicate identical and conflicting fields, conflicting actor ids/names,
digest-preserving substitution, decision-lock and recovery-action substitution,
cross-PR/repo/org/remote/head replay, archive exception/timeout/false/empty/
partial-readback with proof the lock survives, exactly-one permitted clear, and
retry after archive failure. No existing assertion was weakened.

Validation: focused 150 passed in three module orders; full tests/ 2809 passed,
6 skipped, 1 warning (pre-existing StarletteDeprecationWarning in
tests/test_webui_audit.py:8), 161 subtests passed.

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

.env.example

@@ -55,3 +55,12 @@ GITEA_MCP_PROFILE=prgs
 # GITEA_MERGER_WORKTREE=/path/to/repo/branches/merge-pr456
 # GITEA_RECONCILER_WORKTREE=/path/to/repo/branches/reconcile-pr456
 # GITEA_ACTIVE_WORKTREE=/path/to/repo/branches/session-override
+
+# Durable HMAC key for irrecoverable decision-lock provenance artifacts (#709 F4).
+# REQUIRED in production native MCP processes that mint or verify recovery
+# authorization (reconciler + merger must share the same key). Hex (preferred),
+# base64, or utf-8 literal (>=16 bytes). Never generate an ephemeral per-process
+# key — cross-process / post-restart verification would fail closed.
+# GITEA_IRRECOVERABLE_AUTH_HMAC_KEY=0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef
+# Optional key version string bound into the signature (for rotation).
+# GITEA_IRRECOVERABLE_AUTH_HMAC_KEY_VERSION=v1

docs/gitea-execution-profiles.md

@@ -45,7 +45,6 @@ authenticated capability set. Each profile defines the following fields:
 | `authenticated_username` | string | The Gitea login this profile authenticates as (verified at runtime via `gitea_whoami`, not trusted from config). |
 | `allowed_operations` | list | Operation categories this profile may perform. |
 | `forbidden_operations` | list | Operation categories this profile must never perform. |
-| `allowed_repositories` | list | Optional. Canonical `owner/repository` slugs this profile may bind to. An authorization boundary, not the binding itself — see [Repository scope](#repository-scope-714). |
 | `token_source_name` | string | The *name* of the secret source (e.g. env var name or secret key). **Never the token value.** |
 | `audit_label` | string | Short label attached to audit records for actions by this profile. |
 | `can_approve_prs` | bool | May submit an approving PR review. |
@@ -58,47 +57,6 @@ authenticated capability set. Each profile defines the following fields:
 name), never the token itself. Token values are never part of a profile object,
 never logged, never returned by a tool, and never committed.
 
-## Repository scope (#714)
-
-`allowed_repositories` declares the canonical `owner/repository` slugs a profile
-may operate on:
-
-```json
-"prgs-author": {
-  "allowed_repositories": ["Scaled-Tech-Consulting/Gitea-Tools"]
-}
-```
-
-It is an **authorization boundary, not the session binding**. The binding is
-derived and enforced like this:
-
-1. The session repository is derived from the **verified, workspace-aligned git
-   remote** — never from a caller-supplied `org`/`repo` argument, and never from
-   the `REMOTES` table (whose entries are default *targets*: `prgs` defaults to
-   `Timesheet`, which is not this project).
-2. That workspace-derived slug is validated against `allowed_repositories`.
-3. The session binds immutably to that one canonical `owner/repository`. The
-   organization is derived from the slug; there is no independent,
-   caller-controlled organization value.
-4. If a profile authorizes several repositories, the verified workspace still
-   selects exactly one. A session never binds to the whole list and never
-   switches between entries.
-
-Fail-closed rules:
-
-- Activation is rejected when the workspace repository is absent from the
-  allowlist.
-- A mutation is rejected when no verified workspace repository can be
-  established.
-- A tool-level `org`/`repo` override that disagrees with the binding is rejected
-  before the mutation runs.
-- A mutation request can never establish, complete, or replace the binding.
-
-The field is **config-only**: no environment variable can widen or forge it. It
-is optional — a profile that omits it keeps the previous behaviour, so existing
-static-profile namespaces stay functional until an operator provisions the
-scope. Provisioning it is what activates enforcement for that profile.
-
 ## Example profiles
 
 The following are the reference profiles. Booleans express intended capability

gitea-mcp.v2-contexts.example.json

@@ -41,7 +41,6 @@
       "execution_profile": "example-author",
       "audit_label": "example-author",
       "auth": { "type": "keychain", "id": "example-gitea-author-token" },
-      "allowed_repositories": ["Example-Org/Example-Repo"],
       "allowed_operations": ["read", "branch", "commit", "push", "open_pr", "comment", "issue.comment"],
       "forbidden_operations": ["approve", "request_changes", "merge"]
     },

gitea_auth.py

@@ -182,51 +182,11 @@ def get_auth_header(host):
 
 def resolve_remote(args):
     """Given parsed argparse args with --remote/--host/--org/--repo,
-    return (host, org, repo) with overrides applied.
+    return (host, org, repo) with overrides applied."""
-
-    #714 / #530: when the caller omits org and/or repo, prefer the
-    workspace-aligned git remote over REMOTES defaults (e.g. bare
-    ``--remote prgs`` must not force Timesheet when the checkout is
-    Gitea-Tools). Explicit --org/--repo always win.
-    """
     profile = REMOTES[args.remote]
     host = args.host or profile["host"]
-    org_explicit = getattr(args, "org", None) is not None
+    org = args.org or profile["org"]
-    repo_explicit = getattr(args, "repo", None) is not None
+    repo = args.repo or profile["repo"]
-    org = args.org if org_explicit else profile["org"]
-    repo = args.repo if repo_explicit else profile["repo"]
-    if not org_explicit or not repo_explicit:
-        try:
-            import remote_repo_guard
-            import subprocess
-            # Prefer the named remote URL when present; fall back to origin.
-            url = None
-            for remote_name in (args.remote, "origin"):
-                try:
-                    proc = subprocess.run(
-                        ["git", "remote", "get-url", remote_name],
-                        capture_output=True,
-                        text=True,
-                        timeout=5,
-                        check=False,
-                    )
-                    if proc.returncode == 0 and (proc.stdout or "").strip():
-                        url = proc.stdout.strip()
-                        break
-                except Exception:
-                    continue
-            # Allow tests to inject a deterministic remote URL without Git.
-            env_url = os.environ.get("GITEA_TEST_WORKSPACE_REMOTE_URL")
-            if env_url:
-                url = env_url
-            parsed = remote_repo_guard.parse_org_repo_from_remote_url(url)
-            if parsed:
-                if not org_explicit:
-                    org = parsed[0]
-                if not repo_explicit:
-                    repo = parsed[1]
-        except Exception:
-            pass
     return host, org, repo
 
 
@@ -610,10 +570,6 @@ def get_profile():
         "profile_name": name,
         "allowed_operations": ops,
         "forbidden_operations": forbidden,
-        # #714 repository authorization boundary. Config-only on purpose: an
-        # environment variable must never widen or forge the set of
-        # repositories a session may bind to.
-        "allowed_repositories": _json_list("allowed_repositories"),
         "audit_label": audit_label,
         "token_source_name": token_source,
         "auth_source_type": auth_type,

gitea_config.py

@@ -475,33 +475,6 @@ def _require_enabled(kind, name, obj):
     return enabled
 
 
-_REPO_SCOPE_RE = re.compile(r"^[^/\s]+/[^/\s]+$")
-
-
-def _validate_allowed_repositories(name, raw):
-    """Validate the optional per-profile repository authorization scope (#714).
-
-    ``allowed_repositories`` is an authorization boundary of canonical
-    ``owner/repository`` slugs. It is not the session binding: the verified
-    workspace repository selects exactly one entry at bind time. Absent means
-    "no repository scope configured" and is allowed, so existing profiles keep
-    working until an operator provisions the field.
-    """
-    if raw is None:
-        return
-    if not isinstance(raw, list):
-        raise ConfigError(
-            f"profile '{name}' allowed_repositories must be a list of "
-            "'owner/repository' strings"
-        )
-    for entry in raw:
-        if not isinstance(entry, str) or not _REPO_SCOPE_RE.match(entry.strip()):
-            raise ConfigError(
-                f"profile '{name}' allowed_repositories entry {entry!r} is not "
-                "a canonical 'owner/repository' slug"
-            )
-
-
 def _reject_inline_secrets(kind, name, obj):
     for key in _INLINE_SECRET_KEYS:
         if key in obj:
@@ -576,7 +549,6 @@ def _load_v2_contexts(data, path):
         forbidden = raw.get("forbidden_operations") or []
         if not isinstance(allowed, list) or not isinstance(forbidden, list):
             raise ConfigError(f"profile '{name}' operation fields must be lists")
-        _validate_allowed_repositories(name, raw.get("allowed_repositories"))
         allowed_n = {_normalize_op("gitea", op, name) for op in allowed}
         forbidden_n = {_normalize_op("gitea", op, name) for op in forbidden}
         # Reviewer-identity deadlock rule (#100/#103) applies here unchanged.

mcp_session_state.py

@@ -36,7 +36,24 @@ KIND_REVIEW_DRAFT = "review_draft"
 # other session proofs; a contaminated session fails closed on gated mutations
 # until a reconciler audits and clears it.
 KIND_STABLE_BRANCH_CONTAMINATION = "stable_branch_contamination"
+# #709: archive prior terminal decision ledgers instead of silent overwrite.
+KIND_DECISION_LOCK_ARCHIVE = "review_decision_lock_archive"
+# #709: post-merge cleanup/audit reconciliation-required durable record.
+KIND_POST_MERGE_DECISION_RECOVERY = "post_merge_decision_recovery"
+# #709: truthful record when historical terminal evidence is irrecoverably gone.
+KIND_IRRECOVERABLE_DECISION_PROVENANCE = "irrecoverable_decision_provenance"
+# #709 F1: server-side non-forgeable authorization artifact for recovery.
+KIND_IRRECOVERABLE_PROVENANCE_AUTH = "irrecoverable_provenance_authorization"
 
+# Kinds that must survive the default session-state TTL (forensic / recovery).
+RECOVERY_CRITICAL_KINDS = frozenset(
+    {
+        KIND_DECISION_LOCK_ARCHIVE,
+        KIND_POST_MERGE_DECISION_RECOVERY,
+        KIND_IRRECOVERABLE_DECISION_PROVENANCE,
+        KIND_IRRECOVERABLE_PROVENANCE_AUTH,
+    }
+)
 
 
 _SAFE_SEGMENT_RE = re.compile(r"[^A-Za-z0-9._+-]+")
@@ -270,7 +287,11 @@ def identity_match_reasons(
         reasons.append("session state missing recorded_at timestamp (fail closed)")
     else:
         age = _now_utc() - recorded_at
-        if age > timedelta(hours=ttl_hours()):
+        kind = (record.get("kind") or "").strip()
+        ttl_exempt = kind in RECOVERY_CRITICAL_KINDS or bool(
+            record.get("recovery_critical")
+        )
+        if age > timedelta(hours=ttl_hours()) and not ttl_exempt:
             reasons.append(
                 f"session state expired after {ttl_hours():g}h (fail closed)"
             )
@@ -447,3 +468,177 @@ def clear_state(
         profile_identity=profile_identity,
         state_dir=state_dir,
     )
+
+def list_decision_lock_profile_identities(
+    state_dir: str | None = None,
+) -> list[str]:
+    """Return profile identities that have a durable review_decision_lock file (#709).
+
+    Filename form: ``review_decision_lock-<profile>.json`` (see ``state_key``).
+    Does not validate TTL or identity — callers must load via ``load_state``.
+    """
+    root = (state_dir or default_state_dir()).strip()
+    if not root or not os.path.isdir(root):
+        return []
+    prefix = f"{_sanitize_segment(KIND_DECISION_LOCK)}-"
+    suffix = ".json"
+    found: list[str] = []
+    try:
+        names = os.listdir(root)
+    except OSError:
+        return []
+    for name in names:
+        if not name.startswith(prefix) or not name.endswith(suffix):
+            continue
+        if name.endswith(".lock"):
+            continue
+        mid = name[len(prefix) : -len(suffix)]
+        if mid:
+            found.append(mid)
+    return sorted(set(found))
+
+
+def load_state_for_profile(
+    *,
+    kind: str,
+    profile_identity: str,
+    remote: str | None = None,
+    org: str | None = None,
+    repo: str | None = None,
+    state_dir: str | None = None,
+    skip_identity_match: bool = False,
+    enforce_repo_scope: bool = True,
+) -> dict[str, Any] | None:
+    """Load durable state for an explicit profile identity (#709 cross-profile).
+
+    When *skip_identity_match* is True, still requires the file's recorded
+    profile_identity to equal the requested profile (anti-stomp), but does not
+    require the *active* session identity to match — needed so a merger can
+    inspect a reviewer lock after merge.
+
+    #709 F3: remote/org/repo filter reasons are **enforced** (not merely
+    computed). When *enforce_repo_scope* is True (default) and the caller
+    supplies remote/org/repo, mismatches fail closed with ``None``.
+    """
+    # #709 F3: refuse traversal / malformed profile identities.
+    try:
+        from irrecoverable_provenance import assess_profile_path_identity
+
+        path_gate = assess_profile_path_identity(profile_identity)
+        if not path_gate.get("valid"):
+            return None
+    except Exception:
+        # Module may be mid-import in edge bootstraps; fall through to
+        # conservative checks below.
+        raw = str(profile_identity or "")
+        if ".." in raw or "/" in raw or "\\" in raw or "\x00" in raw:
+            return None
+
+    profile = current_profile_identity(profile_identity=profile_identity)
+    root = _ensure_state_dir(state_dir)
+    # Refuse symlink escape of the state root (#709 F3).
+    try:
+        real_root = os.path.realpath(root)
+        path = state_file_path(
+            kind=kind,
+            remote=remote,
+            org=org,
+            repo=repo,
+            profile_identity=profile,
+            state_dir=root,
+        )
+        real_path = os.path.realpath(path) if os.path.exists(path) else path
+        if os.path.exists(path) and not str(real_path).startswith(
+            str(real_root) + os.sep
+        ) and str(real_path) != str(real_root):
+            return None
+    except OSError:
+        return None
+
+    path = state_file_path(
+        kind=kind,
+        remote=remote,
+        org=org,
+        repo=repo,
+        profile_identity=profile,
+        state_dir=root,
+    )
+    lock_path = f"{path}.lock"
+    with _exclusive_file_lock(lock_path):
+        envelope = _read_json(path)
+    if not envelope:
+        return None
+    payload = envelope.get("payload")
+    if not isinstance(payload, dict):
+        return None
+    merged = dict(payload)
+    for key in (
+        "kind",
+        "remote",
+        "org",
+        "repo",
+        "profile_identity",
+        "session_profile_lock",
+        "recorded_at",
+        "updated_at",
+        "writer_pid",
+    ):
+        if key in envelope and key not in merged:
+            merged[key] = envelope[key]
+    stored = (merged.get("profile_identity") or "").strip()
+    if stored and stored != profile:
+        return None
+    if skip_identity_match:
+        # Enforce remote/org/repo when caller provides them (#709 F3).
+        # Drop only *active session* profile-identity mismatches; keep
+        # expiry, spoof, and repository-scope reasons.
+        scope_remote = remote if enforce_repo_scope else None
+        scope_org = org if enforce_repo_scope else None
+        scope_repo = repo if enforce_repo_scope else None
+        reasons = identity_match_reasons(
+            merged,
+            remote=scope_remote,
+            org=scope_org,
+            repo=scope_repo,
+            profile_identity=stored or profile,
+        )
+        filtered = [
+            r
+            for r in reasons
+            if "profile identity mismatch" not in r
+            or (stored and stored != profile)
+        ]
+        # When caller requested a specific remote/org/repo, also fail if the
+        # stored record lacks those identity fields entirely (legacy incomplete).
+        if enforce_repo_scope:
+            for field, want in (
+                ("remote", remote),
+                ("org", org),
+                ("repo", repo),
+            ):
+                want_s = (want or "").strip()
+                if not want_s:
+                    continue
+                have = (str(merged.get(field) or "")).strip()
+                if not have:
+                    filtered.append(
+                        f"session state {field} missing on durable record "
+                        f"(expected={want_s!r}; fail closed, #709 F3)"
+                    )
+                elif have != want_s:
+                    # identity_match_reasons already adds mismatch; ensure kept
+                    pass
+        if filtered:
+            return None
+        return merged
+    reasons = identity_match_reasons(
+        merged,
+        remote=remote,
+        org=org,
+        repo=repo,
+        profile_identity=profile,
+    )
+    if reasons:
+        return None
+    return merged
+

session_context_binding.py

@@ -1,595 +0,0 @@
-"""Session-immutable MCP mutation context (#714).
-
-Pins profile, remote, host, repository, identity, and role for the life of an
-MCP process (or until an explicit ``gitea_activate_profile`` re-bind).
-
-Capability resolution and mutation gates must evaluate only the active
-profile for the requested remote. Silent cross-host / cross-profile
-substitution is forbidden.
-"""
-
-from __future__ import annotations
-
-import os
-import re
-import threading
-import urllib.parse
-from dataclasses import dataclass
-from typing import Any, Mapping
-
-
-@dataclass(frozen=True, slots=True)
-class _SessionContext:
-    """One atomic, immutable process-session binding."""
-
-    profile_name: str | None
-    remote: str | None
-    host: str | None
-    identity: str | None
-    repository: str | None
-    org: str | None
-    role_kind: str | None
-    expected_username: str | None
-    source: str
-    pid: int
-
-    def as_dict(self) -> dict[str, Any]:
-        return {
-            "profile_name": self.profile_name,
-            "remote": self.remote,
-            "host": self.host,
-            "identity": self.identity,
-            "repository": self.repository,
-            "org": self.org,
-            "role_kind": self.role_kind,
-            "expected_username": self.expected_username,
-            "source": self.source,
-            "pid": self.pid,
-        }
-
-
-# Process-local only — never a shared file (same rationale as mutation authority).
-# The frozen value prevents partial mutation, while the lock makes first-bind and
-# sanctioned rebind atomic across concurrent MCP calls.
-_SESSION_CONTEXT: _SessionContext | None = None
-_SESSION_CONTEXT_LOCK = threading.RLock()
-
-
-def _reset_session_context_for_testing() -> None:
-    """Reset at a pytest test boundary; unavailable to production callers.
-
-    Production sessions transition only through process startup/PID change or
-    the explicit profile-activation rebind.  Keeping this helper private and
-    requiring pytest's per-test marker prevents it from becoming an MCP/runtime
-    bypass.
-    """
-    if "PYTEST_CURRENT_TEST" not in os.environ:
-        raise RuntimeError("session context reset is restricted to pytest boundaries")
-    global _SESSION_CONTEXT
-    with _SESSION_CONTEXT_LOCK:
-        _SESSION_CONTEXT = None
-
-
-def get_session_context() -> dict[str, Any] | None:
-    """Return a detached snapshot of the bound context, or None if unbound."""
-    with _SESSION_CONTEXT_LOCK:
-        if _SESSION_CONTEXT is None:
-            return None
-        return _SESSION_CONTEXT.as_dict()
-
-
-def profile_host(profile: dict | None) -> str | None:
-    """Hostname from profile base_url, lowercased, or None."""
-    if not profile:
-        return None
-    base = (profile.get("base_url") or "").strip()
-    if not base:
-        return None
-    try:
-        parsed = urllib.parse.urlparse(base)
-        host = (parsed.netloc or parsed.path or "").strip().lower()
-        return host or None
-    except Exception:
-        return None
-
-
-def remote_host(remote: str | None, remotes: dict | None) -> str | None:
-    """Hostname for a known remote key."""
-    if not remote or not remotes:
-        return None
-    entry = remotes.get(remote) or {}
-    return (entry.get("host") or "").strip().lower() or None
-
-
-def profile_matches_remote(
-    profile: dict | None,
-    remote: str | None,
-    remotes: dict | None,
-    *,
-    contexts: dict | None = None,
-) -> bool:
-    """True when *profile* is bound to the same host/context as *remote*."""
-    if not profile or not remote:
-        return False
-    r_host = remote_host(remote, remotes)
-    p_host = profile_host(profile)
-    if r_host and p_host:
-        return r_host == p_host
-    # Fall back to context name heuristics when base_url missing.
-    ctx = (profile.get("context") or "").strip().lower()
-    if not ctx or not contexts:
-        return False
-    ctx_data = contexts.get(ctx) or {}
-    gitea = ctx_data.get("gitea") or {}
-    base = (gitea.get("base_url") or "").strip()
-    if not base or not r_host:
-        return False
-    try:
-        parsed = urllib.parse.urlparse(base)
-        c_host = (parsed.netloc or parsed.path or "").strip().lower()
-    except Exception:
-        return False
-    return bool(c_host) and c_host == r_host
-
-
-def bind_session_context(
-    *,
-    profile_name: str,
-    remote: str | None,
-    host: str | None,
-    identity: str | None,
-    repository: str | None = None,
-    org: str | None = None,
-    role_kind: str | None = None,
-    expected_username: str | None = None,
-    source: str = "bind",
-) -> dict[str, Any]:
-    """Atomically bind/re-bind context (the explicit activation path)."""
-    with _SESSION_CONTEXT_LOCK:
-        return _bind_session_context_unlocked(
-            profile_name=profile_name,
-            remote=remote,
-            host=host,
-            identity=identity,
-            repository=repository,
-            org=org,
-            role_kind=role_kind,
-            expected_username=expected_username,
-            source=source,
-        )
-
-
-def _bind_session_context_unlocked(
-    *,
-    profile_name: str,
-    remote: str | None,
-    host: str | None,
-    identity: str | None,
-    repository: str | None,
-    org: str | None,
-    role_kind: str | None,
-    expected_username: str | None,
-    source: str,
-) -> dict[str, Any]:
-    """Store a complete immutable context while the caller holds the lock."""
-    global _SESSION_CONTEXT
-    _SESSION_CONTEXT = _SessionContext(
-        profile_name=(profile_name or "").strip() or None,
-        remote=(remote or "").strip() or None,
-        host=(host or "").strip().lower() or None,
-        identity=(identity or "").strip() or None,
-        repository=(repository or "").strip() or None,
-        org=(org or "").strip() or None,
-        role_kind=(role_kind or "").strip() or None,
-        expected_username=(expected_username or "").strip() or None,
-        source=source,
-        pid=os.getpid(),
-    )
-    return _SESSION_CONTEXT.as_dict()
-
-
-def seed_session_context_if_unbound(
-    *,
-    profile_name: str,
-    remote: str | None,
-    host: str | None,
-    identity: str | None,
-    repository: str | None = None,
-    org: str | None = None,
-    role_kind: str | None = None,
-    expected_username: str | None = None,
-    source: str = "seed",
-) -> dict[str, Any]:
-    """Atomically bind only when this process has no current context.
-
-    A changed environment or an interleaved call is not a session boundary and
-    therefore cannot replace an established binding.  A newly started/forked
-    process is recognized by PID; explicit ``gitea_activate_profile`` uses
-    :func:`bind_session_context` as its sanctioned logical-session transition.
-    """
-    with _SESSION_CONTEXT_LOCK:
-        if _SESSION_CONTEXT is None or _SESSION_CONTEXT.pid != os.getpid():
-            return _bind_session_context_unlocked(
-                profile_name=profile_name,
-                remote=remote,
-                host=host,
-                identity=identity,
-                repository=repository,
-                org=org,
-                role_kind=role_kind,
-                expected_username=expected_username,
-                source=source,
-            )
-        return _SESSION_CONTEXT.as_dict()
-
-
-def assess_session_context(
-    *,
-    profile_name: str | None,
-    remote: str | None,
-    host: str | None = None,
-    identity: str | None = None,
-    repository: str | None = None,
-    org: str | None = None,
-    expected_username: str | None = None,
-    require_bound: bool = False,
-    require_complete: bool = False,
-) -> dict[str, Any]:
-    """Compare live values against the bound session context.
-
-    Returns ``proven`` / ``block`` / ``reasons``. When unbound and
-    ``require_bound`` is false, does not block (caller may seed). When
-    unbound and ``require_bound`` is true, fails closed. ``require_complete``
-    additionally fails closed when the binding carries no verified
-    repository/organization identity, so a mutation can never run against an
-    unknown repository (#714).
-    """
-    reasons: list[str] = []
-    with _SESSION_CONTEXT_LOCK:
-        bound = _SESSION_CONTEXT
-        ctx = bound.as_dict() if bound is not None else None
-    if ctx is None or ctx.get("pid") != os.getpid():
-        if require_bound:
-            reasons.append(
-                "session mutation context is unbound; call gitea_whoami or "
-                "gitea_activate_profile before mutating (fail closed)"
-            )
-            return _assessment(False, reasons, ctx)
-        return _assessment(True, reasons, ctx)
-
-    if require_complete and not (ctx.get("repository") and ctx.get("org")):
-        reasons.append(
-            "session repository/organization identity is unverified "
-            f"(repository={ctx.get('repository')!r}, org={ctx.get('org')!r}); "
-            "a mutation cannot proceed without a verified workspace "
-            "repository (fail closed)"
-        )
-        return _assessment(False, reasons, ctx)
-
-    live_profile = (profile_name or "").strip() or None
-    live_remote = (remote or "").strip() or None
-    live_host = (host or "").strip().lower() or None
-    live_identity = (identity or "").strip() or None
-    live_repo = (repository or "").strip() or None
-    live_org = (org or "").strip() or None
-
-    if ctx.get("profile_name") and live_profile and live_profile != ctx.get("profile_name"):
-        reasons.append(
-            f"profile drift: live '{live_profile}' != bound "
-            f"'{ctx.get('profile_name')}' (fail closed)"
-        )
-    if ctx.get("remote") and live_remote and live_remote != ctx.get("remote"):
-        reasons.append(
-            f"remote drift: live '{live_remote}' != bound "
-            f"'{ctx.get('remote')}' (fail closed)"
-        )
-    if ctx.get("host") and live_host and live_host != ctx.get("host"):
-        reasons.append(
-            f"host drift: live '{live_host}' != bound "
-            f"'{ctx.get('host')}' (fail closed)"
-        )
-    if (
-        ctx.get("identity")
-        and live_identity
-        and live_identity != ctx.get("identity")
-    ):
-        reasons.append(
-            f"identity drift: live '{live_identity}' != bound "
-            f"'{ctx.get('identity')}' (fail closed)"
-        )
-    if ctx.get("repository") and live_repo and live_repo != ctx.get("repository"):
-        reasons.append(
-            f"repository drift: live '{live_repo}' != bound "
-            f"'{ctx.get('repository')}' (fail closed)"
-        )
-    if ctx.get("org") and live_org and live_org != ctx.get("org"):
-        reasons.append(
-            f"org drift: live '{live_org}' != bound "
-            f"'{ctx.get('org')}' (fail closed)"
-        )
-
-    expected = expected_username or ctx.get("expected_username")
-    if expected and live_identity and live_identity != expected:
-        reasons.append(
-            f"identity mismatch: authenticated '{live_identity}' != "
-            f"profile expected '{expected}' (fail closed)"
-        )
-
-    return _assessment(not reasons, reasons, ctx)
-
-
-def assess_identity_match(
-    *,
-    authenticated: str | None,
-    expected_username: str | None,
-) -> dict[str, Any]:
-    """Fail closed when profile declares a username that does not match live auth."""
-    reasons: list[str] = []
-    auth = (authenticated or "").strip() or None
-    expected = (expected_username or "").strip() or None
-    if expected and auth and auth != expected:
-        reasons.append(
-            f"identity mismatch: authenticated '{auth}' != "
-            f"profile expected '{expected}' (fail closed)"
-        )
-    return {
-        "proven": not reasons,
-        "block": bool(reasons),
-        "reasons": reasons,
-        "authenticated": auth,
-        "expected_username": expected,
-    }
-
-
-_REPO_SLUG_RE = re.compile(r"^\s*(?P<org>[^/\s]+)\s*/\s*(?P<repo>[^/\s]+?)(?:\.git)?\s*$")
-
-
-def parse_repository_slug(slug: str | None) -> tuple[str, str] | None:
-    """Split a canonical ``owner/repository`` slug, or None when unparseable."""
-    match = _REPO_SLUG_RE.match(slug or "")
-    if not match:
-        return None
-    return match.group("org"), match.group("repo")
-
-
-def format_repository_slug(org: str | None, repo: str | None) -> str | None:
-    """``owner/repository`` from parts, or None when either side is missing."""
-    left = (org or "").strip()
-    right = (repo or "").strip()
-    if not left or not right:
-        return None
-    return f"{left}/{right}"
-
-
-def declared_allowed_repositories(
-    profile: Mapping[str, Any] | None,
-    *,
-    strict: bool = False,
-) -> list[str]:
-    """Canonical ``owner/repository`` authorization boundary declared by *profile*.
-
-    This list is an authorization boundary, never the session binding itself:
-    the verified workspace selects exactly one entry (see
-    :func:`assess_repository_scope`).
-
-    When *strict* is true (mutation path), missing profile, non-list values, or
-    any malformed entry raise ``ValueError`` instead of silently collapsing to
-    an empty scope (which would fail open). Read-only diagnostics may use
-    strict=False.
-    """
-    if not profile:
-        if strict:
-            raise ValueError(
-                "profile unresolved; cannot declare allowed_repositories "
-                "(fail closed)"
-            )
-        return []
-    raw = profile.get("allowed_repositories")
-    if raw is None:
-        return []
-    if not isinstance(raw, (list, tuple)):
-        if strict:
-            raise ValueError(
-                "allowed_repositories must be a list of owner/repository slugs "
-                "(fail closed)"
-            )
-        return []
-    slugs: list[str] = []
-    errors: list[str] = []
-    for entry in raw:
-        if not isinstance(entry, str):
-            errors.append(f"non-string allowed_repositories entry {entry!r}")
-            continue
-        parsed = parse_repository_slug(entry)
-        if not parsed:
-            errors.append(
-                f"malformed allowed_repositories entry {entry!r} "
-                "(expected owner/repository)"
-            )
-            continue
-        slugs.append(f"{parsed[0]}/{parsed[1]}")
-    if strict and errors:
-        raise ValueError("; ".join(errors) + " (fail closed)")
-    return slugs
-
-
-def assess_repository_scope(
-    *,
-    workspace_slug: str | None,
-    allowed: list[str] | None,
-    profile_name: str | None = None,
-    require_scope: bool = False,
-) -> dict[str, Any]:
-    """Authorize the verified workspace repository against the profile allowlist.
-
-    The workspace-derived slug is the only candidate: a profile that authorizes
-    several repositories still binds to the single verified one, and never to
-    the list as a whole.
-
-    *require_scope* (mutation path): missing/empty allowlist fails closed. The
-    workspace remote cannot self-authorize without a configured boundary.
-    """
-    scope = list(allowed or [])
-    reasons: list[str] = []
-    name = profile_name or "(active profile)"
-    if not scope:
-        if require_scope:
-            reasons.append(
-                f"mutation denied: profile '{name}' has no non-empty "
-                "allowed_repositories authorization boundary (fail closed)"
-            )
-            return {
-                "proven": False,
-                "block": True,
-                "reasons": reasons,
-                "scope_enforced": True,
-                "workspace_slug": workspace_slug,
-                "allowed_repositories": [],
-            }
-        return {
-            "proven": True,
-            "block": False,
-            "reasons": reasons,
-            "scope_enforced": False,
-            "workspace_slug": workspace_slug,
-            "allowed_repositories": [],
-        }
-    if not workspace_slug:
-        reasons.append(
-            f"no verified workspace repository could be established for profile "
-            f"'{name}'; repository scope cannot be authorized (fail closed)"
-        )
-    elif workspace_slug.lower() not in {entry.lower() for entry in scope}:
-        reasons.append(
-            f"repository scope denial: workspace repository '{workspace_slug}' "
-            f"is not authorized by profile '{name}' allowed_repositories "
-            f"{sorted(scope)} (fail closed)"
-        )
-    return {
-        "proven": not reasons,
-        "block": bool(reasons),
-        "reasons": reasons,
-        "scope_enforced": True,
-        "workspace_slug": workspace_slug,
-        "allowed_repositories": sorted(scope),
-    }
-
-
-def assess_repository_override(
-    *,
-    requested_org: str | None,
-    requested_repo: str | None,
-    bound_org: str | None,
-    bound_repo: str | None,
-) -> dict[str, Any]:
-    """Caller-supplied org/repo must agree with the immutable binding.
-
-    A mutation request is never allowed to establish, complete, or replace the
-    binding — it may only be checked against it.
-    """
-    reasons: list[str] = []
-    req_org = (requested_org or "").strip() or None
-    req_repo = (requested_repo or "").strip() or None
-    if req_repo and bound_repo and req_repo.lower() != bound_repo.lower():
-        reasons.append(
-            f"repository override denial: request targets '{req_repo}' but the "
-            f"session is bound to '{bound_repo}' (fail closed)"
-        )
-    if req_org and bound_org and req_org.lower() != bound_org.lower():
-        reasons.append(
-            f"organization override denial: request targets '{req_org}' but the "
-            f"session is bound to '{bound_org}' (fail closed)"
-        )
-    return {"proven": not reasons, "block": bool(reasons), "reasons": reasons}
-
-
-def filter_profiles_for_remote(
-    config: dict | None,
-    remote: str | None,
-    remotes: dict | None,
-) -> list[str]:
-    """Profile names whose host/context matches *remote* (enabled only)."""
-    if not config or not remote:
-        return []
-    profiles = config.get("profiles") or {}
-    contexts = config.get("contexts") or {}
-    names: list[str] = []
-    for name, data in profiles.items():
-        if not isinstance(data, dict):
-            continue
-        if not data.get("enabled", True):
-            continue
-        if profile_matches_remote(data, remote, remotes, contexts=contexts):
-            names.append(name)
-    return sorted(names)
-
-
-def profile_allowed_for_remote(
-    profile: dict | None,
-    remote: str | None,
-    remotes: dict | None,
-    *,
-    contexts: dict | None = None,
-) -> dict[str, Any]:
-    """Assess whether the active profile may serve *remote*."""
-    reasons: list[str] = []
-    if not profile:
-        reasons.append("active profile unresolved (fail closed)")
-        return {"proven": False, "block": True, "reasons": reasons}
-    if not remote:
-        return {"proven": True, "block": False, "reasons": reasons}
-    # Legacy env-only profiles have no configured base URL or v2 context. Their
-    # first call may establish the process remote/host pin; after that,
-    # assess_session_context rejects any drift. Configured v2 profiles still
-    # require positive host/context alignment here.
-    if not profile_host(profile) and not (profile.get("context") or "").strip():
-        return {"proven": True, "block": False, "reasons": reasons}
-    if not profile_matches_remote(profile, remote, remotes, contexts=contexts):
-        p_name = profile.get("profile_name") or profile.get("name") or "(unknown)"
-        p_host = profile_host(profile) or "(none)"
-        r_host = remote_host(remote, remotes) or "(none)"
-        reasons.append(
-            f"cross-host profile denial: profile '{p_name}' (host '{p_host}') "
-            f"cannot serve remote '{remote}' (host '{r_host}') (fail closed)"
-        )
-    return {"proven": not reasons, "block": bool(reasons), "reasons": reasons}
-
-
-def mutation_context_audit_fields(
-    ctx: Mapping[str, Any] | None = None,
-) -> dict[str, Any]:
-    """Fields to include in pre-mutation audit records."""
-    data = ctx if ctx is not None else get_session_context()
-    if not data:
-        return {
-            "session_context_bound": False,
-            "session_profile": None,
-            "session_remote": None,
-            "session_host": None,
-            "session_identity": None,
-            "session_repository": None,
-            "session_org": None,
-        }
-    return {
-        "session_context_bound": True,
-        "session_profile": data.get("profile_name"),
-        "session_remote": data.get("remote"),
-        "session_host": data.get("host"),
-        "session_identity": data.get("identity"),
-        "session_repository": data.get("repository"),
-        "session_org": data.get("org"),
-        "session_role_kind": data.get("role_kind"),
-        "session_context_source": data.get("source"),
-    }
-
-
-def _assessment(
-    proven: bool, reasons: list[str], ctx: Mapping[str, Any] | None
-) -> dict[str, Any]:
-    return {
-        "proven": proven,
-        "block": not proven,
-        "reasons": list(reasons),
-        "bound_context": dict(ctx) if ctx else None,
-        "audit": mutation_context_audit_fields(ctx),
-    }

stale_review_decision_lock.py

@@ -438,3 +438,252 @@ def format_cleanup_audit_comment(audit: dict[str, Any]) -> str:
         "This path only clears a lock when the referenced PR is merged/closed.",
     ]
     return "\n".join(lines)
+
+
+# ---------------------------------------------------------------------------
+# #709 cross-profile cleanup, overwrite protection, recovery provenance
+# ---------------------------------------------------------------------------
+
+
+def has_unresolved_terminal_evidence(lock: dict | None) -> bool:
+    """True when *lock* still carries a terminal live mutation ledger."""
+    return last_terminal_mutation(lock) is not None
+
+
+def assess_init_overwrite(
+    existing_lock: dict | None,
+    *,
+    force: bool = False,
+) -> dict[str, Any]:
+    """Whether init_review_decision_lock may replace an existing durable lock (#709 AC2).
+
+    Unresolved terminal evidence must never be silently replaced by an empty
+    initialized lock. *force* does not authorize destruction of terminal
+    ledgers — only explicit cleanup / archive transitions may remove them.
+    """
+    result: dict[str, Any] = {
+        "overwrite_allowed": True,
+        "has_existing": existing_lock is not None,
+        "has_unresolved_terminal": False,
+        "reasons": [],
+        "last_terminal_pr": None,
+        "last_terminal_action": None,
+        "existing_profile_identity": None,
+    }
+    if existing_lock is None:
+        result["reasons"].append("no existing lock; empty init allowed")
+        return result
+    result["existing_profile_identity"] = (
+        existing_lock.get("profile_identity")
+        or existing_lock.get("session_profile_lock")
+        or existing_lock.get("session_profile")
+    )
+    last = last_terminal_mutation(existing_lock)
+    if last is None:
+        result["reasons"].append(
+            "existing lock has no terminal mutations; re-init allowed"
+        )
+        return result
+    result["has_unresolved_terminal"] = True
+    result["overwrite_allowed"] = False
+    result["last_terminal_pr"] = last.get("pr_number")
+    result["last_terminal_action"] = last.get("action")
+    result["reasons"].append(
+        "refuse empty re-init: unresolved terminal decision-lock evidence "
+        f"present ({last.get('action')} on PR #{last.get('pr_number')}); "
+        "use gitea_cleanup_stale_review_decision_lock when moot, or archive "
+        f"via sanctioned recovery — force={force!r} does not authorize overwrite "
+        "(#709 AC2)"
+    )
+    return result
+
+
+def lock_targets_merged_pr_approval(
+    lock: dict | None,
+    *,
+    pr_number: int,
+    expected_head_sha: str | None = None,
+) -> bool:
+    """True when *lock*'s last terminal mutation is approve of *pr_number*.
+
+    When *expected_head_sha* is provided, a **recorded** terminal head must
+    match. Legacy same-repo approve locks with no recorded head do **not**
+    match (fail closed, #709 F3 residual / review 435) — callers must use the
+    strict secondary path that refuses no-head clears.
+    """
+    last = last_terminal_mutation(lock)
+    if last is None:
+        return False
+    if last.get("action") != "approve":
+        return False
+    if last.get("pr_number") != pr_number:
+        return False
+    if expected_head_sha:
+        want = normalize_head_sha(expected_head_sha)
+        if not want:
+            return False
+        locked = mutation_head_sha(last, lock)
+        # Require recorded-head match; unrecorded head is not a match (#709 F3).
+        if not locked or not heads_equal(locked, want):
+            return False
+    return True
+
+
+def build_post_merge_recovery_record(
+    *,
+    pr_number: int,
+    head_sha: str | None,
+    merge_commit_sha: str | None,
+    target_profile_identity: str | None,
+    failed_step: str,
+    error: str | None,
+    remote: str | None,
+    org: str | None,
+    repo: str | None,
+    actor_username: str | None,
+    profile_name: str | None,
+) -> dict[str, Any]:
+    """Durable recovery-required payload after irreversible merge (#709 AC3)."""
+    return {
+        "event": "post_merge_decision_lock_recovery_required",
+        "status": "recovery_required",
+        "issue_ref": "#709",
+        "recovery_critical": True,
+        "applied": False,  # never claim historical cleanup succeeded
+        "timestamp": datetime.now(timezone.utc).isoformat(),
+        "pr_number": pr_number,
+        "head_sha": normalize_head_sha(head_sha),
+        "merge_commit_sha": merge_commit_sha,
+        "target_profile_identity": target_profile_identity,
+        "failed_step": failed_step,
+        "error": error,
+        "remote": remote,
+        "org": org,
+        "repo": repo,
+        "actor_username": actor_username,
+        "profile_name": profile_name,
+        "required_recovery_action": (
+            "retry cross-profile decision-lock cleanup and audit publication "
+            "for the merged PR; if terminal evidence is gone, use "
+            "gitea_record_irrecoverable_decision_lock_provenance"
+        ),
+    }
+
+
+def build_irrecoverable_provenance_record(
+    *,
+    pr_number: int,
+    head_sha: str | None,
+    remote: str | None,
+    org: str | None,
+    repo: str | None,
+    actor_username: str | None,
+    profile_name: str | None,
+    reason: str,
+    incident_ref: str | None = None,
+    # Deprecated kwargs retained only so stale call sites fail closed:
+    operator_authorized: bool | None = None,
+    # Required for merger-acceptable records (#709 F1):
+    authorization: dict[str, Any] | None = None,
+    incident_issue: int | None = None,
+    incident_comment_id: int | None = None,
+    destroyed_subject: str | None = None,
+    historical_provenance_subject: str | None = None,
+) -> dict[str, Any]:
+    """Truthful absence-of-proof record (#709 AC5). Never sets applied=True.
+
+    Caller-supplied ``operator_authorized`` is **ignored** as authorization
+    evidence (review 434 F1). Prefer
+    :func:`irrecoverable_provenance.build_irrecoverable_provenance_record`
+    with a verified server-side authorization artifact.
+    """
+    # Explicitly ignore deprecated self-assertable Boolean.
+    _ = operator_authorized
+    if authorization is not None and incident_issue is not None and incident_comment_id is not None:
+        from irrecoverable_provenance import (
+            build_irrecoverable_provenance_record as _build,
+        )
+
+        return _build(
+            pr_number=int(pr_number),
+            head_sha=str(head_sha or ""),
+            remote=str(remote or ""),
+            org=str(org or ""),
+            repo=str(repo or ""),
+            actor_username=actor_username,
+            profile_name=profile_name,
+            reason=reason,
+            incident_issue=int(incident_issue),
+            incident_comment_id=int(incident_comment_id),
+            authorization=authorization,
+            destroyed_subject=destroyed_subject,
+            historical_provenance_subject=historical_provenance_subject
+            or destroyed_subject,
+        )
+    # Fail-closed skeleton when no server authorization is supplied: never
+    # sets merger_may_accept True (even if operator_authorized was True).
+    return {
+        "event": "irrecoverable_decision_lock_provenance",
+        "status": "provenance_irrecoverable",
+        "record_type": "irrecoverable_decision_provenance",
+        "operator_recovery_required": True,
+        "issue_ref": "#709",
+        "recovery_critical": True,
+        "applied": False,
+        "historical_cleanup_proven": False,
+        "timestamp": datetime.now(timezone.utc).isoformat(),
+        "pr_number": pr_number,
+        "head_sha": normalize_head_sha(head_sha),
+        "remote": remote,
+        "org": org,
+        "repo": repo,
+        "actor_username": actor_username,
+        "profile_name": profile_name,
+        "reason": reason,
+        "incident_ref": incident_ref,
+        "incident_issue": incident_issue,
+        "incident_comment_id": incident_comment_id,
+        "authorization_verified": False,
+        "merger_may_accept": False,
+        "acceptance_rule": (
+            "Merger may accept this record only when a server-side "
+            "authorization artifact verifies for remote/org/repo/PR/exact "
+            "head/incident, the record is durable and read back, and normal "
+            "merge gates still pass. Caller Booleans never authorize. This "
+            "does not prove historical cleanup."
+        ),
+    }
+
+
+def format_irrecoverable_audit_comment(record: dict[str, Any]) -> str:
+    """Markdown body for irrecoverable provenance audit (no applied=true claim)."""
+    from irrecoverable_provenance import (
+        format_irrecoverable_audit_comment as _fmt,
+    )
+
+    return _fmt(record)
+
+
+def format_post_merge_recovery_comment(record: dict[str, Any]) -> str:
+    """Markdown body for post-merge recovery-required audit."""
+    lines = [
+        "## Post-merge decision-lock recovery required (#709)",
+        "",
+        "Status: **RECOVERY_REQUIRED** (merge is irreversible; cleanup/audit incomplete)",
+        "",
+        f"- actor: `{record.get('actor_username')}`",
+        f"- profile: `{record.get('profile_name')}`",
+        f"- timestamp: `{record.get('timestamp')}`",
+        f"- PR: `#{record.get('pr_number')}`",
+        f"- head_sha: `{record.get('head_sha')}`",
+        f"- merge_commit_sha: `{record.get('merge_commit_sha')}`",
+        f"- target_profile_identity: `{record.get('target_profile_identity')}`",
+        f"- failed_step: `{record.get('failed_step')}`",
+        f"- error: `{record.get('error')}`",
+        "",
+        f"Required action: {record.get('required_recovery_action')}",
+        "",
+        "applied=false — do not treat this as successful cleanup evidence.",
+    ]
+    return "\n".join(lines)
+

task_capability_map.py

@@ -127,6 +127,32 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = {
         "permission": "gitea.pr.review",
         "role": "reviewer",
     },
+    # #709: truthful absence-of-proof recovery (server-side auth + record + consume).
+    # Dedicated mutation capability — gitea.read is insufficient (review 434 F1).
+    "issue_irrecoverable_provenance_authorization": {
+        "permission": "gitea.decision_lock.irrecoverable_recovery",
+        "role": "reconciler",
+    },
+    "gitea_issue_irrecoverable_provenance_authorization": {
+        "permission": "gitea.decision_lock.irrecoverable_recovery",
+        "role": "reconciler",
+    },
+    "record_irrecoverable_decision_lock_provenance": {
+        "permission": "gitea.decision_lock.irrecoverable_recovery",
+        "role": "reconciler",
+    },
+    "gitea_record_irrecoverable_decision_lock_provenance": {
+        "permission": "gitea.decision_lock.irrecoverable_recovery",
+        "role": "reconciler",
+    },
+    "consume_irrecoverable_decision_lock_provenance": {
+        "permission": "gitea.pr.merge",
+        "role": "merger",
+    },
+    "gitea_consume_irrecoverable_decision_lock_provenance": {
+        "permission": "gitea.pr.merge",
+        "role": "merger",
+    },
     "delete_branch": {
         "permission": "gitea.branch.delete",
         "role": "author",

tests/conftest.py

@@ -26,14 +26,6 @@ def _reset_mutation_authority(monkeypatch):
     Pin ``default_state_dir`` / ``DEFAULT_STATE_DIR`` to a per-test temp dir
     so durable load/save never touches host state even after env clears.
     """
-    import session_context_binding as session_ctx
-
-    # Each pytest item is an independent logical MCP session. Reset both before
-    # and after the item; the post-yield reset is in a finally block so an
-    # assertion, exception, or unittest teardown failure cannot pollute the
-    # next item. Production code has no automatic per-call reset path.
-    session_ctx._reset_session_context_for_testing()
-
     for env_key in [
         "GITEA_SESSION_PROFILE_LOCK",
         "GITEA_ACTIVE_WORKTREE",
@@ -88,15 +80,9 @@ def _reset_mutation_authority(monkeypatch):
     try:
         import mcp_server
     except Exception:
-        try:
-            yield
-        finally:
-            session_ctx._reset_session_context_for_testing()
         _state_tmp.cleanup()
+        yield
         return
-    import gitea_config
-
-    monkeypatch.setattr(gitea_config, "_active_profile_override", None)
     monkeypatch.setattr(mcp_server, "_MUTATION_AUTHORITY", None)
     monkeypatch.setattr(mcp_server, "_IDENTITY_CACHE", {})
     monkeypatch.setattr(mcp_server, "_REVIEW_DECISION_LOCK", None)
@@ -127,9 +113,7 @@ def _reset_mutation_authority(monkeypatch):
         capability_stop_terminal.clear()
     except Exception:
         pass
-    try:
     yield
-    finally:
     try:
         import capability_stop_terminal
         capability_stop_terminal.clear()
@@ -144,20 +128,4 @@ def _reset_mutation_authority(monkeypatch):
         mcp_server._REVIEW_DECISION_LOCK = None
     except Exception:
         pass
-        gitea_config._active_profile_override = None
-        session_ctx._reset_session_context_for_testing()
     _state_tmp.cleanup()
-
-
-# #714: deterministic workspace remotes only (no host git dependency).
-import pytest
-
-
-@pytest.fixture(autouse=True)
-def _deterministic_workspace_remotes():
-    try:
-        from mutation_profile_fixture import install_deterministic_remote_urls
-        install_deterministic_remote_urls()
-    except Exception:
-        pass
-    yield

tests/mutation_profile_fixture.py

@@ -1,483 +0,0 @@
-"""Centralized config-backed mutation fixture for #714.
-
-Mutation tests must opt into this helper explicitly. It never grants
-mutation authority via environment variables alone.
-
-Design rules:
-- temporary v2 configuration with non-empty ``allowed_repositories``
-- profile-scoped operations (not every mutation op for every test)
-- deterministic workspace remotes (no host Git dependency)
-- one canonical repository per profile by default
-- isolated state + cleanup in ``finally``
-"""
-from __future__ import annotations
-
-import json
-import os
-import tempfile
-from contextlib import contextmanager
-from copy import deepcopy
-from typing import Iterable, Sequence
-from unittest.mock import patch
-
-PRGS_SLUG = "Scaled-Tech-Consulting/Gitea-Tools"
-PRGS_URL = f"https://gitea.prgs.cc/{PRGS_SLUG}.git"
-MDCPS_SLUG = "913443/eAgenda"
-MDCPS_URL = f"https://gitea.dadeschools.net/{MDCPS_SLUG}.git"
-EXAMPLE_SLUG = "Example-Org/Example-Repo"
-EXAMPLE_URL = f"https://gitea.example.com/{EXAMPLE_SLUG}.git"
-TIMESHEET_SLUG = "Scaled-Tech-Consulting/Timesheet"
-
-
-def _author_ops() -> list[str]:
-    return [
-        "gitea.read",
-        "gitea.issue.create",
-        "gitea.issue.close",
-        "gitea.issue.comment",
-        "gitea.pr.create",
-        "gitea.pr.comment",
-        "gitea.branch.create",
-        "gitea.branch.push",
-        "gitea.repo.commit",
-    ]
-
-
-def _author_forbidden() -> list[str]:
-    return [
-        "gitea.pr.approve",
-        "gitea.pr.merge",
-        "gitea.pr.request_changes",
-        "gitea.pr.review",
-    ]
-
-
-def _reviewer_ops() -> list[str]:
-    return [
-        "gitea.read",
-        "gitea.pr.review",
-        "gitea.pr.approve",
-        "gitea.pr.comment",
-        "gitea.issue.comment",
-    ]
-
-
-def _merger_ops() -> list[str]:
-    return [
-        "gitea.read",
-        "gitea.pr.merge",
-        "gitea.pr.comment",
-    ]
-
-
-def _profile(
-    *,
-    name: str,
-    context: str,
-    role: str,
-    base_url: str,
-    auth_env: str,
-    allowed_operations: Sequence[str],
-    forbidden_operations: Sequence[str],
-    allowed_repositories: Sequence[str],
-    username: str | None = None,
-) -> dict:
-    body = {
-        "enabled": True,
-        "context": context,
-        "role": role,
-        "base_url": base_url,
-        "auth": {"type": "env", "name": auth_env},
-        "allowed_operations": list(allowed_operations),
-        "forbidden_operations": list(forbidden_operations),
-        "allowed_repositories": list(allowed_repositories),
-        "execution_profile": name,
-    }
-    if username:
-        body["username"] = username
-    return body
-
-
-def build_dual_remote_config(
-    *,
-    allowed_operations: Iterable[str] | None = None,
-    allowed_repositories_prgs: Sequence[str] | None = None,
-    allowed_repositories_mdcps: Sequence[str] | None = None,
-    extra_profiles: dict | None = None,
-    include_example_repo: bool = False,
-) -> dict:
-    """Build a minimal dual-host v2 config.
-
-    Each profile authorizes only its host-canonical repository by default.
-    ``include_example_repo`` adds Example-Org/Example-Repo when a test patches
-    REMOTES to that synthetic unit-test identity.
-    """
-    ops = list(allowed_operations or _author_ops())
-    forb = _author_forbidden()
-    prgs_repos = list(allowed_repositories_prgs or [PRGS_SLUG])
-    mdcps_repos = list(allowed_repositories_mdcps or [MDCPS_SLUG])
-    if include_example_repo:
-        for repos in (prgs_repos, mdcps_repos):
-            if EXAMPLE_SLUG not in repos:
-                repos.append(EXAMPLE_SLUG)
-
-    profiles = {
-        "test-author-prgs": _profile(
-            name="test-author-prgs",
-            context="prgs",
-            role="author",
-            base_url="https://gitea.prgs.cc",
-            auth_env="GITEA_TOKEN_TEST",
-            allowed_operations=ops,
-            forbidden_operations=forb,
-            allowed_repositories=prgs_repos,
-        ),
-        "test-author-dadeschools": _profile(
-            name="test-author-dadeschools",
-            context="mdcps",
-            role="author",
-            base_url="https://gitea.dadeschools.net",
-            auth_env="GITEA_TOKEN_TEST",
-            allowed_operations=ops,
-            forbidden_operations=forb,
-            allowed_repositories=mdcps_repos,
-        ),
-        "test-reviewer-prgs": _profile(
-            name="test-reviewer-prgs",
-            context="prgs",
-            role="reviewer",
-            base_url="https://gitea.prgs.cc",
-            auth_env="GITEA_TOKEN_TEST",
-            allowed_operations=_reviewer_ops(),
-            forbidden_operations=[
-                "gitea.pr.create",
-                "gitea.branch.push",
-                "gitea.pr.merge",
-                "gitea.issue.create",
-            ],
-            allowed_repositories=prgs_repos,
-        ),
-        "test-merger-prgs": _profile(
-            name="test-merger-prgs",
-            context="prgs",
-            role="merger",
-            base_url="https://gitea.prgs.cc",
-            auth_env="GITEA_TOKEN_TEST",
-            allowed_operations=_merger_ops(),
-            forbidden_operations=[
-                "gitea.pr.create",
-                "gitea.branch.push",
-                "gitea.pr.approve",
-                "gitea.issue.create",
-            ],
-            allowed_repositories=prgs_repos,
-        ),
-    }
-    if extra_profiles:
-        profiles.update(deepcopy(extra_profiles))
-
-    # Legacy aliases used by older tests — same host scope as the base profile.
-    for alias, base in (
-        ("gitea-author", "test-author-prgs"),
-        ("author-test", "test-author-dadeschools"),
-        ("author", "test-author-dadeschools"),
-        ("full-author", "test-author-dadeschools"),
-        ("test-author", "test-author-dadeschools"),
-        ("prgs-author", "test-author-prgs"),
-        ("gitea-reviewer", "test-reviewer-prgs"),
-        ("prgs-reviewer", "test-reviewer-prgs"),
-        ("gitea-merger", "test-merger-prgs"),
-        ("prgs-merger", "test-merger-prgs"),
-    ):
-        if alias not in profiles and base in profiles:
-            clone = deepcopy(profiles[base])
-            clone["execution_profile"] = alias
-            profiles[alias] = clone
-
-    return {
-        "version": 2,
-        "rules": {"allow_runtime_switching": True},
-        "contexts": {
-            "prgs": {
-                "enabled": True,
-                "gitea": {"enabled": True, "base_url": "https://gitea.prgs.cc"},
-            },
-            "mdcps": {
-                "enabled": True,
-                "gitea": {
-                    "enabled": True,
-                    "base_url": "https://gitea.dadeschools.net",
-                },
-            },
-        },
-        "profiles": profiles,
-    }
-
-
-def build_v2_config(**kwargs):
-    """Back-compat alias."""
-    return build_dual_remote_config(
-        allowed_operations=kwargs.get("allowed_operations"),
-        extra_profiles=kwargs.get("extra_profiles"),
-        include_example_repo=bool(kwargs.get("include_example_repo")),
-    )
-
-
-def inject_allowed_repositories(
-    config: dict,
-    repos: Sequence[str],
-    *,
-    profiles: Sequence[str] | None = None,
-) -> dict:
-    """Return a deep copy of *config* with allowlists set on selected profiles."""
-    out = deepcopy(config)
-    targets = set(profiles) if profiles is not None else set(out.get("profiles") or {})
-    for name, prof in (out.get("profiles") or {}).items():
-        if name in targets:
-            prof["allowed_repositories"] = list(repos)
-    return out
-
-
-def ensure_all_profiles_have_scope(
-    config: dict,
-    default_repos: Sequence[str],
-) -> dict:
-    """Add non-empty allowed_repositories to every profile that lacks one."""
-    out = deepcopy(config)
-    for _name, prof in (out.get("profiles") or {}).items():
-        raw = prof.get("allowed_repositories")
-        if not isinstance(raw, (list, tuple)) or not raw:
-            prof["allowed_repositories"] = list(default_repos)
-    return out
-
-
-@contextmanager
-def mutation_profile_env(
-    *,
-    profile_name: str = "test-author-dadeschools",
-    allowed_operations: Iterable[str] | None = None,
-    allowed_repositories: Sequence[str] | None = None,
-    workspace_urls: dict | None = None,
-    clear_env: bool = False,
-    extra_env: dict | None = None,
-    extra_profiles: dict | None = None,
-    include_example_repo: bool = False,
-    config: dict | None = None,
-):
-    """Context manager: config-backed mutation authority + deterministic remotes."""
-    import gitea_config
-    import gitea_mcp_server as srv
-    import session_context_binding as session_ctx
-
-    if config is None:
-        prgs_repos = None
-        mdcps_repos = None
-        if allowed_repositories is not None:
-            # Caller-specified single allowlist applied to both host profiles.
-            prgs_repos = list(allowed_repositories)
-            mdcps_repos = list(allowed_repositories)
-        cfg = build_dual_remote_config(
-            allowed_operations=allowed_operations,
-            allowed_repositories_prgs=prgs_repos,
-            allowed_repositories_mdcps=mdcps_repos,
-            extra_profiles=extra_profiles,
-            include_example_repo=include_example_repo,
-        )
-    else:
-        cfg = deepcopy(config)
-        if allowed_repositories is not None:
-            cfg = ensure_all_profiles_have_scope(cfg, list(allowed_repositories))
-
-    urls = (
-        {"prgs": PRGS_URL, "dadeschools": MDCPS_URL}
-        if workspace_urls is None
-        else dict(workspace_urls)
-    )
-
-    tmp = tempfile.TemporaryDirectory(prefix="gitea-mut-cfg-")
-    try:
-        cfg_path = os.path.join(tmp.name, "profiles.json")
-        with open(cfg_path, "w", encoding="utf-8") as fh:
-            json.dump(cfg, fh)
-        env = {
-            "GITEA_MCP_CONFIG": cfg_path,
-            "GITEA_MCP_PROFILE": profile_name,
-            "GITEA_TOKEN_TEST": "test-token",
-            "PYTEST_CURRENT_TEST": os.environ.get(
-                "PYTEST_CURRENT_TEST", "mutation_profile_env"
-            ),
-        }
-        if extra_env:
-            env.update(extra_env)
-
-        def _remote_url(name: str):
-            if name in urls:
-                return urls[name]
-            # Prefer live REMOTES (tests often patch Example-Org).
-            try:
-                entry = srv.REMOTES.get(name) or {}
-                host = entry.get("host")
-                org = entry.get("org")
-                repo = entry.get("repo")
-                if host and org and repo:
-                    if (
-                        name == "prgs"
-                        and org == "Scaled-Tech-Consulting"
-                        and repo == "Timesheet"
-                    ):
-                        return PRGS_URL
-                    if name == "dadeschools" and repo == "Timesheet":
-                        return MDCPS_URL
-                    return f"https://{host}/{org}/{repo}.git"
-            except Exception:
-                pass
-            return None
-
-        gitea_config._active_profile_override = None
-        try:
-            session_ctx._reset_session_context_for_testing()
-        except Exception:
-            pass
-
-        payload = {
-            "env": env,
-            "config_path": cfg_path,
-            "profile_name": profile_name,
-            "urls": urls,
-            "config": cfg,
-        }
-        with patch.dict(os.environ, env, clear=clear_env):
-            os.environ.setdefault(
-                "PYTEST_CURRENT_TEST",
-                env.get("PYTEST_CURRENT_TEST", "mutation_profile_env"),
-            )
-            with patch.object(srv, "_local_git_remote_url", side_effect=_remote_url):
-                mcp_srv = None
-                try:
-                    import mcp_server as mcp_srv  # type: ignore
-                except Exception:
-                    mcp_srv = None
-                if mcp_srv is not None:
-                    with patch.object(
-                        mcp_srv, "_local_git_remote_url", side_effect=_remote_url
-                    ):
-                        yield payload
-                else:
-                    yield payload
-    finally:
-        try:
-            session_ctx._reset_session_context_for_testing()
-        except Exception:
-            pass
-        gitea_config._active_profile_override = None
-        tmp.cleanup()
-
-
-# Process-lifetime shared config for mass-migrating env-only mutation tests.
-_SHARED_CFG_PATH = None
-_SHARED_CFG_DIR = None
-_SHARED_CFG_WITH_EXAMPLE_PATH = None
-
-
-def shared_mutation_config_path(*, include_example_repo: bool = False) -> str:
-    """Write (once) a dual-remote mutation config and return its path."""
-    global _SHARED_CFG_PATH, _SHARED_CFG_DIR, _SHARED_CFG_WITH_EXAMPLE_PATH
-    import atexit
-    import shutil
-
-    if include_example_repo:
-        if _SHARED_CFG_WITH_EXAMPLE_PATH and os.path.isfile(
-            _SHARED_CFG_WITH_EXAMPLE_PATH
-        ):
-            return _SHARED_CFG_WITH_EXAMPLE_PATH
-        if _SHARED_CFG_DIR is None:
-            _SHARED_CFG_DIR = tempfile.mkdtemp(prefix="gitea-shared-mut-cfg-")
-            atexit.register(lambda: shutil.rmtree(_SHARED_CFG_DIR, ignore_errors=True))
-        path = os.path.join(_SHARED_CFG_DIR, "profiles-with-example.json")
-        with open(path, "w", encoding="utf-8") as fh:
-            json.dump(build_dual_remote_config(include_example_repo=True), fh)
-        _SHARED_CFG_WITH_EXAMPLE_PATH = path
-        return path
-
-    if _SHARED_CFG_PATH and os.path.isfile(_SHARED_CFG_PATH):
-        return _SHARED_CFG_PATH
-    if _SHARED_CFG_DIR is None:
-        _SHARED_CFG_DIR = tempfile.mkdtemp(prefix="gitea-shared-mut-cfg-")
-        atexit.register(lambda: shutil.rmtree(_SHARED_CFG_DIR, ignore_errors=True))
-    _SHARED_CFG_PATH = os.path.join(_SHARED_CFG_DIR, "profiles.json")
-    with open(_SHARED_CFG_PATH, "w", encoding="utf-8") as fh:
-        json.dump(build_dual_remote_config(), fh)
-    return _SHARED_CFG_PATH
-
-
-def shared_mutation_env(
-    profile_name: str = "test-author-dadeschools",
-    *,
-    include_example_repo: bool = False,
-    **extra,
-) -> dict:
-    """Env dict for mutation tests: config-backed + optional extras.
-
-    Always carries ``PYTEST_CURRENT_TEST`` so ``patch.dict(..., clear=True)``
-    does not strip the pytest marker and trip the #695 native-transport wall.
-    """
-    env = {
-        "GITEA_MCP_CONFIG": shared_mutation_config_path(
-            include_example_repo=include_example_repo
-        ),
-        "GITEA_MCP_PROFILE": profile_name,
-        "GITEA_TOKEN_TEST": "test-token",
-        "PYTEST_CURRENT_TEST": os.environ.get(
-            "PYTEST_CURRENT_TEST", "mutation_profile_env"
-        ),
-    }
-    env.update(extra)
-    # Callers must not be able to drop the pytest marker by accident.
-    env.setdefault(
-        "PYTEST_CURRENT_TEST",
-        os.environ.get("PYTEST_CURRENT_TEST", "mutation_profile_env"),
-    )
-    return env
-
-
-def install_deterministic_remote_urls() -> None:
-    """Patch server modules so workspace remotes are deterministic.
-
-    Prefer live ``REMOTES`` entries (so tests that patch Example-Org keep
-    workspace alignment). Map the historical prgs→Timesheet default to
-    Gitea-Tools so session binding matches the control repository under test.
-    """
-    import gitea_mcp_server as srv
-
-    def _url(name: str):
-        try:
-            entry = srv.REMOTES.get(name) or {}
-            host = entry.get("host")
-            org = entry.get("org")
-            repo = entry.get("repo")
-            if host and org and repo:
-                if (
-                    name == "prgs"
-                    and org == "Scaled-Tech-Consulting"
-                    and repo == "Timesheet"
-                ):
-                    return PRGS_URL
-                if name == "dadeschools" and repo == "Timesheet":
-                    # Prefer mdcps eAgenda canonical for dual-config tests.
-                    return MDCPS_URL
-                return f"https://{host}/{org}/{repo}.git"
-        except Exception:
-            pass
-        if name == "prgs":
-            return PRGS_URL
-        if name == "dadeschools":
-            return MDCPS_URL
-        return None
-
-    srv._local_git_remote_url = _url  # type: ignore[method-assign]
-    try:
-        import mcp_server as mcp_srv
-
-        mcp_srv._local_git_remote_url = _url  # type: ignore[method-assign]
-    except Exception:
-        pass

tests/test_agent_temp_artifacts.py

@@ -1,7 +1,3 @@
-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
 """Tests for agent temp artifact detection and preflight warnings (#261)."""
 import json
 import os
@@ -69,9 +65,11 @@ class TestPreflightWarnings(unittest.TestCase):
 # Issue-write tools are profile-gated (#69); gitea_lock_issue requires
 # gitea.issue.comment (see task_capability_map), so the gate must be
 # seeded exactly like tests/test_mcp_server.py::TestIssueLocking (#359).
-ISSUE_WRITE_ENV = shared_mutation_env(
+ISSUE_WRITE_ENV = {
-    "test-author-prgs",
+    "GITEA_ALLOWED_OPERATIONS": (
-)
+        "gitea.issue.create,gitea.issue.close,gitea.issue.comment"
+    ),
+}
 
 
 class TestIssueLockArtifactWarning(unittest.TestCase):

tests/test_assess_conflict_fix_push_tool.py

@@ -1,7 +1,3 @@
-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
 """Regression tests for gitea_assess_conflict_fix_push structured failures (#519)."""
 
 import os

tests/test_audit.py

@@ -1,7 +1,3 @@
-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
 """Tests for Gitea MCP mutating-action audit logging (issue #18).
 
 Covers the pure audit module (redaction, event building, sink writes) and the
@@ -165,23 +161,11 @@ class _AuditWiringBase(unittest.TestCase):
         self._dir.cleanup()
 
     def _env(self, **extra):
-        # Default: prgs-aligned author with create/close (remote=prgs tests).
+        env = {"GITEA_AUDIT_LOG": self.audit_path,
-        # Callers override GITEA_MCP_PROFILE for merger/reviewer paths.
+               "GITEA_PROFILE_NAME": "gitea-author",
-        profile = extra.pop("GITEA_MCP_PROFILE", None) or extra.pop(
+               "GITEA_ALLOWED_OPERATIONS": (
-            "GITEA_PROFILE_NAME", None
+                   "read,merge,gitea.issue.create,gitea.issue.close")}
-        ) or "test-author-prgs"
-        env = shared_mutation_env(
-            profile,
-            GITEA_AUDIT_LOG=self.audit_path,
-            GITEA_TOKEN_TEST="test-token",
-        )
-        # Strip legacy env-only authority knobs if callers still pass them;
-        # config-backed profiles own operations and repositories (#714).
-        extra.pop("GITEA_ALLOWED_OPERATIONS", None)
-        extra.pop("GITEA_FORBIDDEN_OPERATIONS", None)
-        extra.pop("GITEA_PROFILE_NAME", None)
         env.update(extra)
-        env["GITEA_MCP_PROFILE"] = profile
         return env
 
     def _records(self):
@@ -212,7 +196,7 @@ class TestSimpleToolAudit(_AuditWiringBase):
         rec = recs[0]
         self.assertEqual(rec["action"], "create_issue")
         self.assertEqual(rec["result"], "succeeded")
-        self.assertIn(rec["profile_name"], ("gitea-author", "test-author-prgs", "prgs-author"))
+        self.assertEqual(rec["profile_name"], "gitea-author")
         self.assertEqual(rec["authenticated_username"], "author-bot")
         self.assertEqual(rec["issue_number"], 11)
         self.assertEqual(rec["request_metadata"]["title"], "Add thing")
@@ -255,11 +239,10 @@ class TestSimpleToolAudit(_AuditWiringBase):
     def test_disabled_writes_nothing_and_no_extra_call(self, _auth, _get_all, mock_api, _role):
         # No GITEA_AUDIT_LOG -> audit is a no-op: one create POST, no file.
         mock_api.return_value = {"number": 1, "html_url": "http://x/1"}
-        with patch.dict(
+        with patch.dict(os.environ, {
-            os.environ,
+            "GITEA_PROFILE_NAME": "gitea-author",
-            shared_mutation_env("test-author-prgs"),
+            "GITEA_ALLOWED_OPERATIONS": "gitea.issue.create",
-            clear=True,
+        }, clear=True):
-        ):
             gitea_create_issue(title="x", remote="prgs")
         issue_posts = [
             c for c in mock_api.call_args_list if c.args[0] == "POST"
@@ -359,7 +342,8 @@ class TestGatedToolAudit(_AuditWiringBase):
             self._pr("author-bot"), approval,  # gate 7 feedback
             {}, {"merged_commit_sha": "c1"},
         ]
-        env = self._env(GITEA_MCP_PROFILE="test-merger-prgs")
+        env = self._env(GITEA_PROFILE_NAME="gitea-merger",
+                        GITEA_ALLOWED_OPERATIONS="read,merge")
         with patch.dict(os.environ, env, clear=True):
             mcp_server.gitea_load_review_workflow()
             r = gitea_merge_pr(pr_number=8, confirmation="MERGE PR 8",
@@ -378,7 +362,8 @@ class TestGatedToolAudit(_AuditWiringBase):
     def test_merge_blocked_audited(self, _auth, mock_api):
         # Self-author merge is blocked; must still be recorded as blocked.
         mock_api.side_effect = [{"login": "jcwalker3"}, self._pr("jcwalker3")]
-        env = self._env(GITEA_MCP_PROFILE="test-merger-prgs")
+        env = self._env(GITEA_PROFILE_NAME="gitea-merger",
+                        GITEA_ALLOWED_OPERATIONS="read,merge")
         with patch.dict(os.environ, env, clear=True):
             mcp_server.gitea_load_review_workflow()
             r = gitea_merge_pr(pr_number=8, confirmation="MERGE PR 8", remote="prgs")
@@ -400,23 +385,11 @@ class TestGatedToolAudit(_AuditWiringBase):
             [{"id": 7, "user": {"login": "reviewer-bot"}, "state": "APPROVED",
               "submitted_at": "2026-07-06T10:00:00Z", "dismissed": False}],
         ]
-        env = self._env(GITEA_MCP_PROFILE="test-reviewer-prgs")
+        env = self._env(GITEA_PROFILE_NAME="gitea-reviewer",
+                        GITEA_ALLOWED_OPERATIONS="read,review,approve")
         with patch.dict(os.environ, env, clear=True):
-            import session_context_binding as _sc
-            from tests.test_mcp_server import (
-                _init_reviewer_session,
-                _install_owned_reviewer_lease,
-            )
             from mcp_server import gitea_mark_final_review_decision
 
-            # Rebind after profile env is applied so durable session state
-            # matches the active reviewer profile (#714 / #695).
-            _sc._reset_session_context_for_testing()
-            _init_reviewer_session("prgs")
-            lease = _install_owned_reviewer_lease(8)
-            lease.start()
-            self.addCleanup(lease.stop)
-            mcp_server.gitea_load_review_workflow()
             gitea_mark_final_review_decision(
                 8, "approve", expected_head_sha="abc123", remote="prgs",
             )
@@ -425,10 +398,7 @@ class TestGatedToolAudit(_AuditWiringBase):
                 body="LGTM", remote="prgs",
                 final_review_decision_ready=True,
             )
-        self.assertTrue(
+        self.assertTrue(r["performed"])
-            r["performed"],
-            msg=f"submit_pr_review blocked: {r}",
-        )
         recs = self._records()
         self.assertEqual(len(recs), 1)
         self.assertEqual(recs[0]["action"], "submit_pr_review")

tests/test_commit_files_capability.py

@@ -29,7 +29,6 @@ CONFIG = {
             "allowed_operations": ["gitea.read", "gitea.repo.commit"],
             "forbidden_operations": ["gitea.pr.create"],
             "execution_profile": "commit-author",
-        "allowed_repositories": ["Example-Org/Example-Repo"],
         },
         "pr-only-author": {
             "enabled": True,
@@ -40,7 +39,6 @@ CONFIG = {
             "allowed_operations": ["gitea.read", "gitea.pr.create"],
             "forbidden_operations": ["gitea.repo.commit"],
             "execution_profile": "pr-only-author",
-        "allowed_repositories": ["Example-Org/Example-Repo"],
         },
         "reviewer-profile": {
             "enabled": True,
@@ -53,7 +51,6 @@ CONFIG = {
                 "gitea.repo.commit", "gitea.pr.create", "gitea.branch.push",
             ],
             "execution_profile": "reviewer-profile",
-        "allowed_repositories": ["Example-Org/Example-Repo"],
         },
     },
     "rules": {"allow_runtime_switching": False},

tests/test_commit_files_gate.py

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

tests/test_commit_payloads.py

@@ -35,7 +35,6 @@ CONFIG = {
                 "gitea.pr.review",
             ],
             "execution_profile": "full-author",
-        "allowed_repositories": ["Example-Org/Example-Repo"],
         },
     },
 }
@@ -228,7 +227,6 @@ class TestCommitPayloads(unittest.TestCase):
     @patch("mcp_server.api_request")
     @patch("mcp_server.get_auth_header", return_value="token author-pass")
     def test_commit_files_traversal_blocked(self, _auth, mock_api):
-        mock_api.return_value = {"login": "author-user"}
         # Remove active lock file to ensure it fails on traversal/invalid locks
         os.remove(self.lock_file_path)
 
@@ -250,7 +248,6 @@ class TestCommitPayloads(unittest.TestCase):
     @patch("mcp_server.api_request")
     @patch("mcp_server.get_auth_header", return_value="token author-pass")
     def test_commit_files_outside_scope_blocked(self, _auth, mock_api):
-        mock_api.return_value = {"login": "author-user"}
         with patch.dict(os.environ, self._env("full-author"), clear=True):
             with self.assertRaises(ValueError) as ctx:
                 mcp_server.gitea_commit_files(
@@ -269,7 +266,6 @@ class TestCommitPayloads(unittest.TestCase):
     @patch("mcp_server.api_request")
     @patch("mcp_server.get_auth_header", return_value="token author-pass")
     def test_commit_files_multiple_sources_blocked(self, _auth, mock_api):
-        mock_api.return_value = {"login": "author-user"}
         with patch.dict(os.environ, self._env("full-author"), clear=True):
             with self.assertRaises(ValueError) as ctx:
                 mcp_server.gitea_commit_files(

tests/test_config.py

@@ -1,7 +1,3 @@
-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
 """Tests for canonical JSON runtime-profile configuration (gitea_config) and
 its integration into gitea_auth.get_profile / get_auth_header.
 

tests/test_create_issue.py

@@ -1,4 +1,3 @@
-import os
 """Tests for create_issue.py.
 
 Every test mocks auth functions so no real network calls or keychain
@@ -13,7 +12,7 @@ import sys
 import tempfile
 import unittest
 import contextlib
-from unittest.mock import patch, MagicMock, patch
+from unittest.mock import MagicMock, patch
 
 # The module under test lives in the repo root, not a package.
 sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
@@ -93,26 +92,6 @@ class TestRemoteResolution(unittest.TestCase):
 # ---------------------------------------------------------------------------
 @unittest.skipIf(_SKIP, _REASON)
 class TestAPIPayload(unittest.TestCase):
-    """#714: omitted --org/--repo uses workspace-bound repo (Gitea-Tools),
-    not the historical REMOTES Timesheet default. Intentional security-policy
-    migration of legacy omitted-target assertions.
-    """
-
-    def setUp(self):
-        self._ws_env = patch.dict(
-            os.environ,
-            {
-                "GITEA_TEST_WORKSPACE_REMOTE_URL": (
-                    "https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools.git"
-                )
-            },
-            clear=False,
-        )
-        self._ws_env.start()
-
-    def tearDown(self):
-        self._ws_env.stop()
-
     """Ensure the JSON payload sent to Gitea is correct."""
 
     @patch("create_issue.get_credentials", return_value=FAKE_CREDS)
@@ -163,7 +142,7 @@ class TestAPIPayload(unittest.TestCase):
             url = mock_api.call_args[0][1]
             self.assertEqual(
                 url,
-                "https://gitea.prgs.cc/api/v1/repos/Scaled-Tech-Consulting/Gitea-Tools/issues",
+                "https://gitea.prgs.cc/api/v1/repos/Scaled-Tech-Consulting/Timesheet/issues",
             )
 
 

tests/test_create_pr.py

@@ -1,4 +1,3 @@
-import os
 """Tests for create_pr.py.
 
 Every test mocks `get_credentials` and `urllib.request.urlopen` so no real
@@ -9,7 +8,7 @@ import json
 import sys
 import unittest
 import contextlib
-from unittest.mock import patch, MagicMock, patch
+from unittest.mock import MagicMock, patch
 
 sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
 import create_pr  # noqa: E402
@@ -75,26 +74,6 @@ class TestRemoteResolution(unittest.TestCase):
 # API payload
 # ---------------------------------------------------------------------------
 class TestAPIPayload(unittest.TestCase):
-    """#714: omitted --org/--repo uses workspace-bound repo (Gitea-Tools),
-    not the historical REMOTES Timesheet default. Intentional security-policy
-    migration of legacy omitted-target assertions.
-    """
-
-    def setUp(self):
-        self._ws_env = patch.dict(
-            os.environ,
-            {
-                "GITEA_TEST_WORKSPACE_REMOTE_URL": (
-                    "https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools.git"
-                )
-            },
-            clear=False,
-        )
-        self._ws_env.start()
-
-    def tearDown(self):
-        self._ws_env.stop()
-
 
     @patch("create_pr.get_credentials", return_value=FAKE_CREDS)
     def test_payload_fields(self, _cred):
@@ -134,7 +113,7 @@ class TestAPIPayload(unittest.TestCase):
             url = MockReq.call_args[0][0]
             self.assertEqual(
                 url,
-                "https://gitea.prgs.cc/api/v1/repos/Scaled-Tech-Consulting/Gitea-Tools/pulls",
+                "https://gitea.prgs.cc/api/v1/repos/Scaled-Tech-Consulting/Timesheet/pulls",
             )
 
 

tests/test_issue_714_production_first_bind.py

@@ -1,509 +0,0 @@
-"""#714: production first-bind pins the workspace-verified repository scope.
-
-These tests drive the real entry points (``gitea_whoami``,
-``gitea_get_runtime_context``, ``gitea_resolve_task_capability``,
-``gitea_activate_profile``) in production order. They deliberately do not
-construct a ``_SessionContext`` directly: the defect they cover is that the
-production first-bind path left ``repository``/``org`` unbound, so
-``assess_session_context`` skipped its repository/org drift checks and a
-same-host mutation against another repository was not blocked.
-
-The trusted repository identity comes from the workspace-aligned git remote
-(``Scaled-Tech-Consulting/Gitea-Tools`` for this checkout), never from
-``REMOTES`` (whose ``prgs`` default repo is ``Timesheet``) and never from a
-caller-supplied ``org``/``repo`` argument.
-"""
-
-from __future__ import annotations
-
-import json
-import os
-import sys
-import tempfile
-import threading
-import unittest
-from pathlib import Path
-from unittest.mock import patch
-
-sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
-
-import gitea_config  # noqa: E402
-import gitea_mcp_server as srv  # noqa: E402
-import mcp_server  # noqa: E402
-import session_context_binding as session_ctx  # noqa: E402
-
-WORKSPACE_ORG = "Scaled-Tech-Consulting"
-WORKSPACE_REPO = "Gitea-Tools"
-WORKSPACE_SLUG = f"{WORKSPACE_ORG}/{WORKSPACE_REPO}"
-WORKSPACE_URL = f"https://gitea.prgs.cc/{WORKSPACE_SLUG}.git"
-
-_BASE_AUTHOR_OPS = [
-    "gitea.read",
-    "gitea.issue.create",
-    "gitea.issue.comment",
-    "gitea.pr.create",
-    "gitea.pr.comment",
-    "gitea.branch.push",
-    "gitea.repo.commit",
-]
-
-
-def _config(allowed_repositories=None):
-    profile = {
-        "enabled": True,
-        "context": "prgs",
-        "role": "author",
-        "username": "jcwalker3",
-        "base_url": "https://gitea.prgs.cc",
-        "auth": {"type": "env", "name": "GITEA_TOKEN_PRGS_AUTHOR"},
-        "allowed_operations": list(_BASE_AUTHOR_OPS),
-        "forbidden_operations": [
-            "gitea.pr.approve",
-            "gitea.pr.merge",
-            "gitea.pr.request_changes",
-        ],
-        "execution_profile": "prgs-author",
-    }
-    if allowed_repositories is not None:
-        profile["allowed_repositories"] = list(allowed_repositories)
-    return {
-        "version": 2,
-        # v2-contexts normalizes the config and keeps only rules.* — a
-        # top-level allow_runtime_switching would be dropped.
-        "rules": {"allow_runtime_switching": True},
-        "contexts": {
-            "prgs": {
-                "enabled": True,
-                "gitea": {"enabled": True, "base_url": "https://gitea.prgs.cc"},
-            },
-            "mdcps": {
-                "enabled": True,
-                "gitea": {
-                    "enabled": True,
-                    "base_url": "https://gitea.dadeschools.net",
-                },
-            },
-        },
-        "profiles": {"prgs-author": profile},
-    }
-
-
-class _ProductionOrderBase(unittest.TestCase):
-    """Real entry points, pinned workspace remote, mocked network only."""
-
-    allowed_repositories = [WORKSPACE_SLUG]
-
-    def setUp(self):
-        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(self.allowed_repositories)))
-        self._env = {
-            "GITEA_MCP_CONFIG": self.config_path,
-            "GITEA_MCP_PROFILE": "prgs-author",
-            "GITEA_TOKEN_PRGS_AUTHOR": "prgs-author-token",
-        }
-        gitea_config._active_profile_override = None
-        mcp_server._IDENTITY_CACHE.clear()
-        srv._MUTATION_AUTHORITY = None
-        # Pin the workspace git remote so the trusted source is deterministic
-        # and never depends on the developer's checkout layout.
-        self._remote_url = patch.object(
-            srv, "_local_git_remote_url", side_effect=self._local_remote_url
-        )
-        self._remote_url.start()
-
-    def tearDown(self):
-        self._remote_url.stop()
-        gitea_config._active_profile_override = None
-        mcp_server._IDENTITY_CACHE.clear()
-        srv._MUTATION_AUTHORITY = None
-        self._dir.cleanup()
-
-    def _local_remote_url(self, remote_name):
-        return WORKSPACE_URL if remote_name == "prgs" else None
-
-    def _api(self, method, url, header):
-        return {
-            "login": "jcwalker3",
-            "full_name": "Test",
-            "id": 1,
-            "email": "[email protected]",
-        }
-
-    def _live(self):
-        return patch("gitea_mcp_server.api_request", side_effect=self._api)
-
-
-class TestProductionFirstBindPinsRepository(_ProductionOrderBase):
-    def test_whoami_first_bind_is_complete(self):
-        with patch.dict(os.environ, self._env, clear=False), self._live():
-            srv.gitea_whoami(remote="prgs")
-            ctx = session_ctx.get_session_context()
-            self.assertIsNotNone(ctx)
-            self.assertEqual(ctx["org"], WORKSPACE_ORG)
-            self.assertEqual(ctx["repository"], WORKSPACE_REPO)
-            self.assertEqual(ctx["remote"], "prgs")
-            self.assertEqual(ctx["host"], "gitea.prgs.cc")
-
-    def test_runtime_context_first_bind_is_complete(self):
-        with patch.dict(os.environ, self._env, clear=False), self._live():
-            srv.gitea_get_runtime_context(remote="prgs")
-            ctx = session_ctx.get_session_context()
-            self.assertEqual(ctx["org"], WORKSPACE_ORG)
-            self.assertEqual(ctx["repository"], WORKSPACE_REPO)
-
-    def test_capability_preflight_first_bind_is_complete(self):
-        with patch.dict(os.environ, self._env, clear=False), self._live():
-            srv.gitea_resolve_task_capability(task="comment_issue", remote="prgs")
-            ctx = session_ctx.get_session_context()
-            self.assertEqual(ctx["org"], WORKSPACE_ORG)
-            self.assertEqual(ctx["repository"], WORKSPACE_REPO)
-
-    def test_activate_profile_binds_complete_context(self):
-        with patch.dict(os.environ, self._env, clear=False), self._live():
-            srv.gitea_activate_profile(profile_name="prgs-author", remote="prgs")
-            ctx = session_ctx.get_session_context()
-            self.assertEqual(ctx["org"], WORKSPACE_ORG)
-            self.assertEqual(ctx["repository"], WORKSPACE_REPO)
-
-
-class TestProductionOrderRepositoryDriftBlocks(_ProductionOrderBase):
-    def test_whoami_then_same_host_other_repository_blocks(self):
-        with patch.dict(os.environ, self._env, clear=False), self._live():
-            srv.gitea_whoami(remote="prgs")
-            blocked = srv._session_context_mutation_block(
-                remote="prgs", org=WORKSPACE_ORG, repo="Other-Tools"
-            )
-            self.assertIsNotNone(blocked)
-            self.assertTrue(
-                any("Other-Tools" in r for r in blocked["reasons"]), blocked["reasons"]
-            )
-
-    def test_whoami_then_timesheet_blocks(self):
-        """REMOTES['prgs'].repo is Timesheet — a default target, not a scope."""
-        with patch.dict(os.environ, self._env, clear=False), self._live():
-            srv.gitea_whoami(remote="prgs")
-            blocked = srv._session_context_mutation_block(
-                remote="prgs",
-                org=WORKSPACE_ORG,
-                repo="Timesheet",
-                org_explicit=True,
-                repo_explicit=True,
-            )
-            self.assertIsNotNone(blocked)
-            self.assertTrue(
-                any("Timesheet" in r for r in blocked["reasons"]), blocked["reasons"]
-            )
-
-    def test_whoami_then_same_host_other_org_blocks(self):
-        with patch.dict(os.environ, self._env, clear=False), self._live():
-            srv.gitea_whoami(remote="prgs")
-            blocked = srv._session_context_mutation_block(
-                remote="prgs", org="Other-Org", repo=WORKSPACE_REPO
-            )
-            self.assertIsNotNone(blocked)
-            self.assertTrue(
-                any("Other-Org" in r for r in blocked["reasons"]), blocked["reasons"]
-            )
-
-    def test_runtime_context_then_other_repository_blocks(self):
-        with patch.dict(os.environ, self._env, clear=False), self._live():
-            srv.gitea_get_runtime_context(remote="prgs")
-            blocked = srv._session_context_mutation_block(
-                remote="prgs", org=WORKSPACE_ORG, repo="Other-Tools"
-            )
-            self.assertIsNotNone(blocked)
-
-    def test_capability_preflight_then_other_repository_blocks(self):
-        with patch.dict(os.environ, self._env, clear=False), self._live():
-            srv.gitea_resolve_task_capability(task="comment_issue", remote="prgs")
-            blocked = srv._session_context_mutation_block(
-                remote="prgs", org=WORKSPACE_ORG, repo="Other-Tools"
-            )
-            self.assertIsNotNone(blocked)
-
-    def test_activate_profile_then_other_repository_blocks(self):
-        with patch.dict(os.environ, self._env, clear=False), self._live():
-            srv.gitea_activate_profile(profile_name="prgs-author", remote="prgs")
-            blocked = srv._session_context_mutation_block(
-                remote="prgs", org="Other-Org", repo="Other-Tools"
-            )
-            self.assertIsNotNone(blocked)
-
-    def test_cross_host_still_blocks(self):
-        with patch.dict(os.environ, self._env, clear=False), self._live():
-            srv.gitea_whoami(remote="prgs")
-            blocked = srv._session_context_mutation_block(remote="dadeschools")
-            self.assertIsNotNone(blocked)
-            self.assertTrue(
-                any("cross-host" in r for r in blocked["reasons"]), blocked["reasons"]
-            )
-
-    def test_authorized_repository_mutation_remains_allowed(self):
-        """Normal PR #715 author comment/push path must stay functional."""
-        with patch.dict(os.environ, self._env, clear=False), self._live():
-            srv.gitea_whoami(remote="prgs")
-            self.assertIsNone(
-                srv._session_context_mutation_block(
-                    remote="prgs", org=WORKSPACE_ORG, repo=WORKSPACE_REPO
-                )
-            )
-            # A bare call with no override resolves to the same bound scope.
-            self.assertIsNone(srv._session_context_mutation_block(remote="prgs"))
-
-
-class TestMutationRequestCannotEstablishBinding(_ProductionOrderBase):
-    def test_caller_values_cannot_establish_binding_on_first_mutation(self):
-        """A mutation-first session must not be pinned by request values."""
-        with patch.dict(os.environ, self._env, clear=False), self._live():
-            self.assertIsNone(session_ctx.get_session_context())
-            srv._session_context_mutation_block(
-                remote="prgs", org="Attacker-Org", repo="Attacker-Repo"
-            )
-            ctx = session_ctx.get_session_context()
-            self.assertIsNotNone(ctx)
-            # Bound to the verified workspace, never to the request values.
-            self.assertEqual(ctx["org"], WORKSPACE_ORG)
-            self.assertEqual(ctx["repository"], WORKSPACE_REPO)
-
-    def test_mutation_first_with_attacker_values_is_blocked(self):
-        with patch.dict(os.environ, self._env, clear=False), self._live():
-            blocked = srv._session_context_mutation_block(
-                remote="prgs", org="Attacker-Org", repo="Attacker-Repo"
-            )
-            self.assertIsNotNone(blocked)
-
-    def test_later_call_cannot_overwrite_bound_repository(self):
-        with patch.dict(os.environ, self._env, clear=False), self._live():
-            srv.gitea_whoami(remote="prgs")
-            before = session_ctx.get_session_context()
-            for _ in range(3):
-                srv._session_context_mutation_block(
-                    remote="prgs", org="Other-Org", repo="Other-Tools"
-                )
-            self.assertEqual(session_ctx.get_session_context(), before)
-
-
-class TestUnverifiedWorkspaceFailsClosed(_ProductionOrderBase):
-    def _local_remote_url(self, remote_name):
-        return None  # no verifiable workspace repository
-
-    def test_mutation_blocks_when_workspace_repository_unverified(self):
-        with patch.dict(os.environ, self._env, clear=False), self._live():
-            srv.gitea_whoami(remote="prgs")
-            ctx = session_ctx.get_session_context()
-            self.assertIsNone(ctx["repository"])
-            blocked = srv._session_context_mutation_block(
-                remote="prgs", org=WORKSPACE_ORG, repo=WORKSPACE_REPO
-            )
-            self.assertIsNotNone(blocked)
-            self.assertTrue(
-                any(
-                    "no verified workspace repository" in r or "unverified" in r
-                    for r in blocked["reasons"]
-                ),
-                blocked["reasons"],
-            )
-
-    def test_activate_profile_rejects_unverifiable_workspace(self):
-        with patch.dict(os.environ, self._env, clear=False), self._live():
-            res = srv.gitea_activate_profile(profile_name="prgs-author", remote="prgs")
-            self.assertFalse(res.get("success", True))
-            self.assertEqual(res.get("blocker_kind"), "repository_scope")
-
-
-class TestUnauthorizedWorkspaceFailsClosed(_ProductionOrderBase):
-    allowed_repositories = ["Scaled-Tech-Consulting/Some-Other-Project"]
-
-    def test_workspace_absent_from_allowlist_blocks_mutation(self):
-        with patch.dict(os.environ, self._env, clear=False), self._live():
-            blocked = srv._session_context_mutation_block(remote="prgs")
-            self.assertIsNotNone(blocked)
-            self.assertTrue(
-                any("not authorized" in r for r in blocked["reasons"]),
-                blocked["reasons"],
-            )
-
-    def test_workspace_absent_from_allowlist_blocks_activation(self):
-        with patch.dict(os.environ, self._env, clear=False), self._live():
-            res = srv.gitea_activate_profile(profile_name="prgs-author", remote="prgs")
-            self.assertFalse(res.get("success", True))
-            self.assertEqual(res.get("blocker_kind"), "repository_scope")
-
-
-class TestTimesheetNotAuthorizedInThisPhase(_ProductionOrderBase):
-    """Cross-project use is paused: only Gitea-Tools is authorized."""
-
-    def test_timesheet_workspace_is_rejected(self):
-        def timesheet_remote(remote_name):
-            return "https://gitea.prgs.cc/Scaled-Tech-Consulting/Timesheet.git"
-
-        with patch.dict(os.environ, self._env, clear=False), self._live():
-            with patch.object(
-                srv, "_local_git_remote_url", side_effect=timesheet_remote
-            ):
-                blocked = srv._session_context_mutation_block(remote="prgs")
-                self.assertIsNotNone(blocked)
-                self.assertTrue(
-                    any("not authorized" in r for r in blocked["reasons"]),
-                    blocked["reasons"],
-                )
-
-
-class TestConcurrentFirstBindSelectsOneRepository(_ProductionOrderBase):
-    def test_concurrent_initialization_cannot_change_selected_repository(self):
-        with patch.dict(os.environ, self._env, clear=False), self._live():
-            self.assertIsNone(session_ctx.get_session_context())
-            thread_count = 8
-            barrier = threading.Barrier(thread_count)
-            results = []
-            lock = threading.Lock()
-
-            def racer(index: int) -> None:
-                barrier.wait()
-                srv._session_context_mutation_block(
-                    remote="prgs", org=f"Racer-Org-{index}", repo=f"Racer-Repo-{index}"
-                )
-                with lock:
-                    results.append(session_ctx.get_session_context())
-
-            threads = [
-                threading.Thread(target=racer, args=(i,)) for i in range(thread_count)
-            ]
-            for thread in threads:
-                thread.start()
-            for thread in threads:
-                thread.join(timeout=10)
-
-            self.assertFalse(any(t.is_alive() for t in threads))
-            winner = session_ctx.get_session_context()
-            self.assertEqual(winner["org"], WORKSPACE_ORG)
-            self.assertEqual(winner["repository"], WORKSPACE_REPO)
-            self.assertEqual(results, [winner] * thread_count)
-
-
-class TestRepositoryScopeUnits(unittest.TestCase):
-    def test_absent_scope_is_not_enforced_for_read_diagnostics(self):
-        scope = session_ctx.assess_repository_scope(
-            workspace_slug=WORKSPACE_SLUG, allowed=[], profile_name="legacy"
-        )
-        self.assertTrue(scope["proven"])
-        self.assertFalse(scope["scope_enforced"])
-
-    def test_require_scope_blocks_empty_or_missing_allowlist(self):
-        scope = session_ctx.assess_repository_scope(
-            workspace_slug=WORKSPACE_SLUG,
-            allowed=[],
-            profile_name="legacy",
-            require_scope=True,
-        )
-        self.assertTrue(scope["block"])
-        self.assertTrue(any("allowed_repositories" in r for r in scope["reasons"]))
-
-    def test_declared_scope_authorizes_only_listed_repository(self):
-        allowed = session_ctx.declared_allowed_repositories(
-            {"allowed_repositories": [WORKSPACE_SLUG]}
-        )
-        self.assertEqual(allowed, [WORKSPACE_SLUG])
-        self.assertTrue(
-            session_ctx.assess_repository_scope(
-                workspace_slug=WORKSPACE_SLUG, allowed=allowed
-            )["proven"]
-        )
-        self.assertTrue(
-            session_ctx.assess_repository_scope(
-                workspace_slug="Scaled-Tech-Consulting/Timesheet", allowed=allowed
-            )["block"]
-        )
-
-    def test_missing_workspace_slug_blocks_when_scope_declared(self):
-        scope = session_ctx.assess_repository_scope(
-            workspace_slug=None, allowed=[WORKSPACE_SLUG]
-        )
-        self.assertTrue(scope["block"])
-
-    def test_override_must_match_binding(self):
-        self.assertTrue(
-            session_ctx.assess_repository_override(
-                requested_org=WORKSPACE_ORG,
-                requested_repo="Other",
-                bound_org=WORKSPACE_ORG,
-                bound_repo=WORKSPACE_REPO,
-            )["block"]
-        )
-        self.assertTrue(
-            session_ctx.assess_repository_override(
-                requested_org="Other-Org",
-                requested_repo=WORKSPACE_REPO,
-                bound_org=WORKSPACE_ORG,
-                bound_repo=WORKSPACE_REPO,
-            )["block"]
-        )
-        self.assertTrue(
-            session_ctx.assess_repository_override(
-                requested_org=WORKSPACE_ORG,
-                requested_repo=WORKSPACE_REPO,
-                bound_org=WORKSPACE_ORG,
-                bound_repo=WORKSPACE_REPO,
-            )["proven"]
-        )
-
-    def test_malformed_allowlist_entries_are_ignored_in_non_strict_mode(self):
-        self.assertEqual(
-            session_ctx.declared_allowed_repositories(
-                {"allowed_repositories": ["not-a-slug", 17, None, WORKSPACE_SLUG]}
-            ),
-            [WORKSPACE_SLUG],
-        )
-
-    def test_strict_mode_rejects_malformed_allowlist_entries(self):
-        with self.assertRaises(ValueError):
-            session_ctx.declared_allowed_repositories(
-                {"allowed_repositories": ["not-a-slug", WORKSPACE_SLUG]},
-                strict=True,
-            )
-
-
-class TestRemotesDefaultNotCallerOverride(_ProductionOrderBase):
-    def test_resolved_timesheet_default_does_not_block_when_bound_to_workspace(self):
-        """Tools historically pass REMOTES-filled Timesheet after _resolve; that
-        must not be treated as an explicit override against Gitea-Tools."""
-        with patch.dict(os.environ, self._env, clear=False), self._live():
-            srv.gitea_whoami(remote="prgs")
-            # Simulate create_issue after resolve: org/repo are REMOTES defaults
-            blocked = srv._session_context_mutation_block(
-                remote="prgs",
-                org="Scaled-Tech-Consulting",
-                repo="Timesheet",
-            )
-            self.assertIsNone(blocked, getattr(blocked, "get", lambda *_: blocked)("reasons") if blocked else None)
-
-    def test_git_unavailable_blocks_mutation(self):
-        def no_remote(_name):
-            return None
-        with patch.dict(os.environ, self._env, clear=False), self._live():
-            with patch.object(srv, "_local_git_remote_url", side_effect=no_remote):
-                blocked = srv._session_context_mutation_block(remote="prgs")
-                self.assertIsNotNone(blocked)
-                self.assertTrue(
-                    any(
-                        "workspace" in r.lower() or "allowed_repositories" in r or "unverified" in r
-                        for r in blocked["reasons"]
-                    ),
-                    blocked["reasons"],
-                )
-
-    def test_explicit_mismatch_still_blocks(self):
-        with patch.dict(os.environ, self._env, clear=False), self._live():
-            srv.gitea_whoami(remote="prgs")
-            blocked = srv._session_context_mutation_block(
-                remote="prgs",
-                org=WORKSPACE_ORG,
-                repo="Other-Tools",
-            )
-            self.assertIsNotNone(blocked)
-
-
-if __name__ == "__main__":
-    unittest.main()

tests/test_issue_714_session_context_immutability.py

@@ -1,608 +0,0 @@
-"""#714: fail closed on cross-host MCP profile drift and capability substitution."""
-
-from __future__ import annotations
-
-import json
-import os
-import sys
-import tempfile
-import threading
-import unittest
-from pathlib import Path
-from unittest.mock import patch
-
-sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
-
-import gitea_config  # noqa: E402
-import mcp_server  # noqa: E402
-import session_context_binding as session_ctx  # noqa: E402
-
-
-CONFIG_714 = {
-    "version": 2,
-    "allow_runtime_switching": True,
-    "contexts": {
-        "prgs": {
-            "enabled": True,
-            "gitea": {"enabled": True, "base_url": "https://gitea.prgs.cc"},
-        },
-        "mdcps": {
-            "enabled": True,
-            "gitea": {"enabled": True, "base_url": "https://gitea.dadeschools.net"},
-        },
-    },
-    "profiles": {
-        "prgs-author": {
-            "enabled": True,
-            "context": "prgs",
-            "role": "author",
-            "username": "jcwalker3",
-            "base_url": "https://gitea.prgs.cc",
-            "auth": {"type": "env", "name": "GITEA_TOKEN_PRGS_AUTHOR"},
-            "allowed_operations": [
-                "gitea.read",
-                "gitea.issue.create",
-                "gitea.issue.comment",
-                "gitea.issue.close",
-                "gitea.pr.create",
-                "gitea.pr.comment",
-                "gitea.branch.create",
-                "gitea.branch.push",
-                "gitea.repo.commit",
-            ],
-            "forbidden_operations": [
-                "gitea.pr.approve",
-                "gitea.pr.merge",
-                "gitea.pr.request_changes",
-            ],
-            "allowed_repositories": ["Scaled-Tech-Consulting/Gitea-Tools"],
-            "execution_profile": "prgs-author",
-        },
-        "mdcps-reviewer": {
-            "enabled": True,
-            "context": "mdcps",
-            "role": "reviewer",
-            "username": "913443",
-            "base_url": "https://gitea.dadeschools.net",
-            "auth": {"type": "env", "name": "GITEA_TOKEN_MDCPS_REVIEWER"},
-            "allowed_operations": [
-                "gitea.read",
-                "gitea.pr.review",
-                "gitea.pr.comment",
-                "gitea.pr.approve",
-                "gitea.pr.request_changes",
-            ],
-            "forbidden_operations": [
-                "gitea.branch.create",
-                "gitea.branch.push",
-                "gitea.pr.create",
-                "gitea.pr.merge",
-                "gitea.repo.commit",
-            ],
-            "allowed_repositories": ["913443/eAgenda"],
-            "execution_profile": "mdcps-reviewer",
-        },
-        "mdcps-author": {
-            "enabled": True,
-            "context": "mdcps",
-            "role": "author",
-            "username": "913443",
-            "base_url": "https://gitea.dadeschools.net",
-            "auth": {"type": "env", "name": "GITEA_TOKEN_MDCPS_AUTHOR"},
-            "allowed_operations": [
-                "gitea.read",
-                "gitea.issue.create",
-                "gitea.issue.comment",
-                "gitea.pr.create",
-                "gitea.pr.comment",
-                "gitea.branch.create",
-                "gitea.branch.push",
-                "gitea.repo.commit",
-            ],
-            "forbidden_operations": [
-                "gitea.pr.approve",
-                "gitea.pr.merge",
-                "gitea.pr.request_changes",
-            ],
-            "allowed_repositories": ["913443/eAgenda"],
-            "execution_profile": "mdcps-author",
-        },
-    },
-}
-
-
-class TestIssue714SessionContextImmutability(unittest.TestCase):
-    def setUp(self):
-        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_714))
-        self._remotes = patch.dict(
-            mcp_server.REMOTES,
-            {
-                "dadeschools": {
-                    "host": "gitea.dadeschools.net",
-                    "org": "913443",
-                    "repo": "eAgenda",
-                },
-                "prgs": {
-                    "host": "gitea.prgs.cc",
-                    "org": "Scaled-Tech-Consulting",
-                    "repo": "Gitea-Tools",
-                },
-            },
-            clear=False,
-        )
-        self._remotes.start()
-        def _url(name):
-            if name == "dadeschools":
-                return "https://gitea.dadeschools.net/913443/eAgenda.git"
-            if name == "prgs":
-                return "https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools.git"
-            return None
-        self._url_patch = patch.object(mcp_server, "_local_git_remote_url", side_effect=_url)
-        self._url_patch.start()
-        mcp_server._IDENTITY_CACHE.clear()
-        gitea_config._active_profile_override = None
-        mcp_server._MUTATION_AUTHORITY = None
-        self._env = {
-            "GITEA_MCP_CONFIG": self.config_path,
-            "GITEA_MCP_PROFILE": "mdcps-reviewer",
-            "GITEA_TOKEN_MDCPS_REVIEWER": "mdcps-reviewer-token",
-            "GITEA_TOKEN_MDCPS_AUTHOR": "mdcps-author-token",
-            "GITEA_TOKEN_PRGS_AUTHOR": "prgs-author-token",
-        }
-
-    def tearDown(self):
-        self._url_patch.stop()
-        self._remotes.stop()
-        mcp_server._IDENTITY_CACHE.clear()
-        gitea_config._active_profile_override = None
-        mcp_server._MUTATION_AUTHORITY = None
-        self._dir.cleanup()
-
-    def _api_side_effect(self, method, url, header):
-        # Token identity mapping for whoami endpoint
-        auth = str(header)
-        if "mdcps-reviewer-token" in auth or "mdcps-author-token" in auth:
-            login = "913443"
-        else:
-            login = "jcwalker3"
-        return {"login": login, "full_name": "Test", "id": 1, "email": "[email protected]"}
-
-    def test_pin_mdcps_reviewer_never_drifts_to_prgs_author(self):
-        """Reproduce the incident: pin mdcps-reviewer then resolve comment_issue."""
-        with patch.dict(os.environ, self._env, clear=False):
-            with patch("mcp_server.api_request", side_effect=self._api_side_effect):
-                act = mcp_server.gitea_activate_profile(
-                    profile_name="mdcps-reviewer", remote="dadeschools"
-                )
-                self.assertTrue(act["success"])
-                self.assertEqual(act["after_profile"], "mdcps-reviewer")
-
-                who1 = mcp_server.gitea_whoami(remote="dadeschools")
-                self.assertEqual(who1["profile"]["profile_name"], "mdcps-reviewer")
-                self.assertEqual(who1["username"], "913443")
-
-                # Capability that mdcps-reviewer lacks — must NOT switch to prgs-author
-                res = mcp_server.gitea_resolve_task_capability(
-                    task="comment_issue", remote="dadeschools"
-                )
-                self.assertEqual(res["active_profile"], "mdcps-reviewer")
-                self.assertFalse(res["allowed_in_current_session"])
-                self.assertFalse(res.get("auto_profile_substitution", True))
-                self.assertNotIn("prgs-author", res["matching_configured_profile"])
-                self.assertIn("mdcps-author", res["matching_configured_profile"])
-
-                who2 = mcp_server.gitea_whoami(remote="dadeschools")
-                rt = mcp_server.gitea_get_runtime_context(remote="dadeschools")
-                self.assertEqual(who2["profile"]["profile_name"], "mdcps-reviewer")
-                self.assertEqual(rt["active_profile"], "mdcps-reviewer")
-                # Still pinned — never prgs-author
-                self.assertEqual(
-                    gitea_config.selected_profile_name(), "mdcps-reviewer"
-                )
-
-    def test_repeated_calls_remain_on_mdcps_reviewer(self):
-        with patch.dict(os.environ, self._env, clear=False):
-            with patch("mcp_server.api_request", side_effect=self._api_side_effect):
-                mcp_server.gitea_activate_profile(
-                    profile_name="mdcps-reviewer", remote="dadeschools"
-                )
-                for _ in range(5):
-                    res = mcp_server.gitea_resolve_task_capability(
-                        task="review_pr", remote="dadeschools"
-                    )
-                    self.assertEqual(res["active_profile"], "mdcps-reviewer")
-                    who = mcp_server.gitea_whoami(remote="dadeschools")
-                    self.assertEqual(who["profile"]["profile_name"], "mdcps-reviewer")
-                    rt = mcp_server.gitea_get_runtime_context(remote="dadeschools")
-                    self.assertEqual(rt["active_profile"], "mdcps-reviewer")
-
-    def test_dadeschools_request_cannot_resolve_through_prgs_author(self):
-        with patch.dict(os.environ, self._env, clear=False):
-            with patch("mcp_server.api_request", side_effect=self._api_side_effect):
-                mcp_server.gitea_activate_profile(
-                    profile_name="mdcps-reviewer", remote="dadeschools"
-                )
-                res = mcp_server.gitea_resolve_task_capability(
-                    task="comment_issue", remote="dadeschools"
-                )
-                self.assertNotEqual(res["active_profile"], "prgs-author")
-                self.assertNotIn("prgs-author", res["matching_configured_profile"])
-                # Gate must not silently switch either
-                blocked = mcp_server._profile_permission_block(
-                    "gitea.issue.comment", remote="dadeschools"
-                )
-                self.assertIsNotNone(blocked)
-                self.assertFalse(blocked.get("success", True))
-                self.assertEqual(
-                    gitea_config.selected_profile_name(), "mdcps-reviewer"
-                )
-
-    def test_prgs_request_cannot_resolve_through_mdcps_profile(self):
-        """Reverse of the incident: a pinned MDCPS profile must never serve a
-        prgs request, nor be silently swapped for the prgs-side profile."""
-        with patch.dict(os.environ, self._env, clear=False):
-            with patch("mcp_server.api_request", side_effect=self._api_side_effect):
-                mcp_server.gitea_activate_profile(
-                    profile_name="mdcps-author", remote="dadeschools"
-                )
-                res = mcp_server.gitea_resolve_task_capability(
-                    task="create_issue", remote="prgs"
-                )
-                self.assertNotEqual(res["active_profile"], "prgs-author")
-                self.assertEqual(res["active_profile"], "mdcps-author")
-                self.assertFalse(res["allowed_in_current_session"])
-                self.assertTrue(res["stop_required"])
-                blocked = mcp_server._session_context_mutation_block(remote="prgs")
-                self.assertIsNotNone(blocked)
-                self.assertTrue(any("cross-host" in r for r in blocked["reasons"]))
-                self.assertEqual(gitea_config.selected_profile_name(), "mdcps-author")
-
-    def test_unsupported_capability_fails_closed_structured(self):
-        with patch.dict(os.environ, self._env, clear=False):
-            with patch("mcp_server.api_request", side_effect=self._api_side_effect):
-                mcp_server.gitea_activate_profile(
-                    profile_name="mdcps-reviewer", remote="dadeschools"
-                )
-                res = mcp_server.gitea_resolve_task_capability(
-                    task="comment_issue", remote="dadeschools"
-                )
-                self.assertFalse(res["allowed_in_current_session"])
-                self.assertTrue(res["stop_required"])
-                self.assertIn("reason", res)
-                self.assertIn("exact_safe_next_action", res)
-                self.assertIn("activate_profile", res["exact_safe_next_action"])
-                # Unknown task still fail closed
-                with self.assertRaises(ValueError):
-                    mcp_server.gitea_resolve_task_capability(
-                        task="reopen_issue", remote="dadeschools"
-                    )
-
-    def test_identity_mismatch_blocks_mutation(self):
-        """Profile expects 913443 but authenticated as jcwalker3."""
-
-        def wrong_identity(method, url, header):
-            return {
-                "login": "jcwalker3",
-                "full_name": "Wrong",
-                "id": 9,
-                "email": "[email protected]",
-            }
-
-        with patch.dict(os.environ, self._env, clear=False):
-            with patch("mcp_server.api_request", side_effect=wrong_identity):
-                mcp_server.gitea_activate_profile(
-                    profile_name="mdcps-reviewer", remote="dadeschools"
-                )
-                blocked = mcp_server._session_context_mutation_block(
-                    remote="dadeschools"
-                )
-                self.assertIsNotNone(blocked)
-                self.assertTrue(
-                    any("identity mismatch" in r for r in blocked["reasons"])
-                )
-
-    def test_repository_host_mismatch_blocks_mutation(self):
-        """mdcps-reviewer cannot serve prgs remote."""
-        with patch.dict(os.environ, self._env, clear=False):
-            with patch("mcp_server.api_request", side_effect=self._api_side_effect):
-                mcp_server.gitea_activate_profile(
-                    profile_name="mdcps-reviewer", remote="dadeschools"
-                )
-                blocked = mcp_server._session_context_mutation_block(remote="prgs")
-                self.assertIsNotNone(blocked)
-                self.assertTrue(
-                    any("cross-host" in r for r in blocked["reasons"])
-                )
-
-    def test_drift_cannot_reach_write(self):
-        """Simulated mid-session profile override cannot pass permission gate."""
-        with patch.dict(os.environ, self._env, clear=False):
-            with patch("mcp_server.api_request", side_effect=self._api_side_effect):
-                mcp_server.gitea_activate_profile(
-                    profile_name="mdcps-reviewer", remote="dadeschools"
-                )
-                # Bind session as mdcps-reviewer
-                self.assertEqual(
-                    session_ctx.get_session_context()["profile_name"],
-                    "mdcps-reviewer",
-                )
-                # Hostile override without activate_profile
-                gitea_config._active_profile_override = "prgs-author"
-                blocked = mcp_server._session_context_mutation_block(
-                    remote="dadeschools"
-                )
-                self.assertIsNotNone(blocked)
-                self.assertTrue(any("drift" in r or "cross-host" in r for r in blocked["reasons"]))
-
-    def test_static_prgs_author_still_works(self):
-        env = {
-            "GITEA_MCP_CONFIG": self.config_path,
-            "GITEA_MCP_PROFILE": "prgs-author",
-            "GITEA_TOKEN_PRGS_AUTHOR": "prgs-author-token",
-        }
-        with patch.dict(os.environ, env, clear=False):
-            with patch("mcp_server.api_request", side_effect=self._api_side_effect):
-                gitea_config._active_profile_override = None
-                res = mcp_server.gitea_resolve_task_capability(
-                    task="create_issue", remote="prgs"
-                )
-                self.assertEqual(res["active_profile"], "prgs-author")
-                self.assertTrue(res["allowed_in_current_session"])
-                self.assertEqual(res["active_identity"], "jcwalker3")
-
-    def test_static_mdcps_author_still_works(self):
-        env = {
-            "GITEA_MCP_CONFIG": self.config_path,
-            "GITEA_MCP_PROFILE": "mdcps-author",
-            "GITEA_TOKEN_MDCPS_AUTHOR": "mdcps-author-token",
-        }
-        with patch.dict(os.environ, env, clear=False):
-            with patch("mcp_server.api_request", side_effect=self._api_side_effect):
-                gitea_config._active_profile_override = None
-                res = mcp_server.gitea_resolve_task_capability(
-                    task="comment_issue", remote="dadeschools"
-                )
-                self.assertEqual(res["active_profile"], "mdcps-author")
-                self.assertTrue(res["allowed_in_current_session"])
-                self.assertNotIn("prgs-author", res["matching_configured_profile"])
-
-
-class TestSessionContextBindingUnit(unittest.TestCase):
-    def test_profile_matches_remote_by_base_url(self):
-        remotes = {
-            "dadeschools": {"host": "gitea.dadeschools.net"},
-            "prgs": {"host": "gitea.prgs.cc"},
-        }
-        mdcps = {"base_url": "https://gitea.dadeschools.net", "context": "mdcps"}
-        prgs = {"base_url": "https://gitea.prgs.cc", "context": "prgs"}
-        self.assertTrue(
-            session_ctx.profile_matches_remote(mdcps, "dadeschools", remotes)
-        )
-        self.assertFalse(session_ctx.profile_matches_remote(mdcps, "prgs", remotes))
-        self.assertTrue(session_ctx.profile_matches_remote(prgs, "prgs", remotes))
-
-    def test_bind_and_detect_drift(self):
-        session_ctx.bind_session_context(
-            profile_name="mdcps-reviewer",
-            remote="dadeschools",
-            host="gitea.dadeschools.net",
-            identity="913443",
-            source="test",
-        )
-        ok = session_ctx.assess_session_context(
-            profile_name="mdcps-reviewer",
-            remote="dadeschools",
-            host="gitea.dadeschools.net",
-            identity="913443",
-        )
-        self.assertTrue(ok["proven"])
-        bad = session_ctx.assess_session_context(
-            profile_name="prgs-author",
-            remote="dadeschools",
-            host="gitea.dadeschools.net",
-            identity="jcwalker3",
-        )
-        self.assertTrue(bad["block"])
-        self.assertTrue(any("drift" in r for r in bad["reasons"]))
-
-    def test_bound_context_survives_multiple_calls_and_exceptions(self):
-        original = session_ctx.bind_session_context(
-            profile_name="mdcps-reviewer",
-            remote="dadeschools",
-            host="gitea.dadeschools.net",
-            identity="913443",
-            source="test",
-        )
-
-        try:
-            for _ in range(3):
-                observed = session_ctx.seed_session_context_if_unbound(
-                    profile_name="prgs-author",
-                    remote="prgs",
-                    host="gitea.prgs.cc",
-                    identity="jcwalker3",
-                    source="interleaved-test-call",
-                )
-                self.assertEqual(observed, original)
-            raise RuntimeError("simulated caller failure")
-        except RuntimeError:
-            pass
-
-        self.assertEqual(session_ctx.get_session_context(), original)
-        drift = session_ctx.assess_session_context(
-            profile_name="prgs-author",
-            remote="prgs",
-            host="gitea.prgs.cc",
-            identity="jcwalker3",
-            require_bound=True,
-        )
-        self.assertTrue(drift["block"])
-
-    def test_parallel_calls_cannot_overwrite_established_binding(self):
-        original = session_ctx.bind_session_context(
-            profile_name="mdcps-reviewer",
-            remote="dadeschools",
-            host="gitea.dadeschools.net",
-            identity="913443",
-            source="test",
-        )
-        barrier = threading.Barrier(9)
-        results = []
-        results_lock = threading.Lock()
-
-        def competing_seed(index: int) -> None:
-            barrier.wait()
-            result = session_ctx.seed_session_context_if_unbound(
-                profile_name=f"prgs-author-{index}",
-                remote="prgs",
-                host="gitea.prgs.cc",
-                identity=f"other-{index}",
-                source="parallel-test-call",
-            )
-            with results_lock:
-                results.append(result)
-
-        threads = [threading.Thread(target=competing_seed, args=(i,)) for i in range(8)]
-        for thread in threads:
-            thread.start()
-        barrier.wait()
-        for thread in threads:
-            thread.join(timeout=5)
-
-        self.assertFalse(any(thread.is_alive() for thread in threads))
-        self.assertEqual(results, [original] * 8)
-        self.assertEqual(session_ctx.get_session_context(), original)
-
-    def test_concurrent_initialization_has_single_winning_binding(self):
-        """Racing first-binds from an unbound context: exactly one winner, and
-        every racer observes that same complete binding (never a partial one)."""
-        self.assertIsNone(session_ctx.get_session_context())
-        thread_count = 8
-        barrier = threading.Barrier(thread_count)
-        results = []
-        results_lock = threading.Lock()
-
-        def racing_seed(index: int) -> None:
-            barrier.wait()
-            result = session_ctx.seed_session_context_if_unbound(
-                profile_name=f"prgs-author-{index}",
-                remote="prgs",
-                host="gitea.prgs.cc",
-                identity=f"user-{index}",
-                source="concurrent-init-test",
-            )
-            with results_lock:
-                results.append(result)
-
-        threads = [
-            threading.Thread(target=racing_seed, args=(i,))
-            for i in range(thread_count)
-        ]
-        for thread in threads:
-            thread.start()
-        for thread in threads:
-            thread.join(timeout=5)
-
-        self.assertFalse(any(thread.is_alive() for thread in threads))
-        winner = session_ctx.get_session_context()
-        self.assertIsNotNone(winner)
-        self.assertEqual(len(results), thread_count)
-        self.assertEqual(results, [winner] * thread_count)
-        # A losing racer's identity must never be spliced into the winner.
-        self.assertEqual(
-            winner["identity"], winner["profile_name"].replace("prgs-author-", "user-")
-        )
-
-    def test_repository_mismatch_blocks_before_mutation(self):
-        """Same host and profile, different repository, must still fail closed."""
-        session_ctx.bind_session_context(
-            profile_name="prgs-author",
-            remote="prgs",
-            host="gitea.prgs.cc",
-            identity="jcwalker3",
-            repository="Gitea-Tools",
-            org="Scaled-Tech-Consulting",
-            source="test",
-        )
-        same = session_ctx.assess_session_context(
-            profile_name="prgs-author",
-            remote="prgs",
-            host="gitea.prgs.cc",
-            identity="jcwalker3",
-            repository="Gitea-Tools",
-            org="Scaled-Tech-Consulting",
-            require_bound=True,
-        )
-        self.assertTrue(same["proven"])
-        other_repo = session_ctx.assess_session_context(
-            profile_name="prgs-author",
-            remote="prgs",
-            host="gitea.prgs.cc",
-            identity="jcwalker3",
-            repository="eAgenda",
-            org="Scaled-Tech-Consulting",
-            require_bound=True,
-        )
-        self.assertTrue(other_repo["block"])
-        self.assertTrue(
-            any("repository drift" in r for r in other_repo["reasons"])
-        )
-        other_org = session_ctx.assess_session_context(
-            profile_name="prgs-author",
-            remote="prgs",
-            host="gitea.prgs.cc",
-            identity="jcwalker3",
-            repository="Gitea-Tools",
-            org="913443",
-            require_bound=True,
-        )
-        self.assertTrue(other_org["block"])
-        self.assertTrue(any("org drift" in r for r in other_org["reasons"]))
-
-    def test_returned_snapshot_cannot_mutate_binding(self):
-        session_ctx.bind_session_context(
-            profile_name="mdcps-reviewer",
-            remote="dadeschools",
-            host="gitea.dadeschools.net",
-            identity="913443",
-            source="test",
-        )
-        snapshot = session_ctx.get_session_context()
-        snapshot["profile_name"] = "prgs-author"
-        self.assertEqual(
-            session_ctx.get_session_context()["profile_name"], "mdcps-reviewer"
-        )
-
-
-class TestSessionContextTestBoundaryIsolation(unittest.TestCase):
-    def test_mdcps_binding_starts_clean_and_does_not_escape_test(self):
-        self.assertIsNone(session_ctx.get_session_context())
-        session_ctx.bind_session_context(
-            profile_name="mdcps-reviewer",
-            remote="dadeschools",
-            host="gitea.dadeschools.net",
-            identity="913443",
-            source="test-boundary",
-        )
-
-    def test_prgs_binding_starts_clean_and_does_not_escape_test(self):
-        self.assertIsNone(session_ctx.get_session_context())
-        session_ctx.bind_session_context(
-            profile_name="prgs-author",
-            remote="prgs",
-            host="gitea.prgs.cc",
-            identity="jcwalker3",
-            source="test-boundary",
-        )
-
-    def test_reset_helper_is_rejected_outside_pytest_boundary(self):
-        with patch.dict(os.environ, {}, clear=True):
-            with self.assertRaises(RuntimeError):
-                session_ctx._reset_session_context_for_testing()
-
-
-if __name__ == "__main__":
-    unittest.main()

tests/test_issue_comment_workspace_guard.py

@@ -1,7 +1,3 @@
-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
 import os
 import sys
 import unittest

tests/test_issue_content_gate.py

@@ -1,7 +1,3 @@
-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
 """Tests for the pre-create issue content gate (Issue #582)."""
 import sys
 import unittest
@@ -19,10 +15,9 @@ from issue_content_gate import (  # noqa: E402
 from mcp_server import gitea_create_issue  # noqa: E402
 
 FAKE_AUTH = "Basic dGVzdDp0ZXN0"
-ISSUE_WRITE_ENV = shared_mutation_env(
+ISSUE_WRITE_ENV = {
-    "test-author-prgs",
+    "GITEA_ALLOWED_OPERATIONS": "gitea.issue.create",
-    GITEA_ALLOWED_OPERATIONS="gitea.issue.create",
+}
-)
 
 
 class TestNormalizeAndPointers(unittest.TestCase):

tests/test_issue_duplicate_gate.py

@@ -1,7 +1,3 @@
-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
 """Tests for pre-create issue duplicate gate (Issue #207)."""
 import sys
 import unittest
@@ -23,10 +19,9 @@ from mcp_server import gitea_create_issue  # noqa: E402
 from review_proofs import assess_duplicate_search_proof as proof_assess  # noqa: E402
 
 FAKE_AUTH = "Basic dGVzdDp0ZXN0"
-ISSUE_WRITE_ENV = shared_mutation_env(
+ISSUE_WRITE_ENV = {
-    "test-author-prgs",
+    "GITEA_ALLOWED_OPERATIONS": "gitea.issue.create",
-    GITEA_ALLOWED_OPERATIONS="gitea.issue.create",
+}
-)
 CANONICAL_TITLE = (
     "Add hard queue-target resolution wall before PR inventory "
     "or empty-queue claims"
@@ -167,7 +162,7 @@ class TestCreateIssueMCPGate(unittest.TestCase):
         mock_role_check.return_value = (True, [])
         mock_api.return_value = {"number": 1, "html_url": "https://gitea.example.com/issues/1"}
         with patch.dict(__import__("os").environ, ISSUE_WRITE_ENV, clear=True):
-            result = gitea_create_issue(title="Unique new issue", body="body text", remote="prgs")
+            result = gitea_create_issue(title="Unique new issue", body="body text")
         self.assertEqual(result["number"], 1)
 
 

tests/test_issue_work_duplicate_gate.py

@@ -1,7 +1,3 @@
-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
 """Tests for early duplicate-work detection (#400)."""
 import os
 import sys
@@ -148,12 +144,10 @@ class TestInjectableDuplicateFetcher(unittest.TestCase):
             },
         ):
             with tempfile.TemporaryDirectory() as lock_dir:
-                env = shared_mutation_env(
+                with patch.dict(os.environ, {
-                    "test-author-prgs",
+                    "GITEA_ALLOWED_OPERATIONS": "gitea.issue.comment",
-                    include_example_repo=True,
+                    "GITEA_ISSUE_LOCK_DIR": lock_dir,
-                    GITEA_ISSUE_LOCK_DIR=lock_dir,
+                }, clear=True):
-                )
-                with patch.dict(os.environ, env, clear=True):
                     mcp_server.gitea_lock_issue(
                         issue_number=400,
                         branch_name="feat/issue-400-duplicate-work-preflight",
@@ -167,11 +161,7 @@ class TestMcpDuplicateRecheck(unittest.TestCase):
         self._dir = tempfile.TemporaryDirectory()
         self._env_patch = patch.dict(
             os.environ,
-            shared_mutation_env(
+            {"GITEA_ISSUE_LOCK_DIR": self._dir.name},
-                "test-author-prgs",
-                include_example_repo=True,
-                GITEA_ISSUE_LOCK_DIR=self._dir.name,
-            ),
             clear=False,
         )
         self._env_patch.start()
@@ -181,11 +171,6 @@ class TestMcpDuplicateRecheck(unittest.TestCase):
         })
         self._remotes.start()
         mcp_server._IDENTITY_CACHE.clear()
-        try:
-            import session_context_binding as _sc
-            _sc._reset_session_context_for_testing()
-        except Exception:
-            pass
 
     def tearDown(self):
         patch.stopall()
@@ -220,8 +205,6 @@ class TestMcpDuplicateRecheck(unittest.TestCase):
         "allowed_operations": ["gitea.read", "gitea.repo.commit"],
         "forbidden_operations": [],
         "audit_label": "test-author",
-        "allowed_repositories": ["Example-Org/Example-Repo"],
-        "base_url": "https://gitea.example.com",
     })
     @patch("mcp_server.get_auth_header", return_value="token x")
     def test_commit_files_blocked_on_recheck(self, _auth, _profile, mock_gate):
@@ -256,8 +239,6 @@ class TestMcpDuplicateRecheck(unittest.TestCase):
         "allowed_operations": ["gitea.read", "gitea.pr.create"],
         "forbidden_operations": [],
         "audit_label": "test-author",
-        "allowed_repositories": ["Example-Org/Example-Repo"],
-        "base_url": "https://gitea.example.com",
     })
     @patch("mcp_server.get_auth_header", return_value="token x")
     def test_create_pr_returns_handoff_on_duplicate(self, _auth, _profile, mock_gate):

tests/test_issue_write_tool_gates.py

@@ -1,8 +1,3 @@
-import sys
-from pathlib import Path
-sys.path.insert(0, str(Path(__file__).resolve().parent))
-from mutation_profile_fixture import install_deterministic_remote_urls  # noqa: E402
-install_deterministic_remote_urls()
 """Regression tests: issue-write tool gates match gitea_resolve_task_capability (#69)."""
 import json
 import os
@@ -46,7 +41,6 @@ CONFIG = {
                 "gitea.read", "gitea.issue.create", "gitea.issue.close",
                 "gitea.issue.comment"],
             "forbidden_operations": [],
-            "allowed_repositories": ["Scaled-Tech-Consulting/Gitea-Tools", "Example-Org/Example-Repo", "913443/eAgenda"],
             "execution_profile": "full-author",
         },
     },

tests/test_llm_agent_sha.py

@@ -1,7 +1,3 @@
-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
 """Negative tests for LLM-Agent-SHA attribution (#86, Phase 0).
 
 ``LLM-Agent-SHA`` (docs/llm-agent-sha.md) is attribution metadata ONLY. These

tests/test_lock_issue_mcp_registration.py

@@ -1,11 +1,7 @@
+"""Regression tests for gitea_lock_issue MCP tool registration (#521)."""
+
 from __future__ import annotations
 
-
-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
-
 import os
 import tempfile
 import unittest
@@ -17,20 +13,22 @@ from mcp_server import gitea_create_pr, gitea_lock_issue
 
 FAKE_AUTH = "Basic dGVzdDp0ZXN0"
 
-ISSUE_WRITE_ENV = shared_mutation_env(
+ISSUE_WRITE_ENV = {
-    "test-author-prgs",
+    "GITEA_ALLOWED_OPERATIONS": (
-)
+        "gitea.issue.create,gitea.issue.close,gitea.issue.comment"
+    ),
+}
 
-CREATE_PR_ENV = shared_mutation_env(
+CREATE_PR_ENV = {
-    "test-author-prgs",
+    "GITEA_PROFILE_NAME": "author-test",
-    GITEA_ALLOWED_OPERATIONS=(
+    "GITEA_ALLOWED_OPERATIONS": (
         "gitea.read,gitea.pr.create,gitea.branch.push,"
         "gitea.issue.create,gitea.issue.close,gitea.issue.comment"
     ),
-    GITEA_FORBIDDEN_OPERATIONS=(
+    "GITEA_FORBIDDEN_OPERATIONS": (
         "gitea.pr.approve,gitea.pr.merge,gitea.pr.review"
     ),
-)
+}
 
 
 def _clean_master_git_state_for_lock():

tests/test_mcp_server.py

@@ -1,11 +1,3 @@
-import sys
-from pathlib import Path
-sys.path.insert(0, str(Path(__file__).resolve().parent))
-from mutation_profile_fixture import (
-    mutation_profile_env,
-    shared_mutation_env,
-    install_deterministic_remote_urls,
-)  # noqa: E402
 """Tests for the MCP server tool functions.
 
 Each tool is tested by calling the underlying function directly (not through
@@ -55,7 +47,6 @@ from gitea_auth import get_profile  # noqa: E402
 import gitea_config  # noqa: E402
 
 import mcp_server
-install_deterministic_remote_urls()
 import issue_lock_store
 import gitea_auth
 
@@ -230,26 +221,22 @@ def _seed_ready_review_decision(
 
 
 # Issue-write tools are profile-gated (#69).
-# Default remote is dadeschools — use the host-aligned config profile so
+ISSUE_WRITE_ENV = {
-# cross-host session binding does not fail closed (#714).
+    "GITEA_ALLOWED_OPERATIONS": (
-ISSUE_WRITE_ENV = shared_mutation_env(
-    "test-author-dadeschools",
-    GITEA_ALLOWED_OPERATIONS=(
-        "gitea.issue.create,gitea.issue.close,gitea.issue.comment,"
-        "gitea.pr.create,gitea.branch.push,gitea.repo.commit,gitea.read"
-    ),
-)
-
-# create_pr is gated on author profile + namespace (#69, #209).
-# Default remote is dadeschools; use matching config-backed author profile.
-CREATE_PR_ENV = shared_mutation_env(
-    "test-author-dadeschools",
-    GITEA_ALLOWED_OPERATIONS=(
-        "gitea.read,gitea.pr.create,gitea.branch.push,"
         "gitea.issue.create,gitea.issue.close,gitea.issue.comment"
     ),
-    GITEA_FORBIDDEN_OPERATIONS="gitea.pr.approve,gitea.pr.merge,gitea.pr.review",
+}
-)
+
+# create_pr is gated on author profile + namespace (#69, #209).
+CREATE_PR_ENV = {
+    "GITEA_PROFILE_NAME": "author-test",
+    "GITEA_ALLOWED_OPERATIONS": (
+        "gitea.read,gitea.pr.create,gitea.branch.push"
+    ),
+    "GITEA_FORBIDDEN_OPERATIONS": (
+        "gitea.pr.approve,gitea.pr.merge,gitea.pr.review"
+    ),
+}
 
 ISSUE_LOCK_FILE = "/tmp/gitea_issue_lock.json"
 
@@ -300,7 +287,6 @@ def _bind_test_lock(**overrides) -> str:
 # ---------------------------------------------------------------------------
 # Create Issue
 # ---------------------------------------------------------------------------
-
 class TestCreateIssue(unittest.TestCase):
 
     @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
@@ -310,7 +296,7 @@ class TestCreateIssue(unittest.TestCase):
     @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
     def test_creates_issue(self, _auth, _get_all, mock_api, _role):
         mock_api.return_value = {"number": 1, "html_url": "https://gitea.example.com/issues/1"}
-        with patch.dict(os.environ, ISSUE_WRITE_ENV, clear=False):
+        with patch.dict(os.environ, ISSUE_WRITE_ENV, clear=True):
             result = gitea_create_issue(title="Test issue", body="body text")
         self.assertEqual(result["number"], 1)
         self.assertNotIn("url", result)
@@ -328,7 +314,7 @@ class TestCreateIssue(unittest.TestCase):
     def test_create_issue_reveal_opt_in_includes_url(self, _auth, _get_all, mock_api, _role):
         mock_api.return_value = {"number": 1, "html_url": "https://gitea.example.com/issues/1"}
         env = {**ISSUE_WRITE_ENV, "GITEA_MCP_REVEAL_ENDPOINTS": "1"}
-        with patch.dict(os.environ, env, clear=False):
+        with patch.dict(os.environ, env, clear=True):
             result = gitea_create_issue(title="Test issue", body="body text")
         self.assertIn("issues/1", result["url"])
 
@@ -339,18 +325,7 @@ class TestCreateIssue(unittest.TestCase):
     @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
     def test_creates_on_prgs(self, _auth, _get_all, mock_api, _role):
         mock_api.return_value = {"number": 5, "html_url": "https://gitea.prgs.cc/issues/5"}
-        env = {
+        with patch.dict(os.environ, ISSUE_WRITE_ENV, clear=True):
-            **ISSUE_WRITE_ENV,
-            "GITEA_MCP_PROFILE": "test-author-prgs",
-        }
-        with patch.dict(os.environ, env, clear=False):
-            import gitea_config
-            gitea_config._active_profile_override = None
-            try:
-                import session_context_binding as _sc
-                _sc._reset_session_context_for_testing()
-            except Exception:
-                pass
             result = gitea_create_issue(title="Test", remote="prgs")
         self.assertEqual(result["number"], 5)
         url = mock_api.call_args[0][1]
@@ -363,7 +338,7 @@ class TestCreateIssue(unittest.TestCase):
     @patch("mcp_server.api_get_all", return_value=[])
     @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
     def test_create_issue_required_workflow_labels_blocks(self, _auth, _get_all, mock_api, _role):
-        with patch.dict(os.environ, ISSUE_WRITE_ENV, clear=False):
+        with patch.dict(os.environ, ISSUE_WRITE_ENV, clear=True):
             result = gitea_create_issue(
                 title="Test issue",
                 require_workflow_labels=True,
@@ -403,7 +378,7 @@ class TestCreatePR(unittest.TestCase):
         mock_api.side_effect = api_side_effect
         with tempfile.TemporaryDirectory() as lock_dir:
             env = {**CREATE_PR_ENV, "GITEA_ISSUE_LOCK_DIR": lock_dir}
-            with patch.dict(os.environ, env, clear=False):
+            with patch.dict(os.environ, env, clear=True):
                 _bind_test_lock(issue_number=123, branch_name="feat/x", worktree_path=worktree)
                 result = gitea_create_pr(
                     title="feat: X Closes #123",
@@ -446,7 +421,7 @@ class TestCreatePR(unittest.TestCase):
         mock_api.side_effect = api_side_effect
         with tempfile.TemporaryDirectory() as lock_dir:
             env = {**CREATE_PR_ENV, "GITEA_ISSUE_LOCK_DIR": lock_dir, "GITEA_MCP_REVEAL_ENDPOINTS": "1"}
-            with patch.dict(os.environ, env, clear=False):
+            with patch.dict(os.environ, env, clear=True):
                 _bind_test_lock(issue_number=123, branch_name="feat/x", worktree_path=worktree)
                 result = gitea_create_pr(
                     title="feat: X Closes #123",
@@ -472,7 +447,7 @@ class TestCreatePR(unittest.TestCase):
         mock_api.return_value = _issue_with_labels("type:feature", "status:in-progress")
         with tempfile.TemporaryDirectory() as lock_dir:
             env = {**CREATE_PR_ENV, "GITEA_ISSUE_LOCK_DIR": lock_dir}
-            with patch.dict(os.environ, env, clear=False):
+            with patch.dict(os.environ, env, clear=True):
                 _bind_test_lock(issue_number=123, branch_name="feat/x", worktree_path=worktree)
                 result = gitea_create_pr(
                     title="feat: X Closes #123",
@@ -495,7 +470,7 @@ class TestCreatePR(unittest.TestCase):
         worktree = os.path.realpath(os.getcwd())
         with tempfile.TemporaryDirectory() as lock_dir:
             env = {**CREATE_PR_ENV, "GITEA_ISSUE_LOCK_DIR": lock_dir}
-            with patch.dict(os.environ, env, clear=False):
+            with patch.dict(os.environ, env, clear=True):
                 _bind_test_lock(
                     issue_number=123,
                     branch_name="feat/x",
@@ -520,7 +495,7 @@ class TestCloseIssue(unittest.TestCase):
     @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
     def test_closes_issue(self, _auth, mock_api):
         mock_api.return_value = {"state": "closed"}
-        with patch.dict(os.environ, ISSUE_WRITE_ENV, clear=False):
+        with patch.dict(os.environ, ISSUE_WRITE_ENV, clear=True):
             result = gitea_close_issue(issue_number=42)
         self.assertTrue(result["success"])
         self.assertIn("42", result["message"])
@@ -626,7 +601,7 @@ class TestMarkIssue(unittest.TestCase):
             return {}
 
         mock_api.side_effect = api_side_effect
-        with patch.dict(os.environ, ISSUE_WRITE_ENV, clear=False):
+        with patch.dict(os.environ, ISSUE_WRITE_ENV, clear=True):
             result = gitea_mark_issue(issue_number=5, action="start")
         self.assertTrue(result["success"])
         self.assertIn("claimed", result["message"])
@@ -641,7 +616,7 @@ class TestMarkIssue(unittest.TestCase):
         mock_api.side_effect = [
             None,
         ]
-        with patch.dict(os.environ, ISSUE_WRITE_ENV, clear=False):
+        with patch.dict(os.environ, ISSUE_WRITE_ENV, clear=True):
             result = gitea_mark_issue(issue_number=5, action="done")
         self.assertTrue(result["success"])
         self.assertIn("released", result["message"])
@@ -654,7 +629,7 @@ class TestMarkIssue(unittest.TestCase):
     @patch("mcp_server.api_request")
     @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
     def test_missing_label_raises(self, _auth, mock_api, _labels):
-        with patch.dict(os.environ, ISSUE_WRITE_ENV, clear=False):
+        with patch.dict(os.environ, ISSUE_WRITE_ENV, clear=True):
             with self.assertRaises(RuntimeError):
                 gitea_mark_issue(issue_number=5, action="start")
 
@@ -666,12 +641,12 @@ class TestAuthErrors(unittest.TestCase):
 
     @patch("mcp_server.get_auth_header", return_value=None)
     def test_no_credentials_raises(self, _auth):
-        with patch.dict(os.environ, ISSUE_WRITE_ENV, clear=False):
+        with patch.dict(os.environ, ISSUE_WRITE_ENV, clear=True):
             with self.assertRaises(RuntimeError):
                 gitea_create_issue(title="test")
 
     def test_unknown_remote_raises(self):
-        with patch.dict(os.environ, ISSUE_WRITE_ENV, clear=False):
+        with patch.dict(os.environ, ISSUE_WRITE_ENV, clear=True):
             with self.assertRaises(ValueError):
                 gitea_create_issue(title="test", remote="nonexistent")
 
@@ -885,7 +860,7 @@ class TestMergePR(unittest.TestCase):
         ]
         env = {"GITEA_PROFILE_NAME": "gitea-merger",
                "GITEA_ALLOWED_OPERATIONS": "read,merge"}
-        with patch.dict(os.environ, env, clear=False):
+        with patch.dict(os.environ, env, clear=True):
             r = gitea_merge_pr(
                 pr_number=8, confirmation=self._confirm(8),
                 expected_head_sha="abc123", do="squash",
@@ -918,7 +893,7 @@ class TestMergePR(unittest.TestCase):
         ]
         env = {"GITEA_PROFILE_NAME": "gitea-merger",
                "GITEA_ALLOWED_OPERATIONS": "read,merge"}
-        with patch.dict(os.environ, env, clear=False):
+        with patch.dict(os.environ, env, clear=True):
             r = gitea_merge_pr(
                 pr_number=8, confirmation=self._confirm(8),
                 expected_changed_files=["b.py", "a.py"],
@@ -939,7 +914,7 @@ class TestMergePR(unittest.TestCase):
         ]
         env = {"GITEA_PROFILE_NAME": "gitea-merger",
                "GITEA_ALLOWED_OPERATIONS": "read,merge"}
-        with patch.dict(os.environ, env, clear=False):
+        with patch.dict(os.environ, env, clear=True):
             r = gitea_merge_pr(
                 pr_number=8, confirmation=self._confirm(8),
                 expected_head_sha="abc123", remote="prgs",
@@ -970,7 +945,7 @@ class TestMergePR(unittest.TestCase):
         ]
         env = {"GITEA_PROFILE_NAME": "gitea-merger",
                "GITEA_ALLOWED_OPERATIONS": "read,merge"}
-        with patch.dict(os.environ, env, clear=False):
+        with patch.dict(os.environ, env, clear=True):
             r = gitea_merge_pr(
                 pr_number=8, confirmation=self._confirm(8),
                 expected_head_sha="abc123", remote="prgs",
@@ -987,7 +962,7 @@ class TestMergePR(unittest.TestCase):
     def test_missing_confirmation_blocks_with_no_api_call(self, _auth, mock_api):
         env = {"GITEA_PROFILE_NAME": "gitea-merger",
                "GITEA_ALLOWED_OPERATIONS": "read,merge"}
-        with patch.dict(os.environ, env, clear=False):
+        with patch.dict(os.environ, env, clear=True):
             r = gitea_merge_pr(pr_number=8, confirmation="", expected_head_sha="abc123", remote="prgs")
         self.assertFalse(r["performed"])
         self.assertTrue(any("explicit confirmation required" in x for x in r["reasons"]))
@@ -998,7 +973,7 @@ class TestMergePR(unittest.TestCase):
     def test_wrong_confirmation_blocks(self, _auth, mock_api):
         env = {"GITEA_PROFILE_NAME": "gitea-merger",
                "GITEA_ALLOWED_OPERATIONS": "read,merge"}
-        with patch.dict(os.environ, env, clear=False):
+        with patch.dict(os.environ, env, clear=True):
             r = gitea_merge_pr(pr_number=8, confirmation="MERGE PR 9", expected_head_sha="abc123", remote="prgs")
         self.assertFalse(r["performed"])
         mock_api.assert_not_called()
@@ -1010,7 +985,7 @@ class TestMergePR(unittest.TestCase):
     def test_reviewer_profile_cannot_merge(self, _auth, mock_api):
         env = {"GITEA_PROFILE_NAME": "prgs-reviewer",
                "GITEA_ALLOWED_OPERATIONS": "read,approve"}
-        with patch.dict(os.environ, env, clear=False):
+        with patch.dict(os.environ, env, clear=True):
             with self.assertRaises(RuntimeError) as ctx:
                 gitea_merge_pr(
                     pr_number=8, confirmation=self._confirm(8), remote="prgs"
@@ -1024,7 +999,7 @@ class TestMergePR(unittest.TestCase):
         mock_api.side_effect = [{"login": "jcwalker3"}, self._pr("jcwalker3")]
         env = {"GITEA_PROFILE_NAME": "gitea-merger",
                "GITEA_ALLOWED_OPERATIONS": "read,merge"}
-        with patch.dict(os.environ, env, clear=False):
+        with patch.dict(os.environ, env, clear=True):
             r = gitea_merge_pr(
                 pr_number=8, confirmation=self._confirm(8), remote="prgs")
         self.assertFalse(r["performed"])
@@ -1035,7 +1010,7 @@ class TestMergePR(unittest.TestCase):
     def test_unknown_identity_blocks(self, _auth):
         env = {"GITEA_PROFILE_NAME": "gitea-merger",
                "GITEA_ALLOWED_OPERATIONS": "read,merge"}
-        with patch.dict(os.environ, env, clear=False):
+        with patch.dict(os.environ, env, clear=True):
             r = gitea_merge_pr(
                 pr_number=8, confirmation=self._confirm(8), remote="prgs")
         self.assertFalse(r["performed"])
@@ -1047,7 +1022,7 @@ class TestMergePR(unittest.TestCase):
     def test_unknown_profile_blocks(self, _auth, mock_api):
         mock_api.side_effect = [{"login": "merger-bot"}, self._pr("author-bot")]
         env = {}  # no GITEA_ALLOWED_OPERATIONS → empty allowed ops
-        with patch.dict(os.environ, env, clear=False):
+        with patch.dict(os.environ, env, clear=True):
             r = gitea_merge_pr(
                 pr_number=8, confirmation=self._confirm(8), remote="prgs")
         self.assertFalse(r["performed"])
@@ -1119,7 +1094,7 @@ class TestMergePR(unittest.TestCase):
     def test_profile_without_merge_permission_blocks(self, _auth, mock_api):
         env = {"GITEA_PROFILE_NAME": "gitea-reviewer",
                "GITEA_ALLOWED_OPERATIONS": "read,review,approve"}
-        with patch.dict(os.environ, env, clear=False):
+        with patch.dict(os.environ, env, clear=True):
             with self.assertRaises(RuntimeError) as ctx:
                 gitea_merge_pr(
                     pr_number=8, confirmation=self._confirm(8), remote="prgs")
@@ -1135,7 +1110,7 @@ class TestMergePR(unittest.TestCase):
             {"login": "merger-bot"}, self._pr("author-bot", state="closed")]
         env = {"GITEA_PROFILE_NAME": "gitea-merger",
                "GITEA_ALLOWED_OPERATIONS": "read,merge"}
-        with patch.dict(os.environ, env, clear=False):
+        with patch.dict(os.environ, env, clear=True):
             r = gitea_merge_pr(
                 pr_number=8, confirmation=self._confirm(8), remote="prgs")
         self.assertFalse(r["performed"])
@@ -1149,7 +1124,7 @@ class TestMergePR(unittest.TestCase):
             {"login": "merger-bot"}, self._pr("author-bot", mergeable=False)]
         env = {"GITEA_PROFILE_NAME": "gitea-merger",
                "GITEA_ALLOWED_OPERATIONS": "read,merge"}
-        with patch.dict(os.environ, env, clear=False):
+        with patch.dict(os.environ, env, clear=True):
             r = gitea_merge_pr(
                 pr_number=8, confirmation=self._confirm(8), remote="prgs")
         self.assertFalse(r["performed"])
@@ -1163,7 +1138,7 @@ class TestMergePR(unittest.TestCase):
             {"login": "merger-bot"}, self._pr("author-bot", mergeable=None)]
         env = {"GITEA_PROFILE_NAME": "gitea-merger",
                "GITEA_ALLOWED_OPERATIONS": "read,merge"}
-        with patch.dict(os.environ, env, clear=False):
+        with patch.dict(os.environ, env, clear=True):
             r = gitea_merge_pr(
                 pr_number=8, confirmation=self._confirm(8), remote="prgs")
         self.assertFalse(r["performed"])
@@ -1181,7 +1156,7 @@ class TestMergePR(unittest.TestCase):
         ]
         env = {"GITEA_PROFILE_NAME": "gitea-merger",
                "GITEA_ALLOWED_OPERATIONS": "read,merge"}
-        with patch.dict(os.environ, env, clear=False):
+        with patch.dict(os.environ, env, clear=True):
             r = gitea_merge_pr(
                 pr_number=8, confirmation=self._confirm(8),
                 expected_head_sha="deadbeef", remote="prgs")
@@ -1202,7 +1177,7 @@ class TestMergePR(unittest.TestCase):
         ]
         env = {"GITEA_PROFILE_NAME": "gitea-merger",
                "GITEA_ALLOWED_OPERATIONS": "read,merge"}
-        with patch.dict(os.environ, env, clear=False):
+        with patch.dict(os.environ, env, clear=True):
             r = gitea_merge_pr(
                 pr_number=8, confirmation=self._confirm(8),
                 expected_changed_files=["a.py", "b.py"],
@@ -1235,7 +1210,7 @@ class TestMergePR(unittest.TestCase):
         env = {"GITEA_PROFILE_NAME": "gitea-merger",
                "GITEA_ALLOWED_OPERATIONS": "read,merge",
                "GITEA_TOKEN": "super-secret-token"}
-        with patch.dict(os.environ, env, clear=False):
+        with patch.dict(os.environ, env, clear=True):
             r = gitea_merge_pr(
                 pr_number=8, confirmation=self._confirm(8),
                 expected_head_sha="abc123", remote="prgs",
@@ -1254,7 +1229,7 @@ class TestMergePR(unittest.TestCase):
         ]
         env = {"GITEA_PROFILE_NAME": "gitea-merger",
                "GITEA_ALLOWED_OPERATIONS": "read,merge"}
-        with patch.dict(os.environ, env, clear=False):
+        with patch.dict(os.environ, env, clear=True):
             r = gitea_merge_pr(
                 pr_number=8, confirmation=self._confirm(8),
                 expected_head_sha="abc123", remote="prgs",
@@ -1277,7 +1252,7 @@ class TestMergePR(unittest.TestCase):
         ]
         env = {"GITEA_PROFILE_NAME": "gitea-merger",
                "GITEA_ALLOWED_OPERATIONS": "read,merge"}
-        with patch.dict(os.environ, env, clear=False):
+        with patch.dict(os.environ, env, clear=True):
             r = gitea_merge_pr(
                 pr_number=8, confirmation=self._confirm(8), remote="prgs",
                 expected_head_sha=new_sha)
@@ -1300,7 +1275,7 @@ class TestMergePR(unittest.TestCase):
         ]
         env = {"GITEA_PROFILE_NAME": "gitea-merger",
                "GITEA_ALLOWED_OPERATIONS": "read,merge"}
-        with patch.dict(os.environ, env, clear=False):
+        with patch.dict(os.environ, env, clear=True):
             r = gitea_merge_pr(
                 pr_number=8, confirmation=self._confirm(8),
                 expected_head_sha="abc123", remote="prgs",
@@ -1332,7 +1307,7 @@ class TestMergePR(unittest.TestCase):
         ]
         env = {"GITEA_PROFILE_NAME": "gitea-merger",
                "GITEA_ALLOWED_OPERATIONS": "read,merge"}
-        with patch.dict(os.environ, env, clear=False):
+        with patch.dict(os.environ, env, clear=True):
             r = gitea_merge_pr(
                 pr_number=8, confirmation=self._confirm(8),
                 expected_head_sha="abc123", remote="prgs",
@@ -1639,8 +1614,10 @@ class TestCommitFiles(unittest.TestCase):
         files = [
             {"operation": "create", "path": "test.txt", "content": "SGVsbG8="}
         ]
-        env = shared_mutation_env("test-author-dadeschools")
+        env = {
-        with patch.dict(os.environ, env, clear=False):
+            "GITEA_ALLOWED_OPERATIONS": "gitea.repo.commit",
+        }
+        with patch.dict(os.environ, env, clear=True):
             result = gitea_commit_files(
                 files=files,
                 message="Initial commit",
@@ -1755,7 +1732,7 @@ class TestRuntimeProfile(unittest.TestCase):
             "GITEA_ALLOWED_OPERATIONS": "read, review , approve",
             "GITEA_BASE_URL": "https://gitea.example.invalid",
         }
-        with patch.dict(os.environ, env, clear=False):
+        with patch.dict(os.environ, env, clear=True):
             p = get_profile()
         self.assertEqual(p["profile_name"], "gitea-reviewer")
         self.assertEqual(p["allowed_operations"], ["read", "review", "approve"])
@@ -1766,7 +1743,7 @@ class TestRuntimeProfile(unittest.TestCase):
             "GITEA_PROFILE_NAME": "gitea-author",
             "GITEA_TOKEN": "super-secret-token",
         }
-        with patch.dict(os.environ, env, clear=False):
+        with patch.dict(os.environ, env, clear=True):
             p = get_profile()
         blob = repr(p).lower()
         # The token VALUE must never appear. (The field name
@@ -1783,7 +1760,7 @@ class TestRuntimeProfile(unittest.TestCase):
             "GITEA_ALLOWED_OPERATIONS": "read,review,approve",
             "GITEA_TOKEN": "super-secret-token",
         }
-        with patch.dict(os.environ, env, clear=False):
+        with patch.dict(os.environ, env, clear=True):
             result = gitea_whoami(remote="prgs")
         self.assertEqual(result["profile"]["profile_name"], "gitea-reviewer")
         self.assertEqual(
@@ -1804,7 +1781,7 @@ class TestRuntimeProfile(unittest.TestCase):
             "GITEA_AUDIT_LABEL": "reviewer-runtime",
             "GITEA_TOKEN_SOURCE": "keychain:prgs-reviewer-token",
         }
-        with patch.dict(os.environ, env, clear=False):
+        with patch.dict(os.environ, env, clear=True):
             result = gitea_whoami(remote="prgs")
         profile = result["profile"]
         self.assertEqual(profile["environment"], None)
@@ -1873,7 +1850,7 @@ class TestProfileDiscovery(unittest.TestCase):
             "GITEA_AUDIT_LABEL": "reviewer-runtime",
             "GITEA_TOKEN_SOURCE": "GITEA_TOKEN",
         }
-        with patch.dict(os.environ, env, clear=False):
+        with patch.dict(os.environ, env, clear=True):
             p = get_profile()
         self.assertEqual(p["forbidden_operations"], ["merge", "branch.push"])
         self.assertEqual(p["audit_label"], "reviewer-runtime")
@@ -1889,7 +1866,7 @@ class TestProfileDiscovery(unittest.TestCase):
             "GITEA_TOKEN_SOURCE": "GITEA_TOKEN",
             "GITEA_TOKEN": "super-secret-token",
         }
-        with patch.dict(os.environ, env, clear=False):
+        with patch.dict(os.environ, env, clear=True):
             result = gitea_get_profile(remote="prgs")
         self.assertEqual(result["profile_name"], "gitea-reviewer")
         self.assertEqual(result["allowed_operations"], ["read", "review", "approve"])
@@ -1910,7 +1887,7 @@ class TestProfileDiscovery(unittest.TestCase):
     def test_discovery_never_exposes_secrets(self, _auth, mock_api):
         mock_api.return_value = {"id": 3, "login": "reviewer-bot"}
         env = {"GITEA_PROFILE_NAME": "gitea-reviewer", "GITEA_TOKEN": "super-secret-token"}
-        with patch.dict(os.environ, env, clear=False):
+        with patch.dict(os.environ, env, clear=True):
             result = gitea_get_profile(remote="prgs")
         blob = repr(result).lower()
         for secret in ("super-secret-token", "token dgvzd", "authorization", "basic ", FAKE_AUTH.lower()):
@@ -1992,7 +1969,7 @@ class TestPrEligibility(unittest.TestCase):
         mock_api.side_effect = [{"login": "reviewer-bot"}, self._pr("author-bot")]
         env = {"GITEA_PROFILE_NAME": "gitea-reviewer",
                "GITEA_ALLOWED_OPERATIONS": "read,review,approve"}
-        with patch.dict(os.environ, env, clear=False):
+        with patch.dict(os.environ, env, clear=True):
             r = gitea_check_pr_eligibility(pr_number=5, action="review", remote="prgs")
         self.assertTrue(r["eligible"])
         self.assertEqual(r["authenticated_user"], "reviewer-bot")
@@ -2005,7 +1982,7 @@ class TestPrEligibility(unittest.TestCase):
         mock_api.side_effect = [{"login": "jcwalker3"}, self._pr("jcwalker3")]
         env = {"GITEA_PROFILE_NAME": "gitea-merger",
                "GITEA_ALLOWED_OPERATIONS": "read,merge"}
-        with patch.dict(os.environ, env, clear=False):
+        with patch.dict(os.environ, env, clear=True):
             r = gitea_check_pr_eligibility(pr_number=8, action="merge", remote="prgs")
         self.assertFalse(r["eligible"])
         self.assertIn("authenticated user is PR author", r["reasons"])
@@ -2016,7 +1993,7 @@ class TestPrEligibility(unittest.TestCase):
         mock_api.side_effect = [{"login": "jcwalker3"}, self._pr("jcwalker3")]
         env = {"GITEA_PROFILE_NAME": "gitea-reviewer",
                "GITEA_ALLOWED_OPERATIONS": "read,review,approve"}
-        with patch.dict(os.environ, env, clear=False):
+        with patch.dict(os.environ, env, clear=True):
             r = gitea_check_pr_eligibility(pr_number=8, action="approve", remote="prgs")
         self.assertFalse(r["eligible"])
         self.assertIn("authenticated user is PR author", r["reasons"])
@@ -2031,7 +2008,7 @@ class TestPrEligibility(unittest.TestCase):
         ]
         env = {"GITEA_PROFILE_NAME": "gitea-merger",
                "GITEA_ALLOWED_OPERATIONS": "read,merge"}
-        with patch.dict(os.environ, env, clear=False):
+        with patch.dict(os.environ, env, clear=True):
             r = gitea_check_pr_eligibility(pr_number=8, action="merge", remote="prgs")
         self.assertFalse(r["eligible"])
         self.assertIsNone(r["mergeable"])
@@ -2043,7 +2020,7 @@ class TestPrEligibility(unittest.TestCase):
         mock_api.side_effect = [{"login": "author-bot"}, self._pr("someone-else")]
         env = {"GITEA_PROFILE_NAME": "gitea-author",
                "GITEA_ALLOWED_OPERATIONS": "read,pr.create"}
-        with patch.dict(os.environ, env, clear=False):
+        with patch.dict(os.environ, env, clear=True):
             r = gitea_check_pr_eligibility(pr_number=8, action="merge", remote="prgs")
         self.assertFalse(r["eligible"])
         self.assertIn("profile is not allowed to merge", r["reasons"])
@@ -2055,7 +2032,7 @@ class TestPrEligibility(unittest.TestCase):
         env = {"GITEA_PROFILE_NAME": "gitea-reviewer",
                "GITEA_ALLOWED_OPERATIONS": "read,review",
                "GITEA_FORBIDDEN_OPERATIONS": "merge"}
-        with patch.dict(os.environ, env, clear=False):
+        with patch.dict(os.environ, env, clear=True):
             r = gitea_check_pr_eligibility(pr_number=8, action="merge", remote="prgs")
         self.assertFalse(r["eligible"])
         self.assertIn("profile forbids 'merge'", r["reasons"])
@@ -2066,7 +2043,7 @@ class TestPrEligibility(unittest.TestCase):
         mock_api.side_effect = [{"login": "reviewer-bot"}, self._pr("author-bot", state="closed")]
         env = {"GITEA_PROFILE_NAME": "gitea-reviewer",
                "GITEA_ALLOWED_OPERATIONS": "read,review,approve"}
-        with patch.dict(os.environ, env, clear=False):
+        with patch.dict(os.environ, env, clear=True):
             r = gitea_check_pr_eligibility(pr_number=8, action="approve", remote="prgs")
         self.assertFalse(r["eligible"])
         self.assertIn("PR is not open (state=closed)", r["reasons"])
@@ -2075,7 +2052,7 @@ class TestPrEligibility(unittest.TestCase):
     def test_unknown_identity_fails_closed(self, _auth):
         env = {"GITEA_PROFILE_NAME": "gitea-merger",
                "GITEA_ALLOWED_OPERATIONS": "read,merge"}
-        with patch.dict(os.environ, env, clear=False):
+        with patch.dict(os.environ, env, clear=True):
             r = gitea_check_pr_eligibility(pr_number=8, action="merge", remote="prgs")
         self.assertFalse(r["eligible"])
         self.assertIn("authenticated identity could not be determined", r["reasons"])
@@ -2094,7 +2071,7 @@ class TestPrEligibility(unittest.TestCase):
         env = {"GITEA_PROFILE_NAME": "gitea-reviewer",
                "GITEA_ALLOWED_OPERATIONS": "read,review",
                "GITEA_TOKEN": "super-secret-token"}
-        with patch.dict(os.environ, env, clear=False):
+        with patch.dict(os.environ, env, clear=True):
             r = gitea_check_pr_eligibility(pr_number=5, action="review", remote="prgs")
         blob = repr(r).lower()
         for secret in ("super-secret-token", "authorization", "basic ", FAKE_AUTH.lower()):
@@ -2291,7 +2268,7 @@ class TestSubmitPrReview(unittest.TestCase):
         mock_api.side_effect = [{"login": "jcwalker3"}, self._pr("jcwalker3")]
         env = {"GITEA_PROFILE_NAME": "gitea-reviewer",
                "GITEA_ALLOWED_OPERATIONS": "read,review,approve"}
-        with patch.dict(os.environ, env, clear=False):
+        with patch.dict(os.environ, env, clear=True):
             r = gitea_submit_pr_review(
                 pr_number=8, action="approve", remote="prgs",
                 final_review_decision_ready=True,
@@ -2310,7 +2287,7 @@ class TestSubmitPrReview(unittest.TestCase):
         ]
         env = {"GITEA_PROFILE_NAME": "gitea-reviewer",
                "GITEA_ALLOWED_OPERATIONS": "read,review,approve"}
-        with patch.dict(os.environ, env, clear=False):
+        with patch.dict(os.environ, env, clear=True):
             r = gitea_submit_pr_review(
                 pr_number=8, action="approve", body="LGTM", remote="prgs",
                 final_review_decision_ready=True,
@@ -2340,7 +2317,7 @@ class TestSubmitPrReview(unittest.TestCase):
         ]
         env = {"GITEA_PROFILE_NAME": "gitea-reviewer",
                "GITEA_ALLOWED_OPERATIONS": "read,review,approve"}
-        with patch.dict(os.environ, env, clear=False):
+        with patch.dict(os.environ, env, clear=True):
             r = gitea_submit_pr_review(
                 pr_number=8, action="approve", body="LGTM", remote="prgs",
                 final_review_decision_ready=True,
@@ -2360,7 +2337,7 @@ class TestSubmitPrReview(unittest.TestCase):
         ]
         env = {"GITEA_PROFILE_NAME": "gitea-reviewer",
                "GITEA_ALLOWED_OPERATIONS": "read,review,approve"}
-        with patch.dict(os.environ, env, clear=False):
+        with patch.dict(os.environ, env, clear=True):
             r = gitea_submit_pr_review(
                 pr_number=8, action="approve", body="LGTM", remote="prgs",
                 final_review_decision_ready=True,
@@ -2386,7 +2363,7 @@ class TestSubmitPrReview(unittest.TestCase):
         ]
         env = {"GITEA_PROFILE_NAME": "gitea-reviewer",
                "GITEA_ALLOWED_OPERATIONS": "read,review,request_changes"}
-        with patch.dict(os.environ, env, clear=False):
+        with patch.dict(os.environ, env, clear=True):
             r = gitea_submit_pr_review(
                 pr_number=8, action="request_changes",
                 body="needs work", remote="prgs",
@@ -2407,7 +2384,7 @@ class TestSubmitPrReview(unittest.TestCase):
             mock_api.side_effect = [{"login": "reviewer-bot"}, self._pr("author-bot")]
             env = {"GITEA_PROFILE_NAME": "gitea-reviewer",
                    "GITEA_ALLOWED_OPERATIONS": "read,review"}  # no request_changes
-            with patch.dict(os.environ, env, clear=False):
+            with patch.dict(os.environ, env, clear=True):
                 r = gitea_submit_pr_review(
                     pr_number=8, action="request_changes", remote="prgs",
                     final_review_decision_ready=True,
@@ -2428,7 +2405,7 @@ class TestSubmitPrReview(unittest.TestCase):
         gitea_mark_final_review_decision(8, "comment", remote="prgs", expected_head_sha="abc123")
         env = {"GITEA_PROFILE_NAME": "gitea-reviewer",
                "GITEA_ALLOWED_OPERATIONS": "read,review"}
-        with patch.dict(os.environ, env, clear=False):
+        with patch.dict(os.environ, env, clear=True):
             r = gitea_submit_pr_review(
                 pr_number=8, action="comment", body="finding", remote="prgs",
                 final_review_decision_ready=True,
@@ -2446,7 +2423,7 @@ class TestSubmitPrReview(unittest.TestCase):
             ]
             env = {"GITEA_PROFILE_NAME": "gitea-reviewer",
                    "GITEA_ALLOWED_OPERATIONS": "read,review"}
-            with patch.dict(os.environ, env, clear=False):
+            with patch.dict(os.environ, env, clear=True):
                 gitea_mark_final_review_decision(
                     8, "comment", remote="prgs", expected_head_sha="abc123",
                 )
@@ -2462,7 +2439,7 @@ class TestSubmitPrReview(unittest.TestCase):
     def test_unknown_identity_blocks(self, _auth):
         env = {"GITEA_PROFILE_NAME": "gitea-reviewer",
                "GITEA_ALLOWED_OPERATIONS": "read,review,approve"}
-        with patch.dict(os.environ, env, clear=False):
+        with patch.dict(os.environ, env, clear=True):
             r = gitea_submit_pr_review(
                 pr_number=8, action="approve", remote="prgs",
                 final_review_decision_ready=True,
@@ -2477,7 +2454,7 @@ class TestSubmitPrReview(unittest.TestCase):
             mock_api.side_effect = [{"login": "reviewer-bot"}, self._pr("author-bot")]
             env = {"GITEA_PROFILE_NAME": "gitea-author",
                    "GITEA_ALLOWED_OPERATIONS": "read,pr.create"}
-            with patch.dict(os.environ, env, clear=False):
+            with patch.dict(os.environ, env, clear=True):
                 r = gitea_submit_pr_review(
                     pr_number=8, action="approve", remote="prgs",
                     final_review_decision_ready=True,
@@ -2496,7 +2473,7 @@ class TestSubmitPrReview(unittest.TestCase):
             ]
             env = {"GITEA_PROFILE_NAME": "gitea-reviewer",
                    "GITEA_ALLOWED_OPERATIONS": "read,review,approve"}
-            with patch.dict(os.environ, env, clear=False):
+            with patch.dict(os.environ, env, clear=True):
                 r = gitea_submit_pr_review(
                     pr_number=8, action="approve",
                     expected_head_sha="abc123", remote="prgs",
@@ -2520,7 +2497,7 @@ class TestSubmitPrReview(unittest.TestCase):
             ]
             env = {"GITEA_PROFILE_NAME": "gitea-reviewer",
                    "GITEA_ALLOWED_OPERATIONS": "read,review,approve"}
-            with patch.dict(os.environ, env, clear=False):
+            with patch.dict(os.environ, env, clear=True):
                 r = gitea_submit_pr_review(
                     pr_number=8, action="approve",
                     expected_head_sha="abc123", remote="prgs",
@@ -2549,7 +2526,7 @@ class TestSubmitPrReview(unittest.TestCase):
         env = {"GITEA_PROFILE_NAME": "gitea-reviewer",
                "GITEA_ALLOWED_OPERATIONS": "read,review,approve",
                "GITEA_TOKEN": "super-secret-token"}
-        with patch.dict(os.environ, env, clear=False):
+        with patch.dict(os.environ, env, clear=True):
             gitea_mark_final_review_decision(5, "approve", remote="prgs", expected_head_sha="abc123")
             r = gitea_submit_pr_review(
                 pr_number=5, action="approve", remote="prgs",
@@ -2569,7 +2546,7 @@ class TestSubmitPrReview(unittest.TestCase):
         ]
         env = {"GITEA_PROFILE_NAME": "gitea-reviewer",
                "GITEA_ALLOWED_OPERATIONS": "read,review,approve"}
-        with patch.dict(os.environ, env, clear=False):
+        with patch.dict(os.environ, env, clear=True):
             r = gitea_submit_pr_review(
                 pr_number=8, action="approve", remote="prgs",
                 final_review_decision_ready=True,
@@ -2646,7 +2623,7 @@ class TestSubmitPrReview(unittest.TestCase):
         try:
             env = {"GITEA_PROFILE_NAME": "gitea-reviewer",
                    "GITEA_ALLOWED_OPERATIONS": "read,review,approve"}
-            with patch.dict(os.environ, env, clear=False):
+            with patch.dict(os.environ, env, clear=True):
                 r = gitea_submit_pr_review(
                     pr_number=8, action="approve", remote="prgs",
                     final_review_decision_ready=True,
@@ -2679,7 +2656,7 @@ class TestSubmitPrReview(unittest.TestCase):
                "GITEA_ALLOWED_OPERATIONS": "read,review,approve,request_changes"}
         with patch("mcp_server.gitea_get_pr_review_feedback",
                    return_value=dict(_NO_BLOCKER_FEEDBACK)):
-            with patch.dict(os.environ, env, clear=False):
+            with patch.dict(os.environ, env, clear=True):
                 first = gitea_submit_pr_review(
                     pr_number=8, action="approve", remote="prgs",
                     final_review_decision_ready=True,
@@ -2704,7 +2681,7 @@ class TestSubmitPrReview(unittest.TestCase):
     def test_remote_org_repo_validation_mismatch(self, _auth, mock_api):
         env = {"GITEA_PROFILE_NAME": "gitea-reviewer",
                "GITEA_ALLOWED_OPERATIONS": "read,review,approve"}
-        with patch.dict(os.environ, env, clear=False):
+        with patch.dict(os.environ, env, clear=True):
             # Mark decision for specific remote, org, repo
             gitea_mark_final_review_decision(8, "approve", remote="prgs", expected_head_sha="abc123", org="MyOrg", repo="MyRepo")
 
@@ -2742,29 +2719,13 @@ if __name__ == "__main__":
 class TestTrackerHygieneCleanup(unittest.TestCase):
 
     def setUp(self):
-        self._mut_env = patch.dict(
-            os.environ,
-            shared_mutation_env("test-author-dadeschools"),
-            clear=False,
-        )
-        self._mut_env.start()
         mcp_server.gitea_load_review_workflow()
         self.mock_api = patch("mcp_server.api_request").start()
         self.mock_auth = patch("mcp_server.get_auth_header", return_value=FAKE_AUTH).start()
         patch("gitea_audit.audit_enabled", return_value=True).start()
         self.mock_audit = patch("gitea_audit.write_event").start()
         # gitea.pr.close: closing a PR via gitea_edit_pr is capability-gated (#216).
-        patch("mcp_server.get_profile", return_value={
+        patch("mcp_server.get_profile", return_value={"profile_name": "test", "allowed_operations": ["read", "merge", "edit", "close", "gitea.pr.close", "gitea.issue.close"], "audit_label": "test", "forbidden_operations": []}).start()
-            "profile_name": "test-author-dadeschools",
-            "allowed_operations": [
-                "read", "merge", "edit", "close",
-                "gitea.pr.close", "gitea.issue.close", "gitea.read",
-            ],
-            "audit_label": "test",
-            "forbidden_operations": [],
-            "allowed_repositories": ["913443/eAgenda"],
-            "base_url": "https://gitea.dadeschools.net",
-        }).start()
         patch("mcp_server._list_pr_lease_comments", return_value=[]).start()
         patch(
             "mcp_server._pr_work_lease_reviewer_block",
@@ -2772,11 +2733,6 @@ class TestTrackerHygieneCleanup(unittest.TestCase):
         ).start()
 
     def tearDown(self):
-        try:
-            self._mut_env.stop()
-        except Exception:
-            pass
-
         patch.stopall()
 
     def test_close_issue_removes_in_progress(self):
@@ -3182,7 +3138,7 @@ class TestEndpointRedaction(unittest.TestCase):
             "GITEA_BASE_URL": "https://gitea.example.invalid",
             "GITEA_TOKEN_SOURCE": "keychain:some-item-id",
         }
-        with patch.dict(os.environ, env, clear=False):
+        with patch.dict(os.environ, env, clear=True):
             result = gitea_get_profile(remote="prgs",
                                        resolve_identity=False)
         blob = repr(result)
@@ -3204,7 +3160,7 @@ class TestEndpointRedaction(unittest.TestCase):
             "GITEA_TOKEN_SOURCE": "keychain:some-item-id",
             "GITEA_MCP_REVEAL_ENDPOINTS": "1",
         }
-        with patch.dict(os.environ, env, clear=False):
+        with patch.dict(os.environ, env, clear=True):
             result = gitea_get_profile(remote="prgs",
                                        resolve_identity=False)
         self.assertEqual(result["server"], "https://gitea.prgs.cc")
@@ -3216,7 +3172,7 @@ class TestEndpointRedaction(unittest.TestCase):
         try:
             env = {"GITEA_MCP_CONFIG": path,
                    "GITEA_MCP_PROFILE": "prgs-author"}
-            with patch.dict(os.environ, env, clear=False), \
+            with patch.dict(os.environ, env, clear=True), \
                  patch("gitea_config._keychain_token", return_value="x"):
                 result = gitea_audit_config()
         finally:
@@ -3304,7 +3260,7 @@ class TestIssueCommentTools(unittest.TestCase):
     def test_list_reveal_opt_in_includes_url(self, _auth, mock_api):
         mock_api.return_value = [self._comment()]
         env = dict(self.AUTHOR_ENV, GITEA_MCP_REVEAL_ENDPOINTS="1")
-        with patch.dict(os.environ, env, clear=False):
+        with patch.dict(os.environ, env, clear=True):
             result = gitea_list_issue_comments(issue_number=9, remote="prgs")
         self.assertIn("issuecomment-101", result["comments"][0]["url"])
 
@@ -3313,7 +3269,7 @@ class TestIssueCommentTools(unittest.TestCase):
     def test_list_blocked_without_read_permission(self, _auth, mock_api):
         env = {"GITEA_PROFILE_NAME": "gitea-writer-only",
                "GITEA_ALLOWED_OPERATIONS": "gitea.issue.comment"}
-        with patch.dict(os.environ, env, clear=False):
+        with patch.dict(os.environ, env, clear=True):
             result = gitea_list_issue_comments(issue_number=9, remote="prgs")
         self.assertFalse(result["success"])
         self.assertTrue(result["reasons"])
@@ -3360,7 +3316,7 @@ class TestIssueCommentTools(unittest.TestCase):
     def test_create_reveal_opt_in_includes_url(self, _auth, mock_api):
         mock_api.return_value = self._comment(555)
         env = dict(self.AUTHOR_ENV, GITEA_MCP_REVEAL_ENDPOINTS="1")
-        with patch.dict(os.environ, env, clear=False):
+        with patch.dict(os.environ, env, clear=True):
             result = gitea_create_issue_comment(
                 issue_number=9, body="posted", remote="prgs")
         self.assertIn("issuecomment-555", result["url"])
@@ -3372,7 +3328,7 @@ class TestIssueCommentTools(unittest.TestCase):
                "GITEA_ALLOWED_OPERATIONS":
                    "gitea.read,gitea.pr.review,gitea.pr.comment,"
                    "gitea.pr.approve,gitea.pr.merge"}
-        with patch.dict(os.environ, env, clear=False):
+        with patch.dict(os.environ, env, clear=True):
             result = gitea_create_issue_comment(
                 issue_number=9, body="posted", remote="prgs")
         self.assertFalse(result["success"])
@@ -3386,7 +3342,7 @@ class TestIssueCommentTools(unittest.TestCase):
         env = {"GITEA_PROFILE_NAME": "gitea-author",
                "GITEA_ALLOWED_OPERATIONS": "gitea.read,gitea.issue.comment",
                "GITEA_FORBIDDEN_OPERATIONS": "gitea.issue.comment"}
-        with patch.dict(os.environ, env, clear=False):
+        with patch.dict(os.environ, env, clear=True):
             result = gitea_create_issue_comment(
                 issue_number=9, body="posted", remote="prgs")
         self.assertFalse(result["success"])
@@ -3522,7 +3478,7 @@ class TestIssueCommentPermissionSeparation(unittest.TestCase):
                "GITEA_ALLOWED_OPERATIONS":
                    "gitea.branch.create,gitea.branch.push,gitea.pr.comment,"
                    "gitea.pr.create,gitea.read,gitea.repo.commit"}
-        with patch.dict(os.environ, env, clear=False):
+        with patch.dict(os.environ, env, clear=True):
             result = gitea_create_issue_comment(
                 issue_number=51, body="audit findings", remote="prgs")
         self.assertFalse(result["success"])
@@ -3707,12 +3663,11 @@ class TestIssueLocking(unittest.TestCase):
 
     def setUp(self):
         self._lock_dir = tempfile.TemporaryDirectory()
-        # Most locking tests target remote=prgs; host-align profile (#714).
         env = {
-            **shared_mutation_env("test-author-prgs"),
+            **ISSUE_WRITE_ENV,
             "GITEA_ISSUE_LOCK_DIR": self._lock_dir.name,
         }
-        self._env_patcher = patch.dict(os.environ, env, clear=False)
+        self._env_patcher = patch.dict(os.environ, env, clear=True)
         self._env_patcher.start()
         self._dup_fetcher_patcher = patch(
             "mcp_server.issue_duplicate_context_fetcher",
@@ -3726,9 +3681,8 @@ class TestIssueLocking(unittest.TestCase):
         self._lock_dir.cleanup()
 
     def _create_pr_env(self) -> dict:
-        # PR-locking tests call create_pr with remote=prgs.
         return {
-            **shared_mutation_env("test-author-prgs"),
+            **CREATE_PR_ENV,
             "GITEA_ISSUE_LOCK_DIR": self._lock_dir.name,
         }
 
@@ -3747,7 +3701,7 @@ class TestIssueLocking(unittest.TestCase):
         self.assertIn("created_at", res["work_lease"])
         self.assertIn("expires_at", res["work_lease"])
         self.assertIn("last_heartbeat_at", res["work_lease"])
-        self.assertIn(res["work_lease"]["claimant"]["profile"], ("gitea-default", "test-author-prgs", "prgs-author", "gitea-author"))
+        self.assertEqual(res["work_lease"]["claimant"]["profile"], "gitea-default")
         self.assertIn("lock_file_path", res)
         lock = issue_lock_store.read_lock_file(res["lock_file_path"])
         self.assertIn("worktree_path", lock)
@@ -3866,7 +3820,7 @@ class TestIssueLocking(unittest.TestCase):
     @patch("mcp_server.api_get_all", return_value=[])
     @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
     def test_lock_issue_blocks_active_same_operation_lease(self, _auth, _api, _git_state):
-        prgs_repo = "Gitea-Tools"  # #714 session-bound repo (not REMOTES Timesheet default)
+        prgs_repo = mcp_server.REMOTES["prgs"]["repo"]
         issue_lock_store.save_lock_file(
             issue_lock_store.lock_file_path(
                 remote="prgs",
@@ -3903,7 +3857,7 @@ class TestIssueLocking(unittest.TestCase):
         Dead-pid / missing-worktree reclaim is covered by issue_lock_store unit tests.
         This MCP path must remain fail-closed for sticky expired foreign ownership.
         """
-        prgs_repo = "Gitea-Tools"  # #714 session-bound repo (not REMOTES Timesheet default)
+        prgs_repo = mcp_server.REMOTES["prgs"]["repo"]
         # Present worktree + live pid => reclaim_allowed=False even though expires_at is past.
         sticky_worktree = tempfile.mkdtemp(prefix="gitea-sticky-lease-")
         self.addCleanup(lambda: shutil.rmtree(sticky_worktree, ignore_errors=True))
@@ -4097,12 +4051,12 @@ class TestIssueLocking(unittest.TestCase):
         worktree = os.path.realpath(os.getcwd())
         with tempfile.TemporaryDirectory() as lock_dir:
             env = {**self._create_pr_env(), "GITEA_ISSUE_LOCK_DIR": lock_dir}
-            with patch.dict(os.environ, env, clear=False):
+            with patch.dict(os.environ, env, clear=True):
                 issue_lock_store.save_lock_file(
                     issue_lock_store.lock_file_path(
                         remote="prgs",
                         org="Scaled-Tech-Consulting",
-                        repo="Gitea-Tools",  # #714 session-bound
+                        repo=mcp_server.REMOTES["prgs"]["repo"],
                         issue_number=447,
                         lock_dir=lock_dir,
                     ),
@@ -4111,7 +4065,7 @@ class TestIssueLocking(unittest.TestCase):
                     branch_name="feat/issue-447-lock-provenance",
                     remote="prgs",
                     org="Scaled-Tech-Consulting",
-                    repo="Gitea-Tools",  # #714 session-bound
+                    repo=mcp_server.REMOTES["prgs"]["repo"],
                     worktree_path=worktree,
                     lock_provenance=None,
                 ),

tests/test_merger_lease_adoption.py

@@ -1,7 +1,3 @@
-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
 """Merger lease adoption / recovery (#536)."""
 
 import os

tests/test_op_normalization.py

@@ -1,7 +1,3 @@
-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
 """Operation-name normalization table and enforcement tests — issue #106.
 
 Covers the required matrix from #106:

tests/test_operation_scoped_roles.py

@@ -1,9 +1,4 @@
-"""Tests for operation-scoped role selection (#228, updated by #714).
+"""Tests for operation-scoped role selection and automatic dispatch switching (#228)."""
-
-#714 removes silent automatic profile substitution. Capability resolution
-and mutation gates evaluate only the active profile; explicit
-``gitea_activate_profile`` is required to switch roles.
-"""
 import os
 import sys
 import json
@@ -20,7 +15,6 @@ from reviewer_worktree import assess_author_worktree_continuity
 
 CONFIG_TEST = {
     "version": 2,
-    "allow_runtime_switching": True,
     "contexts": {
         "ctx": {
             "enabled": True,
@@ -36,11 +30,9 @@ CONFIG_TEST = {
             "context": "ctx",
             "role": "author",
             "username": "author-user",
-            "base_url": "https://gitea.example.com",
             "auth": {"type": "env", "name": "GITEA_TOKEN_AUTHOR"},
             "allowed_operations": ["gitea.read", "gitea.issue.create", "gitea.pr.create", "gitea.branch.push", "gitea.issue.comment"],
             "forbidden_operations": ["gitea.pr.approve", "gitea.pr.merge"],
-            "allowed_repositories": ["Scaled-Tech-Consulting/Gitea-Tools", "913443/eAgenda"],
             "execution_profile": "author-profile"
         },
         "reviewer-profile": {
@@ -48,11 +40,9 @@ CONFIG_TEST = {
             "context": "ctx",
             "role": "reviewer",
             "username": "reviewer-user",
-            "base_url": "https://gitea.example.com",
             "auth": {"type": "env", "name": "GITEA_TOKEN_REVIEWER"},
             "allowed_operations": ["gitea.read", "gitea.pr.review", "gitea.pr.approve", "gitea.pr.merge", "gitea.issue.comment"],
             "forbidden_operations": ["gitea.pr.create", "gitea.branch.push"],
-            "allowed_repositories": ["Scaled-Tech-Consulting/Gitea-Tools", "913443/eAgenda"],
             "execution_profile": "reviewer-profile"
         }
     }
@@ -67,15 +57,6 @@ class TestOperationScopedRoles(unittest.TestCase):
         })
         self._remotes_patch.start()
         mcp_server._IDENTITY_CACHE.clear()
-        self._url = patch.object(
-            mcp_server,
-            "_local_git_remote_url",
-            side_effect=lambda n: {
-                "prgs": "https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools.git",
-                "dadeschools": "https://gitea.dadeschools.net/913443/eAgenda.git",
-            }.get(n),
-        )
-        self._url.start()
         gitea_config._active_profile_override = None
         mcp_server._MUTATION_AUTHORITY = None
         self._dir = tempfile.TemporaryDirectory()
@@ -83,11 +64,6 @@ class TestOperationScopedRoles(unittest.TestCase):
         self._write_config(CONFIG_TEST)
 
     def tearDown(self):
-        try:
-            self._url.stop()
-        except Exception:
-            pass
-
         self._remotes_patch.stop()
         mcp_server._IDENTITY_CACHE.clear()
         gitea_config._active_profile_override = None
@@ -109,71 +85,62 @@ class TestOperationScopedRoles(unittest.TestCase):
         return env
 
     @patch("mcp_server.api_request")
-    def test_no_auto_switch_to_reviewer(self, mock_api):
+    def test_auto_switch_to_reviewer(self, mock_api):
-        # #714: capability resolution must not silently switch author → reviewer
+        # mock identity resolution to return username matching profile
         mock_api.side_effect = lambda method, url, header: (
             {"login": "reviewer-user"} if "reviewer-pass" in str(header) else {"login": "author-user"}
         )
         with patch.dict(os.environ, self._env("author-profile")):
+            # initially we are author-profile
             self.assertEqual(gitea_config.selected_profile_name(), "author-profile")
             self.assertEqual(mcp_server.get_profile()["profile_name"], "author-profile")
 
+            # resolve a reviewer task (review_pr)
             res = mcp_server.gitea_resolve_task_capability(task="review_pr", remote="prgs")
             
-            self.assertFalse(res["allowed_in_current_session"])
+            # verify it automatically switched to reviewer-profile
-            self.assertFalse(res["available_in_session"])
+            self.assertTrue(res["allowed_in_current_session"])
-            self.assertEqual(res["active_profile"], "author-profile")
+            self.assertTrue(res["available_in_session"])
-            self.assertEqual(res["active_identity"], "author-user")
+            self.assertEqual(res["active_profile"], "reviewer-profile")
-            self.assertEqual(gitea_config.selected_profile_name(), "author-profile")
+            self.assertEqual(res["active_identity"], "reviewer-user")
-            self.assertIn("reviewer-profile", res["matching_configured_profile"])
+            self.assertEqual(gitea_config.selected_profile_name(), "reviewer-profile")
-            self.assertFalse(res.get("auto_profile_substitution", True))
 
     @patch("mcp_server.api_request")
-    def test_no_auto_switch_to_author(self, mock_api):
+    def test_auto_switch_to_author(self, mock_api):
         mock_api.side_effect = lambda method, url, header: (
             {"login": "reviewer-user"} if "reviewer-pass" in str(header) else {"login": "author-user"}
         )
         with patch.dict(os.environ, self._env("reviewer-profile")):
+            # initially we are reviewer-profile
             self.assertEqual(gitea_config.selected_profile_name(), "reviewer-profile")
 
+            # resolve an author task (create_issue)
             res = mcp_server.gitea_resolve_task_capability(task="create_issue", remote="prgs")
             
-            self.assertFalse(res["allowed_in_current_session"])
+            # verify it automatically switched to author-profile
-            self.assertFalse(res["available_in_session"])
-            self.assertEqual(res["active_profile"], "reviewer-profile")
-            self.assertEqual(res["active_identity"], "reviewer-user")
-            self.assertEqual(gitea_config.selected_profile_name(), "reviewer-profile")
-            self.assertIn("author-profile", res["matching_configured_profile"])
-
-    @patch("mcp_server.api_request")
-    def test_explicit_activate_profile_still_works(self, mock_api):
-        mock_api.side_effect = lambda method, url, header: (
-            {"login": "reviewer-user"} if "reviewer-pass" in str(header) else {"login": "author-user"}
-        )
-        with patch.dict(os.environ, self._env("author-profile")):
-            act = mcp_server.gitea_activate_profile(
-                profile_name="reviewer-profile", remote="prgs"
-            )
-            self.assertTrue(act["success"])
-            self.assertEqual(act["after_profile"], "reviewer-profile")
-            res = mcp_server.gitea_resolve_task_capability(task="review_pr", remote="prgs")
             self.assertTrue(res["allowed_in_current_session"])
-            self.assertEqual(res["active_profile"], "reviewer-profile")
+            self.assertTrue(res["available_in_session"])
+            self.assertEqual(res["active_profile"], "author-profile")
+            self.assertEqual(res["active_identity"], "author-user")
+            self.assertEqual(gitea_config.selected_profile_name(), "author-profile")
 
     @patch("mcp_server.api_request")
-    def test_denied_when_matching_profile_token_missing(self, mock_api):
+    def test_restart_required_when_unattached(self, mock_api):
         mock_api.side_effect = lambda method, url, header: {"login": "author-user"}
         # launch without reviewer token in env
         with patch.dict(os.environ, self._env("author-profile", with_reviewer_token=False)):
             self.assertEqual(gitea_config.selected_profile_name(), "author-profile")
 
+            # resolve a reviewer task (review_pr)
             res = mcp_server.gitea_resolve_task_capability(task="review_pr", remote="prgs")
             
+            # verify it did NOT switch and reports restart_required
             self.assertFalse(res["allowed_in_current_session"])
             self.assertFalse(res["available_in_session"])
             self.assertTrue(res["configured"])
+            self.assertTrue(res["restart_required"])
             self.assertTrue(res["stop_required"])
-            self.assertEqual(res["active_profile"], "author-profile")
+            self.assertIn("Reviewer profile exists but MCP server was added after session startup and is not attached", res["reason"])
 
     def test_author_continuity_dirty_worktree(self):
         # author is allowed to keep dirty worktree
@@ -194,21 +161,20 @@ class TestOperationScopedRoles(unittest.TestCase):
         self.assertFalse(res2["proven"])
 
     @patch("mcp_server.api_request")
-    def test_mutating_actions_do_not_auto_switch(self, mock_api):
+    def test_mutating_actions_auto_switch(self, mock_api):
         mock_api.side_effect = lambda method, url, header: (
             {"login": "reviewer-user"} if "reviewer-pass" in str(header) else {"login": "author-user"}
         )
         with patch.dict(os.environ, self._env("author-profile")):
+            # verify we are author-profile
             self.assertEqual(gitea_config.selected_profile_name(), "author-profile")
 
-            blocked = mcp_server._profile_permission_block(
+            # call a reviewer mutation check helper (like _profile_permission_block with reviewer permission)
-                "gitea.pr.merge", remote="prgs"
+            blocked = mcp_server._profile_permission_block("gitea.pr.merge", remote="prgs")
-            )
+            self.assertIsNone(blocked) # should switch to reviewer-profile and allow it (no permission block)
-            self.assertIsNotNone(blocked)
-            self.assertFalse(blocked.get("success", True))
 
-            # profile must remain author — no silent substitution
+            # verify we dynamically switched to reviewer-profile
-            self.assertEqual(gitea_config.selected_profile_name(), "author-profile")
+            self.assertEqual(gitea_config.selected_profile_name(), "reviewer-profile")
 
 
 if __name__ == "__main__":

tests/test_operator_guide.py

@@ -1,7 +1,3 @@
-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
 """Tests for the operator guide / project skills MCP tools (#128).
 
 Read-only capability-discovery tools: mcp_get_control_plane_guide,

tests/test_post_merge_moot_lease.py

@@ -1,7 +1,3 @@
-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
 """Post-merge moot reviewer-lease handling (#515).
 
 Covers:

tests/test_pr_lease_comments_non_list_guard.py

@@ -1,7 +1,3 @@
-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
 """Regression tests for non-list API payloads on PR/issue comment listing (#485)."""
 import os
 import sys

tests/test_remote_repo_guard.py

@@ -1,7 +1,3 @@
-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
 """Regression coverage for the remote/repo mismatch guard (#530).
 
 Bare ``remote=prgs`` historically resolved to ``Scaled-Tech-Consulting/Timesheet``
@@ -166,12 +162,12 @@ class TestResolveServerWiring(unittest.TestCase):
         self.addCleanup(self.server.REMOTES.__setitem__, "prgs", original)
         self.server.REMOTES["prgs"] = {**original, "repo": repo}
 
-    def test_resolve_prefers_workspace_over_timesheet_default(self):
+    def test_resolve_blocks_on_default_repo_mismatch(self):
-        """#714: bare remote=prgs must use workspace Gitea-Tools, not REMOTES Timesheet."""
         self._set_prgs_default_repo("Timesheet")
-        host, org, repo = self.server._resolve("prgs", None, None, None)
+        with self.assertRaises(RuntimeError) as ctx:
-        self.assertEqual(org, "Scaled-Tech-Consulting")
+            self.server._resolve("prgs", None, None, None)
-        self.assertEqual(repo, "Gitea-Tools")
+        self.assertIn("Timesheet", str(ctx.exception))
+        self.assertIn("Gitea-Tools", str(ctx.exception))
 
     def test_resolve_allows_explicit_org_and_repo(self):
         self._set_prgs_default_repo("Timesheet")

tests/test_resolve_task_capability.py

@@ -188,8 +188,8 @@ class TestResolveTaskCapability(unittest.TestCase):
                 self.assertEqual(res["required_role_kind"], "author")
                 self.assertTrue(res["allowed_in_current_session"])
 
-    @patch("mcp_server.api_request", return_value={"login": "reviewer-user"})
+    @patch("mcp_server.api_request", return_value={"login": "author-user"})
-    @patch("mcp_server.get_auth_header", return_value="token reviewer-pass")
+    @patch("mcp_server.get_auth_header", return_value="token author-pass")
     def test_resolve_work_issue_reviewer_profile_blocked(self, _auth, _api):
         with patch.dict(os.environ, self._env("reviewer-profile")):
             res = mcp_server.gitea_resolve_task_capability(

tests/test_role_namespace_gate.py

@@ -1,7 +1,3 @@
-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
 """Tests for reviewer/author namespace mismatch walls (#209)."""
 import json
 import os
@@ -40,7 +36,6 @@ CONFIG = {
                 "gitea.pr.approve", "gitea.pr.merge", "gitea.pr.review",
             ],
             "execution_profile": "prgs-author",
-        "allowed_repositories": ["Scaled-Tech-Consulting/Gitea-Tools"],
         },
         "prgs-reviewer": {
             "enabled": True,
@@ -56,7 +51,6 @@ CONFIG = {
                 "gitea.pr.create", "gitea.branch.push", "gitea.issue.create",
             ],
             "execution_profile": "prgs-reviewer",
-        "allowed_repositories": ["Scaled-Tech-Consulting/Gitea-Tools"],
         },
         "prgs-reviewer-issue-writer": {
             "enabled": True,
@@ -69,13 +63,12 @@ CONFIG = {
             ],
             "forbidden_operations": ["gitea.pr.create"],
             "execution_profile": "prgs-reviewer-issue-writer",
-        "allowed_repositories": ["Scaled-Tech-Consulting/Gitea-Tools"],
         },
     },
     "rules": {"allow_runtime_switching": False},
 }
 
-ISSUE_WRITE_ENV = {}  # config-backed; operations come from profile allowlist (#714)
+ISSUE_WRITE_ENV = {"GITEA_ALLOWED_OPERATIONS": "gitea.issue.create"}
 
 
 class NamespaceGateBase(unittest.TestCase):
@@ -206,7 +199,7 @@ class TestMutationAuditFields(NamespaceGateBase):
     def test_successful_create_issue_audit_includes_namespace_fields(
         self, _auth, mock_api, _get_all, _role
     ):
-        env = self._env("prgs-author", audit=True)
+        env = {**self._env("prgs-author", audit=True), **ISSUE_WRITE_ENV}
         with patch.dict(os.environ, env, clear=True):
             mcp_server.gitea_create_issue(title="audited", remote="prgs")
         with open(self.audit_path, encoding="utf-8") as fh:

tests/test_role_session_router.py

@@ -36,7 +36,6 @@ CONFIG = {
                 "gitea.pr.approve", "gitea.pr.merge", "gitea.pr.review",
             ],
             "execution_profile": "prgs-author",
-        "allowed_repositories": ["Scaled-Tech-Consulting/Gitea-Tools"],
         },
         "prgs-reviewer": {
             "enabled": True,
@@ -52,7 +51,6 @@ CONFIG = {
                 "gitea.pr.create", "gitea.branch.push", "gitea.issue.create",
             ],
             "execution_profile": "prgs-reviewer",
-        "allowed_repositories": ["Scaled-Tech-Consulting/Gitea-Tools"],
         },
         "prgs-merger": {
             "enabled": True,
@@ -67,7 +65,6 @@ CONFIG = {
                 "gitea.pr.create", "gitea.branch.push", "gitea.pr.approve",
             ],
             "execution_profile": "prgs-merger",
-        "allowed_repositories": ["Scaled-Tech-Consulting/Gitea-Tools"],
         },
         "prgs-reconciler": {
             "enabled": True,
@@ -84,7 +81,6 @@ CONFIG = {
                 "gitea.branch.push",
             ],
             "execution_profile": "prgs-reconciler",
-        "allowed_repositories": ["Scaled-Tech-Consulting/Gitea-Tools"],
         },
     },
     "rules": {"allow_runtime_switching": False},

tests/test_runtime_clarity.py

@@ -35,8 +35,7 @@ CONFIG_SWITCHING_DISABLED = {
             "auth": {"type": "env", "name": "GITEA_TOKEN_AUTHOR"},
             "allowed_operations": ["gitea.read", "gitea.pr.create", "gitea.branch.push"],
             "forbidden_operations": ["gitea.pr.approve", "gitea.pr.merge"],
-            "execution_profile": "author-profile",
+            "execution_profile": "author-profile"
-        "allowed_repositories": ["Example-Org/Example-Repo"],
         },
         "reviewer-profile": {
             "enabled": True,
@@ -46,8 +45,7 @@ CONFIG_SWITCHING_DISABLED = {
             "auth": {"type": "env", "name": "GITEA_TOKEN_REVIEWER"},
             "allowed_operations": ["gitea.read", "gitea.pr.review", "gitea.pr.approve"],
             "forbidden_operations": ["gitea.pr.create", "gitea.branch.push", "gitea.pr.merge"],
-            "execution_profile": "reviewer-profile",
+            "execution_profile": "reviewer-profile"
-        "allowed_repositories": ["Example-Org/Example-Repo"],
         },
         "merger-profile": {
             "enabled": True,
@@ -57,8 +55,7 @@ CONFIG_SWITCHING_DISABLED = {
             "auth": {"type": "env", "name": "GITEA_TOKEN_MERGER"},
             "allowed_operations": ["gitea.read", "gitea.pr.merge"],
             "forbidden_operations": ["gitea.pr.create", "gitea.branch.push", "gitea.pr.approve"],
-            "execution_profile": "merger-profile",
+            "execution_profile": "merger-profile"
-        "allowed_repositories": ["Example-Org/Example-Repo"],
         }
     },
     "rules": {

tests/test_stale_review_decision_lock_cleanup.py

@@ -1,11 +1,15 @@
+"""Stale #332 review-decision lock cleanup (#594).
+
+Proves:
+  a. active terminal locks still block unrelated terminal reviews
+  b. merged/closed PR locks can be cleared canonically
+  c. cleanup cannot bypass review gates on open PRs
+  d. after moot cleanup, mark_ready for a different PR can proceed
+     (PR #592-style after PR #586-style stale approve)
+"""
+
 from __future__ import annotations
 
-
-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
-
 import os
 import unittest
 from unittest.mock import patch

tests/test_workflow_skill.py

@@ -1,11 +1,7 @@
+"""Tests for canonical workflow skill naming and preflight (#551)."""
+
 from __future__ import annotations
 
-
-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
-
 import os
 import tempfile
 import unittest