fix(session): bind workspace-verified repository scope at first bind (#714)

The independent review reproduced a real hole: the production first-bind path
never pinned repository/org, so the drift checks it feeds were skipped.

Root cause
----------
gitea_whoami, gitea_get_runtime_context, gitea_resolve_task_capability and
gitea_activate_profile all seeded the session context without repository or
org. First-bind-wins then made the mutation gate's later seed a no-op, and
assess_session_context only compares repository/org when the bound fields are
non-null. Result, reproduced in production order with the real REMOTES:

  after whoami:               repository=None org=None
  same-host repo=Other-Tools  -> NOT BLOCKED
  same-host org=Other-Org     -> NOT BLOCKED
  cross-host                  -> blocked (this part always worked)

Binding REMOTES defaults would not fix it: REMOTES['prgs'].repo is 'Timesheet'
(a default *target*, not an authorization scope, see #530), so pinning it fails
closed on every legitimate Gitea-Tools mutation. There was no trusted source
for repository scope, so this adds one.

Design
------
- New optional profile field `allowed_repositories`: canonical owner/repository
  slugs. An authorization boundary, never the binding itself. Config-only —
  no env var can widen or forge it.
- The session repository is derived from the verified, workspace-aligned git
  remote, never from caller-supplied org/repo, and validated against that
  allowlist. The session binds immutably to exactly one canonical slug; org is
  derived from it. A profile authorizing several repositories still binds to
  the single verified one.
- Every first-bind entry point now pins the same complete context. Activation
  rejects an unauthorized/unverifiable workspace. Mutations fail closed when
  repository/org is unverified, and a tool-level org/repo override that
  disagrees with the binding is rejected before the write.
- A mutation request can no longer establish, complete, or replace the binding
  (it previously seeded from `org or REMOTES[...]`, i.e. caller-controlled).
- Enforcement is opt-in per profile: a profile without the field keeps prior
  behaviour, so existing static-profile namespaces stay functional.

First-bind-wins, immutability, the RLock, profile isolation, host/identity
validation and the private pytest-only reset are unchanged.

gitea_auth.get_profile() built an explicit whitelist dict and silently dropped
`allowed_repositories`; the new integration tests caught that the scope was
never enforced through the real path.

Tests
-----
New tests/test_issue_714_production_first_bind.py drives the real entry points
in production order rather than constructing _SessionContext directly. Covers
complete first bind via each entry point, same-host repo/org drift, Timesheet
and other-repo/other-org rejection, unverified and unauthorized workspaces,
caller values unable to establish the binding, concurrent init selecting one
repository, and the PR #715 author path staying functional. Mutation-verified:
disabling trusted derivation, override validation, the scope check, or the
get_profile passthrough each fails these tests (9/8/4/5 failures).

No security assertion was weakened, removed, skipped, or rewritten.

Focused: 26 passed. Issue #714 total: 46 passed. Prior failing modules: 240
passed. Config/profile/remote modules: 130 passed. Full suite: 2739 passed,
6 skipped, 1 pre-existing warning, 161 subtests in 25.80s.

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

docs/gitea-execution-profiles.md

@@ -45,6 +45,7 @@ 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. |
@@ -57,6 +58,47 @@ 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,6 +41,7 @@
       "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

@@ -570,6 +570,10 @@ 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,6 +475,33 @@ 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:
@@ -549,6 +576,7 @@ 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.

gitea_mcp_server.py

@@ -1118,6 +1118,78 @@ import workflow_scope_guard  # noqa: E402  # #683 production scope / force-on gu
 import stable_branch_push_guard  # noqa: E402
 import remote_repo_guard  # noqa: E402
 import issue_claim_heartbeat  # noqa: E402
+import session_context_binding as session_ctx  # noqa: E402  # #714 immutable session context
+
+
+def _workspace_repository_slug(remote: str | None) -> str | None:
+    """Trusted ``owner/repository`` derived from the workspace git remote (#714).
+
+    This is the only source the session binding accepts for repository identity:
+    it comes from the verified, workspace-aligned checkout rather than from any
+    caller-supplied tool argument. ``REMOTES`` is deliberately not consulted —
+    its ``org``/``repo`` entries are default *targets*, not an authorization
+    scope (#530).
+    """
+    if not remote:
+        return None
+    parsed = remote_repo_guard.parse_org_repo_from_remote_url(
+        _local_git_remote_url(remote)
+    )
+    if not parsed:
+        return None
+    return session_ctx.format_repository_slug(parsed[0], parsed[1])
+
+
+def _trusted_session_repository(profile: dict, remote: str | None) -> dict:
+    """Resolve and authorize the session repository from the verified workspace.
+
+    Returns ``org`` / ``repository`` / ``reasons``. A profile's
+    ``allowed_repositories`` is an authorization boundary: the verified
+    workspace selects exactly one entry from it, and the session never binds to
+    the list itself nor switches between entries.
+    """
+    slug = _workspace_repository_slug(remote)
+    scope = session_ctx.assess_repository_scope(
+        workspace_slug=slug,
+        allowed=session_ctx.declared_allowed_repositories(profile),
+        profile_name=profile.get("profile_name"),
+    )
+    if scope.get("block"):
+        return {"org": None, "repository": None, "reasons": list(scope["reasons"])}
+    parts = session_ctx.parse_repository_slug(slug) if slug else None
+    if not parts:
+        return {"org": None, "repository": None, "reasons": []}
+    return {"org": parts[0], "repository": parts[1], "reasons": []}
+
+
+def _seed_session_context(
+    *,
+    profile: dict,
+    remote: str | None,
+    host: str | None,
+    identity: str | None,
+    source: str = "seed",
+) -> dict:
+    """Seed immutable session context once for the current process (#714).
+
+    Repository and organization are always derived from the verified workspace,
+    never from caller arguments, so every first-bind entry point (whoami,
+    runtime context, capability preflight, activation, mutation gate) pins the
+    same complete context.
+    """
+    expected = (profile.get("username") or "").strip() or None
+    trusted = _trusted_session_repository(profile, remote)
+    return session_ctx.seed_session_context_if_unbound(
+        profile_name=profile.get("profile_name") or "",
+        remote=remote,
+        host=host,
+        identity=identity,
+        repository=trusted["repository"],
+        org=trusted["org"],
+        role_kind=_profile_role_kind(profile),
+        expected_username=expected,
+        source=source,
+    )
 import issue_work_duplicate_gate  # noqa: E402
 import issue_workflow_labels  # noqa: E402
 import reviewer_pr_lease  # noqa: E402
@@ -1846,61 +1918,44 @@ def _authenticated_username(host: str):
     return user
 
 
-def _ensure_matching_profile(required_permission: str, required_role: str, remote: str | None, host: str | None = None) -> str | None:
-    """Check if the active profile is allowed to perform *required_permission*.
-    If not, automatically switch to the first matching usable configured profile.
+def _ensure_matching_profile(
+    required_permission: str,
+    required_role: str,
+    remote: str | None,
+    host: str | None = None,
+) -> str | None:
+    """Return the active profile name only when it is allowed for *permission*.
+
+    #714: never silently switch profiles (including cross-host substitution).
+    Capability resolution and mutation gates evaluate only the active profile
+    for the requested remote. Explicit ``gitea_activate_profile`` is the sole
+    in-process switch path.
     """
+    del required_role, host  # kept for call-site compatibility
     try:
         profile = get_profile()
     except Exception:
         return None
     active_profile = profile.get("profile_name")
+    # Cross-host denial: mdcps profile cannot serve a prgs remote (and vice versa).
+    config = None
+    try:
+        config = gitea_config.load_config()
+    except Exception:
+        config = None
+    contexts = (config or {}).get("contexts") if config else None
+    remote_ok = session_ctx.profile_allowed_for_remote(
+        profile, remote, REMOTES, contexts=contexts
+    )
+    if remote_ok.get("block"):
+        return None
     active_allowed = profile.get("allowed_operations") or []
     active_forbidden = profile.get("forbidden_operations") or []
-    allowed, _ = gitea_config.check_operation(required_permission, active_allowed, active_forbidden)
+    allowed, _ = gitea_config.check_operation(
+        required_permission, active_allowed, active_forbidden
+    )
     if allowed:
         return active_profile
-
-    # Try to find a matching usable profile in config
-    if gitea_config.is_runtime_switching_enabled():
-        config = gitea_config.load_config()
-        if config and "profiles" in config:
-            for p_name, p_data in config["profiles"].items():
-                p_allowed = p_data.get("allowed_operations") or []
-                p_forbidden = p_data.get("forbidden_operations") or []
-                p_allowed_n = []
-                for op in p_allowed:
-                    try:
-                        p_allowed_n.append(gitea_config.normalize_operation(op))
-                    except Exception:
-                        pass
-                p_forbidden_n = []
-                for op in p_forbidden:
-                    try:
-                        p_forbidden_n.append(gitea_config.normalize_operation(op))
-                    except Exception:
-                        pass
-                ok, _ = gitea_config.check_operation(required_permission, p_allowed_n, p_forbidden_n)
-                if ok:
-                    # Verify credentials/token are available
-                    try:
-                        tok = gitea_config.resolve_token(p_data)
-                        if tok:
-                            # Perform automatic switch
-                            gitea_config._active_profile_override = p_name
-                            h = host or (REMOTES.get(remote, {}).get("host") if remote in REMOTES else None)
-                            if h:
-                                _IDENTITY_CACHE.pop(h, None)
-                            username = _authenticated_username(h) if h else None
-                            # Update mutation authority
-                            global _MUTATION_AUTHORITY
-                            if _MUTATION_AUTHORITY is not None:
-                                _MUTATION_AUTHORITY["current_profile"] = p_name
-                                _MUTATION_AUTHORITY["current_identity"] = username
-                                _MUTATION_AUTHORITY["role_pivot_authorized"] = True
-                            return p_name
-                    except Exception:
-                        pass
     return None
 
 
@@ -1922,6 +1977,12 @@ def _audit(action: str, *, host, remote, result, org=None, repo=None,
         if mutation_task:
             ns_ctx = role_namespace_gate.mutation_audit_context(
                 mutation_task, profile, remote=remote, repository=repo)
+        # #714: always record the bound session context at mutation time.
+        session_audit = session_ctx.mutation_context_audit_fields()
+        if isinstance(request_metadata, dict):
+            request_metadata = {**request_metadata, **session_audit}
+        elif request_metadata is None:
+            request_metadata = dict(session_audit)
         event = gitea_audit.build_event(
             action=action,
             result=result,
@@ -2080,6 +2141,10 @@ def gitea_create_issue(
     blocked = _profile_permission_block(
         task_capability_map.required_permission("create_issue"),
         number=None,
+        remote=remote,
+        host=h,
+        org=o,
+        repo=r,
     )
     if blocked:
         return blocked
@@ -2520,6 +2585,10 @@ def gitea_create_pr(
     blocked = _profile_permission_block(
         task_capability_map.required_permission("create_pr"),
         number=None,
+        remote=remote,
+        host=host,
+        org=org,
+        repo=repo,
     )
     if blocked:
         return blocked
@@ -5356,7 +5425,7 @@ def gitea_commit_files(
         return blocked
     blocked = _profile_permission_block(
         task_capability_map.required_permission("commit_files"),
-        commit="", branch="",
+        commit="", branch="", remote=remote, host=host, org=org, repo=repo,
     )
     if blocked:
         return blocked
@@ -7499,48 +7568,13 @@ def _role_for_operation(op: str) -> str | None:
 
 
 def _try_auto_switch_for_operation(op: str, host: str | None = None) -> bool:
-    """Try to find a profile in config that allows op and has valid credentials.
+    """#714: automatic profile substitution is removed (fail closed).
 
-    If found, switch to it, clear identity cache, and return True.
-    Otherwise return False.
+    Always returns False. Callers must use explicit ``gitea_activate_profile``
+    to change roles; capability and mutation gates evaluate only the active
+    profile. Parameters retained for call-site compatibility.
     """
-    role = _role_for_operation(op)
-    if not role:
-        return False
-    if not gitea_config.is_runtime_switching_enabled():
-        return False
-    config = gitea_config.load_config()
-    if not config or "profiles" not in config:
-        return False
-    for p_name, p_data in config["profiles"].items():
-        # Role classification matching
-        p_role = p_data.get("role") or _role_kind(p_data.get("allowed_operations", []), p_data.get("forbidden_operations", []))
-        if p_role != role:
-            continue
-        p_allowed = p_data.get("allowed_operations") or []
-        p_forbidden = p_data.get("forbidden_operations") or []
-        p_allowed_n = []
-        for op_val in p_allowed:
-            try:
-                p_allowed_n.append(gitea_config.normalize_operation(op_val))
-            except Exception:
-                pass
-        p_forbidden_n = []
-        for op_val in p_forbidden:
-            try:
-                p_forbidden_n.append(gitea_config.normalize_operation(op_val))
-            except Exception:
-                pass
-        ok, _ = gitea_config.check_operation(op, p_allowed_n, p_forbidden_n)
-        if ok:
-            try:
-                tok = gitea_config.resolve_token(p_data)
-                if tok:
-                    gitea_config._active_profile_override = p_name
-                    _IDENTITY_CACHE.clear()
-                    return True
-            except Exception:
-                pass
+    del op, host
     return False
 
 
@@ -7607,6 +7641,127 @@ def _profile_operation_gate(op: str) -> list[str]:
     return [f"profile is not allowed to {op}"]
 
 
+def _session_context_mutation_block(
+    *,
+    remote: str | None,
+    host: str | None = None,
+    org: str | None = None,
+    repo: str | None = None,
+    username: str | None = None,
+    **extra_fields,
+) -> dict | None:
+    """#714: fail closed on session context / identity / cross-host drift."""
+    try:
+        profile = get_profile()
+    except Exception as exc:
+        return {
+            "success": False,
+            "performed": False,
+            "reasons": [
+                f"profile could not be resolved (fail closed): {_redact(str(exc))}"
+            ],
+            "blocker_kind": "session_context",
+            **extra_fields,
+        }
+    profile_name = profile.get("profile_name")
+    expected = (profile.get("username") or "").strip() or None
+    remote_config = REMOTES.get(remote or "", {}) if remote else {}
+    h = host or remote_config.get("host")
+    # #714: repository identity comes from the verified workspace only. The
+    # caller-supplied org/repo are a *request target* to be checked against the
+    # binding — they must never establish, complete, or replace it.
+    trusted = _trusted_session_repository(profile, remote)
+    resolved_org = trusted["org"]
+    resolved_repo = trusted["repository"]
+    identity = username
+    # Legacy env-only profiles do not declare an expected identity. Their
+    # first mutation still pins remote/host/repository, while existing audit
+    # paths resolve identity at the actual write. Configured identities are
+    # always verified here before a mutation can proceed.
+    if identity is None and expected and h:
+        try:
+            identity = _authenticated_username(h)
+        except Exception:
+            identity = None
+
+    config = None
+    try:
+        config = gitea_config.load_config()
+    except Exception:
+        config = None
+    contexts = (config or {}).get("contexts") if config else None
+    remote_gate = session_ctx.profile_allowed_for_remote(
+        profile, remote, REMOTES, contexts=contexts
+    )
+    reasons: list[str] = list(remote_gate.get("reasons") or [])
+
+    id_gate = session_ctx.assess_identity_match(
+        authenticated=identity, expected_username=expected
+    )
+    reasons.extend(id_gate.get("reasons") or [])
+
+    # Repository scope denial (unauthorized workspace) fails closed here.
+    reasons.extend(trusted.get("reasons") or [])
+
+    # Seed if unbound so subsequent tools share one context; then assess.
+    _seed_session_context(
+        profile=profile,
+        remote=remote,
+        host=h,
+        identity=identity,
+        source="mutation-gate-seed",
+    )
+    ctx_gate = session_ctx.assess_session_context(
+        profile_name=profile_name,
+        remote=remote,
+        host=h,
+        identity=identity,
+        repository=resolved_repo,
+        org=resolved_org,
+        expected_username=expected,
+        require_bound=True,
+        require_complete=bool(
+            session_ctx.declared_allowed_repositories(profile)
+        ),
+    )
+    reasons.extend(ctx_gate.get("reasons") or [])
+
+    # The request may only be checked against the immutable binding (#714).
+    bound_ctx = session_ctx.get_session_context() or {}
+    override_gate = session_ctx.assess_repository_override(
+        requested_org=org,
+        requested_repo=repo,
+        bound_org=bound_ctx.get("org"),
+        bound_repo=bound_ctx.get("repository"),
+    )
+    reasons.extend(override_gate.get("reasons") or [])
+    # De-dupe while preserving order
+    seen: set[str] = set()
+    uniq: list[str] = []
+    for r in reasons:
+        if r not in seen:
+            seen.add(r)
+            uniq.append(r)
+    if not uniq:
+        return None
+    blocked = {
+        "success": False,
+        "performed": False,
+        "reasons": uniq,
+        "blocker_kind": "session_context",
+        "session_context_audit": session_ctx.mutation_context_audit_fields(),
+        "exact_next_action": (
+            "BLOCKED + DIAGNOSE: session profile/remote/host/identity context "
+            "is inconsistent or unauthorized for this mutation. Re-pin the "
+            "correct profile with gitea_activate_profile, re-verify with "
+            "gitea_whoami for the intended remote, and do not use cross-host "
+            "fallback profiles."
+        ),
+    }
+    blocked.update(extra_fields)
+    return blocked
+
+
 def _profile_permission_block(required_operation: str, **extra_fields) -> dict | None:
     """Structured permission denial for gated tools (#69, #142).
 
@@ -7616,25 +7771,37 @@ def _profile_permission_block(required_operation: str, **extra_fields) -> dict |
     req_role = "reviewer" if any(required_operation.startswith(p) for p in (
         "gitea.pr.approve", "gitea.pr.merge", "gitea.pr.request_changes", "gitea.pr.review"
     )) else "author"
+    # #714: evaluate active profile only — never auto-switch.
     _ensure_matching_profile(required_operation, req_role, extra_fields.get("remote"))
 
     reasons = _profile_operation_gate(required_operation)
-    if not reasons:
-        return None
-    blocked = {
-        "success": False,
-        "performed": False,
-        "reasons": reasons,
-        "permission_report": _permission_block_report(required_operation),
-    }
-    blocked.update(extra_fields)
-    return blocked
+    if reasons:
+        blocked = {
+            "success": False,
+            "performed": False,
+            "reasons": reasons,
+            "permission_report": _permission_block_report(required_operation),
+            "session_context_audit": session_ctx.mutation_context_audit_fields(),
+        }
+        blocked.update(extra_fields)
+        return blocked
+
+    return _session_context_mutation_block(
+        remote=extra_fields.get("remote"),
+        host=extra_fields.get("host"),
+        org=extra_fields.get("org"),
+        repo=extra_fields.get("repo"),
+        number=extra_fields.get("number"),
+        issue_number=extra_fields.get("issue_number"),
+        pr_number=extra_fields.get("pr_number"),
+    )
 
 
 def _namespace_mutation_block(mutation_task: str, **extra_fields) -> dict | None:
     """Reviewer/author namespace alignment gate (#209)."""
     required_permission = task_capability_map.required_permission(mutation_task)
     required_role = task_capability_map.required_role(mutation_task)
+    # #714: evaluate active profile only — never auto-switch.
     _ensure_matching_profile(required_permission, required_role, extra_fields.get("remote"))
 
     try:
@@ -9621,9 +9788,22 @@ def gitea_whoami(
     # name is the addressing surface; 'server' appears only under the
     # GITEA_MCP_REVEAL_ENDPOINTS admin opt-in.
     profile = get_profile()
+    identity = data.get("login")
+    expected_username = (profile.get("username") or "").strip() or None
+    # #714: seed immutable session context so later tools share one pin.
+    _seed_session_context(
+        profile=profile,
+        remote=remote,
+        host=h,
+        identity=identity,
+        source="gitea_whoami",
+    )
+    id_match = session_ctx.assess_identity_match(
+        authenticated=identity, expected_username=expected_username
+    )
     result = {
         "authenticated": True,
-        "username": data.get("login"),
+        "username": identity,
         "display_name": data.get("full_name") or None,
         "user_id": data.get("id"),
         "email": data.get("email") or None,
@@ -9640,7 +9820,11 @@ def gitea_whoami(
             "execution_profile": profile.get("execution_profile"),
             "audit_label": profile.get("audit_label"),
             "auth_source_type": profile.get("auth_source_type"),
+            "expected_username": expected_username,
         },
+        "session_context_audit": session_ctx.mutation_context_audit_fields(),
+        "identity_match": not id_match.get("block"),
+        "identity_match_reasons": id_match.get("reasons") or [],
     }
     if _reveal_endpoints():
         result["server"] = gitea_url(h, "").rstrip("/")
@@ -9766,14 +9950,32 @@ _RUNTIME_CAPABILITY_TASKS = (
 def _matching_configured_profiles(
     config: dict | None,
     required_permission: str,
+    remote: str | None = None,
 ) -> list[str]:
-    """Profile names that allow *required_permission* (redacted metadata only)."""
+    """Profile names that allow *required_permission* (redacted metadata only).
+
+    #714: when *remote* is provided, only profiles bound to that remote's host
+    are returned — a dadeschools request never lists prgs profiles.
+    """
     if not config or "profiles" not in config:
         return []
+    contexts = config.get("contexts") or {}
+    remote_ok_names: set[str] | None = None
+    if remote:
+        remote_ok_names = set(
+            session_ctx.filter_profiles_for_remote(config, remote, REMOTES)
+        )
     matches: list[str] = []
     for p_name, p_data in config["profiles"].items():
         if not p_data.get("enabled", True):
             continue
+        if remote_ok_names is not None and p_name not in remote_ok_names:
+            continue
+        # Also require host/context match when remote given (defensive).
+        if remote and not session_ctx.profile_matches_remote(
+            p_data, remote, REMOTES, contexts=contexts
+        ):
+            continue
         p_allowed = p_data.get("allowed_operations") or []
         p_forbidden = p_data.get("forbidden_operations") or []
         p_allowed_n = []
@@ -9800,8 +10002,9 @@ def _build_runtime_task_capabilities(
     allowed: list[str],
     forbidden: list[str],
     config: dict | None,
+    remote: str | None = None,
 ) -> dict:
-    """Per-task capability summary for role-aware runtime context (#139)."""
+    """Per-task capability summary for role-aware runtime context (#139, #714)."""
     task_entries = []
     flags: dict[str, bool] = {}
     flag_keys = {
@@ -9825,7 +10028,7 @@ def _build_runtime_task_capabilities(
             "required_role_kind": task_capability_map.required_role(task),
             "allowed_in_current_session": allowed_here,
             "matching_configured_profiles": _matching_configured_profiles(
-                config, permission
+                config, permission, remote=remote
             ),
         }
         task_entries.append(entry)
@@ -10277,8 +10480,9 @@ def gitea_get_runtime_context(
             "or ask the operator to update GITEA_MCP_PROFILE to a reviewer profile."
         )
 
+    # #714: filter matching profiles by requested remote when building capability summary
     session_capabilities = _build_runtime_task_capabilities(
-        allowed, forbidden, config
+        allowed, forbidden, config, remote=remote if remote in REMOTES else None
     )
 
     preflight = assess_preflight_status(worktree_path)
@@ -10289,6 +10493,16 @@ def gitea_get_runtime_context(
             f"Blocked: {'; '.join(preflight['preflight_block_reasons'])}"
         )
 
+    expected_username = (profile.get("username") or "").strip() or None
+    h_rt = host or (REMOTES.get(remote, {}).get("host") if remote in REMOTES else None)
+    _seed_session_context(
+        profile=profile,
+        remote=remote if remote in REMOTES else None,
+        host=h_rt,
+        identity=username,
+        source="gitea_get_runtime_context",
+    )
+
     result = {
         "active_profile": profile["profile_name"],
         "authenticated_username": username,
@@ -10299,6 +10513,8 @@ def gitea_get_runtime_context(
         "forbidden_operations": forbidden,
         "runtime_switching_supported": switching,
         "profile_mode": profile_mode,
+        "session_context_audit": session_ctx.mutation_context_audit_fields(),
+        "auto_profile_substitution": False,
         "review_merge_allowed": review_merge_allowed,
         "review_merge_blocked_reasons": blocked_reasons,
         "suggested_fix": suggested_fix,
@@ -10660,8 +10876,10 @@ def gitea_activate_profile(
         _IDENTITY_CACHE.pop(h, None)
 
     # 4. Resolve fresh identity
-    after_profile = get_profile()["profile_name"]
+    after_profile_data = get_profile()
+    after_profile = after_profile_data["profile_name"]
     after_identity = _authenticated_username(h) if h else None
+    expected_username = (after_profile_data.get("username") or "").strip() or None
 
     # 4.5 Record the authorized pivot in the in-process mutation authority
     # and keep the session profile lock in sync — this is the ONLY path that
@@ -10676,15 +10894,52 @@ def gitea_activate_profile(
             "from_identity": before_identity,
             "to_identity": after_identity,
         }
+        _MUTATION_AUTHORITY["remote"] = remote
     if os.environ.get(SESSION_PROFILE_LOCK_ENV) and after_profile:
         os.environ[SESSION_PROFILE_LOCK_ENV] = after_profile
 
+    # 4.6 #714: re-bind immutable session context after explicit activation.
+    # Activation is the sanctioned logical-session boundary, so it establishes
+    # the complete binding — including the workspace-verified repository. An
+    # unauthorized workspace fails closed here rather than at first mutation.
+    activated_trusted = _trusted_session_repository(after_profile_data, remote)
+    if activated_trusted.get("reasons"):
+        return {
+            "success": False,
+            "performed": False,
+            "reasons": list(activated_trusted["reasons"]),
+            "blocker_kind": "repository_scope",
+            "exact_next_action": (
+                "BLOCKED + DIAGNOSE: the verified workspace repository is not "
+                "authorized by this profile's allowed_repositories. Run from a "
+                "workspace whose git remote matches an authorized "
+                "owner/repository, or have the operator provision the profile "
+                "scope. Do not pass org/repo to bypass this gate."
+            ),
+        }
+    session_ctx.bind_session_context(
+        profile_name=after_profile or profile_name,
+        remote=remote,
+        host=h,
+        identity=after_identity,
+        repository=activated_trusted["repository"],
+        org=activated_trusted["org"],
+        role_kind=_profile_role_kind(after_profile_data),
+        expected_username=expected_username,
+        source="gitea_activate_profile",
+    )
+
     # 5. Audit the switch if auditing is on
     _audit(
         "activate_profile",
         host=h,
         remote=remote,
-        result={"success": True, "before": before_profile, "after": after_profile},
+        result={
+            "success": True,
+            "before": before_profile,
+            "after": after_profile,
+            "session_context": session_ctx.mutation_context_audit_fields(),
+        },
         username=after_identity,
     )
 
@@ -10695,6 +10950,8 @@ def gitea_activate_profile(
         "before_identity": before_identity,
         "after_profile": after_profile,
         "after_identity": after_identity,
+        "session_context_audit": session_ctx.mutation_context_audit_fields(),
+        "auto_profile_substitution": False,
     }
 
 
@@ -11795,21 +12052,44 @@ def gitea_resolve_task_capability(
 
     record_preflight_check("capability", required_role, resolved_task=task)
 
-    # Try automatic dispatch switching
-    _ensure_matching_profile(required_permission, required_role, remote, host)
-
+    # #714: never auto-switch profiles during capability resolution.
     profile = get_profile()
     config = gitea_config.load_config()
+    contexts = (config or {}).get("contexts") if config else None
 
     h = host or (REMOTES.get(remote, {}).get("host") if remote in REMOTES else None)
     username = _authenticated_username(h) if h else None
+    expected_username = (profile.get("username") or "").strip() or None
+
+    # Seed / re-check immutable session context for consistent multi-tool reads.
+    _seed_session_context(
+        profile=profile,
+        remote=remote,
+        host=h,
+        identity=username,
+        source="resolve_task_capability",
+    )
+    ctx_assess = session_ctx.assess_session_context(
+        profile_name=profile.get("profile_name"),
+        remote=remote,
+        host=h,
+        identity=username,
+        expected_username=expected_username,
+        require_bound=False,
+    )
+    id_assess = session_ctx.assess_identity_match(
+        authenticated=username, expected_username=expected_username
+    )
+    remote_assess = session_ctx.profile_allowed_for_remote(
+        profile, remote, REMOTES, contexts=contexts
+    )
 
     # Load active permissions
     active_allowed = profile.get("allowed_operations") or []
     active_forbidden = profile.get("forbidden_operations") or []
     active_role_kind = _profile_role_kind(profile)
 
-    # Check if allowed in current session
+    # Check if allowed in current session (active profile only — #714)
     permission_allowed_in_current_session, _ = gitea_config.check_operation(
         required_permission, active_allowed, active_forbidden
     )
@@ -11824,8 +12104,15 @@ def gitea_resolve_task_capability(
             f"{required_role} task '{task}' even if nearby permissions are "
             "present (fail closed)."
         )
+    cross_host_block = bool(remote_assess.get("block"))
+    identity_block = bool(id_assess.get("block"))
+    drift_block = bool(ctx_assess.get("block"))
     allowed_in_current_session = (
-        permission_allowed_in_current_session and role_matches_current_session
+        permission_allowed_in_current_session
+        and role_matches_current_session
+        and not cross_host_block
+        and not identity_block
+        and not drift_block
     )
 
     switching = gitea_config.is_runtime_switching_enabled()
@@ -11834,33 +12121,19 @@ def gitea_resolve_task_capability(
     restart_required = False
     reason_msg = None
 
-    # Find matching configured profiles
-    matching_profiles = []
-    if config and "profiles" in config:
-        for p_name, p_data in config["profiles"].items():
-            p_allowed = p_data.get("allowed_operations") or []
-            p_forbidden = p_data.get("forbidden_operations") or []
-            p_allowed_n = []
-            for op in p_allowed:
-                try:
-                    p_allowed_n.append(gitea_config.normalize_operation(op))
-                except Exception:
-                    pass
-            p_forbidden_n = []
-            for op in p_forbidden:
-                try:
-                    p_forbidden_n.append(gitea_config.normalize_operation(op))
-                except Exception:
-                    pass
-            ok, _ = gitea_config.check_operation(required_permission, p_allowed_n, p_forbidden_n)
+    # Matching profiles: same remote/host only (#714) — advisory, never activated.
+    matching_profiles = _matching_configured_profiles(
+        config, required_permission, remote=remote
+    )
+    # Role filter for exclusive tasks
+    if config and "profiles" in config and task in role_exclusive_tasks:
+        filtered = []
+        for p_name in matching_profiles:
+            p_data = (config.get("profiles") or {}).get(p_name) or {}
             p_role = (p_data.get("role") or "").strip()
-            p_role_ok = (
-                task not in role_exclusive_tasks
-                or not p_role
-                or p_role == required_role
-            )
-            if ok and p_role_ok:
-                matching_profiles.append(p_name)
+            if not p_role or p_role == required_role:
+                filtered.append(p_name)
+        matching_profiles = filtered
 
     configured = len(matching_profiles) > 0
     available_in_session = allowed_in_current_session
@@ -11876,16 +12149,35 @@ def gitea_resolve_task_capability(
             )
 
     if not allowed_in_current_session:
-        if configured and switching:
-            restart_required = True
+        deny_parts: list[str] = []
+        if cross_host_block:
+            deny_parts.extend(remote_assess.get("reasons") or [])
+        if identity_block:
+            deny_parts.extend(id_assess.get("reasons") or [])
+        if drift_block:
+            deny_parts.extend(ctx_assess.get("reasons") or [])
+        if not permission_allowed_in_current_session and not deny_parts:
+            deny_parts.append(
+                f"Active profile '{profile.get('profile_name')}' is not allowed "
+                f"to '{required_permission}' (no substitute profile activated; fail closed)."
+            )
+        if role_mismatch_reason:
+            deny_parts.append(role_mismatch_reason)
+        if deny_parts:
+            reason_msg = "; ".join(deny_parts)
+        elif configured and switching:
+            # Same-remote profile exists but is not the active one — explicit switch only.
+            restart_required = False
             available_in_session = False
             reason_msg = (
-                f"{required_role.capitalize()} profile exists but MCP server "
-                "was added after session startup and is not attached."
+                f"Active profile cannot perform '{required_permission}'. "
+                f"Same-remote matching profiles (advisory only, not activated): "
+                f"{matching_profiles}."
             )
         elif not configured:
             reason_msg = (
-                f"No profile configured with permission '{required_permission}'."
+                f"No same-remote profile configured with permission "
+                f"'{required_permission}' for remote '{remote}'."
             )
         elif role_mismatch_reason:
             reason_msg = role_mismatch_reason
@@ -11893,9 +12185,22 @@ def gitea_resolve_task_capability(
     next_safe_action = "None; ready for operations."
 
     if not allowed_in_current_session:
-        if switching:
+        if cross_host_block or identity_block or drift_block:
             different_namespace_required = False
-            next_safe_action = f"Switch to a profile that has the required permission by calling gitea_activate_profile (matching configured profiles: {matching_profiles})."
+            next_safe_action = (
+                "BLOCKED + DIAGNOSE: session context/host/identity is inconsistent "
+                "or the active profile cannot serve this remote. Re-pin the correct "
+                "profile with gitea_activate_profile for the intended remote, verify "
+                "with gitea_whoami, and do not use cross-host profile substitution."
+            )
+        elif switching:
+            different_namespace_required = False
+            next_safe_action = (
+                f"Switch explicitly with gitea_activate_profile to a same-remote "
+                f"profile that allows '{required_permission}' "
+                f"(matching configured profiles: {matching_profiles}). "
+                "Capability resolution never auto-substitutes profiles (#714)."
+            )
         else:
             different_namespace_required = True
             if required_role == "reviewer":
@@ -11980,6 +12285,13 @@ def gitea_resolve_task_capability(
         "runtime_switching_supported": switching,
         "different_mcp_namespace_required": different_namespace_required,
         "exact_safe_next_action": next_safe_action,
+        # #714: immutable session context proof (must match whoami / runtime).
+        "requested_remote": remote,
+        "resolved_host": h,
+        "session_context_audit": session_ctx.mutation_context_audit_fields(),
+        "profile_remote_compatible": not cross_host_block,
+        "identity_match": not identity_block,
+        "auto_profile_substitution": False,
     }
     if reason_msg:
         result["reason"] = reason_msg

session_context_binding.py

@@ -0,0 +1,550 @@
+"""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) -> 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`).  Returns ``[]`` when the profile declares
+    no scope — enforcement is opt-in per profile so existing static-profile
+    namespaces stay functional until an operator provisions the field.
+    """
+    if not profile:
+        return []
+    raw = profile.get("allowed_repositories")
+    if not isinstance(raw, (list, tuple)):
+        return []
+    slugs: list[str] = []
+    for entry in raw:
+        if not isinstance(entry, str):
+            continue
+        parsed = parse_repository_slug(entry)
+        if parsed:
+            slugs.append(f"{parsed[0]}/{parsed[1]}")
+    return slugs
+
+
+def assess_repository_scope(
+    *,
+    workspace_slug: str | None,
+    allowed: list[str] | None,
+    profile_name: str | None = None,
+) -> 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.
+    """
+    scope = list(allowed or [])
+    reasons: list[str] = []
+    if not scope:
+        return {
+            "proven": True,
+            "block": False,
+            "reasons": reasons,
+            "scope_enforced": False,
+            "workspace_slug": workspace_slug,
+            "allowed_repositories": [],
+        }
+    name = profile_name or "(active profile)"
+    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),
+    }

tests/conftest.py

@@ -26,6 +26,14 @@ 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",
@@ -80,9 +88,15 @@ def _reset_mutation_authority(monkeypatch):
     try:
         import mcp_server
     except Exception:
-        _state_tmp.cleanup()
-        yield
+        try:
+            yield
+        finally:
+            session_ctx._reset_session_context_for_testing()
+            _state_tmp.cleanup()
         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)
@@ -113,19 +127,23 @@ def _reset_mutation_authority(monkeypatch):
         capability_stop_terminal.clear()
     except Exception:
         pass
-    yield
     try:
-        import capability_stop_terminal
-        capability_stop_terminal.clear()
-    except Exception:
-        pass
-    try:
-        import review_workflow_load
-        review_workflow_load._REVIEW_WORKFLOW_LOAD = None
-    except Exception:
-        pass
-    try:
-        mcp_server._REVIEW_DECISION_LOCK = None
-    except Exception:
-        pass
-    _state_tmp.cleanup()
+        yield
+    finally:
+        try:
+            import capability_stop_terminal
+            capability_stop_terminal.clear()
+        except Exception:
+            pass
+        try:
+            import review_workflow_load
+            review_workflow_load._REVIEW_WORKFLOW_LOAD = None
+        except Exception:
+            pass
+        try:
+            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()

tests/test_commit_payloads.py

@@ -227,6 +227,7 @@ 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)
 
@@ -248,6 +249,7 @@ 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(
@@ -266,6 +268,7 @@ 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_issue_714_production_first_bind.py

@@ -0,0 +1,448 @@
+"""#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"
+            )
+            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(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_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(self):
+        self.assertEqual(
+            session_ctx.declared_allowed_repositories(
+                {"allowed_repositories": ["not-a-slug", 17, None, WORKSPACE_SLUG]}
+            ),
+            [WORKSPACE_SLUG],
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_issue_714_session_context_immutability.py

@@ -0,0 +1,596 @@
+"""#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",
+            ],
+            "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",
+            ],
+            "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",
+            ],
+            "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()
+        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._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_operation_scoped_roles.py

@@ -1,4 +1,9 @@
-"""Tests for operation-scoped role selection and automatic dispatch switching (#228)."""
+"""Tests for operation-scoped role selection (#228, updated by #714).
+
+#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
@@ -15,6 +20,7 @@ from reviewer_worktree import assess_author_worktree_continuity
 
 CONFIG_TEST = {
     "version": 2,
+    "allow_runtime_switching": True,
     "contexts": {
         "ctx": {
             "enabled": True,
@@ -30,6 +36,7 @@ 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"],
@@ -40,6 +47,7 @@ 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"],
@@ -85,62 +93,71 @@ class TestOperationScopedRoles(unittest.TestCase):
         return env
 
     @patch("mcp_server.api_request")
-    def test_auto_switch_to_reviewer(self, mock_api):
-        # mock identity resolution to return username matching profile
+    def test_no_auto_switch_to_reviewer(self, mock_api):
+        # #714: capability resolution must not silently switch author → reviewer
         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")
-            
-            # verify it automatically switched to reviewer-profile
-            self.assertTrue(res["allowed_in_current_session"])
-            self.assertTrue(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.assertFalse(res["allowed_in_current_session"])
+            self.assertFalse(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")
+            self.assertIn("reviewer-profile", res["matching_configured_profile"])
+            self.assertFalse(res.get("auto_profile_substitution", True))
 
     @patch("mcp_server.api_request")
-    def test_auto_switch_to_author(self, mock_api):
+    def test_no_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")
-            
-            # verify it automatically switched to author-profile
-            self.assertTrue(res["allowed_in_current_session"])
-            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")
+
+            self.assertFalse(res["allowed_in_current_session"])
+            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_restart_required_when_unattached(self, mock_api):
+    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")
+
+    @patch("mcp_server.api_request")
+    def test_denied_when_matching_profile_token_missing(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.assertIn("Reviewer profile exists but MCP server was added after session startup and is not attached", res["reason"])
+            self.assertEqual(res["active_profile"], "author-profile")
 
     def test_author_continuity_dirty_worktree(self):
         # author is allowed to keep dirty worktree
@@ -161,20 +178,21 @@ class TestOperationScopedRoles(unittest.TestCase):
         self.assertFalse(res2["proven"])
 
     @patch("mcp_server.api_request")
-    def test_mutating_actions_auto_switch(self, mock_api):
+    def test_mutating_actions_do_not_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")
 
-            # call a reviewer mutation check helper (like _profile_permission_block with reviewer permission)
-            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)
+            blocked = mcp_server._profile_permission_block(
+                "gitea.pr.merge", remote="prgs"
+            )
+            self.assertIsNotNone(blocked)
+            self.assertFalse(blocked.get("success", True))
 
-            # verify we dynamically switched to reviewer-profile
-            self.assertEqual(gitea_config.selected_profile_name(), "reviewer-profile")
+            # profile must remain author — no silent substitution
+            self.assertEqual(gitea_config.selected_profile_name(), "author-profile")
 
 
 if __name__ == "__main__":

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": "author-user"})
-    @patch("mcp_server.get_auth_header", return_value="token author-pass")
+    @patch("mcp_server.api_request", return_value={"login": "reviewer-user"})
+    @patch("mcp_server.get_auth_header", return_value="token reviewer-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(