feat(mcp): enforce strict cross-project mutation boundaries (Closes #707)

docs/mcp-namespace-eof-recovery.md

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

docs/mcp-namespace-health.md

@@ -86,26 +86,3 @@ When a namespace returns EOF, follow
 
 When blocked, repair the IDE namespace and re-record a healthy
 `client_namespace` assessment before retrying the mutation.
-
-## Connected vs Attached Tool Surface (#708)
-
-MCP servers can report **Connected** at the CLI / host inventory layer while the **active LLM session exposes none of their tool namespaces**.
-
-### Core principle
-
-* **Connected status at host layer ≠ attached tools in active session.**
-* Required preflight proof is **live tool visibility + `gitea_whoami` call** through the target namespace, not host `Connected` status alone.
-* When servers report Connected but namespaces are absent from attached tools, classify as `mcp_connected_namespaces_missing`.
-
-### Forbidden unsafe fallbacks
-
-When `mcp_connected_namespaces_missing` is detected, workflows must **fail closed** and must **never** encourage or perform:
-
-* direct imports of MCP server Python modules
-* CLI or raw Gitea API mutations as a substitute for native tools
-* profile hopping to another MCP profile/namespace to bypass the empty session
-* session-state overrides or hand-edited session/ledger files
-* process kills (`pkill`), config mtime touches, or `.env` edits
-
-Only sanctioned recovery: **client reconnect path**, followed by full preflight (`whoami` → capability resolve → task).
-

gitea_config.py

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

gitea_mcp_server.py

@@ -3325,10 +3325,16 @@ def _effective_remote(remote: str) -> str:
     return remote
 
 
-def _resolve(remote: str, host: str | None, org: str | None, repo: str | None):
+def _resolve(
+    remote: str,
+    host: str | None,
+    org: str | None,
+    repo: str | None,
+    for_mutation: bool = False,
+):
     """Resolve remote + overrides to (host, org, repo).
 
-    #714 / #530: when the caller omits org and/or repo, prefer the
+    #714 / #530 / #707: when the caller omits org and/or repo, prefer the
     workspace-aligned git remote over ``REMOTES`` defaults (e.g. bare
     ``remote=prgs`` must not force ``Timesheet`` when the checkout is
     ``Gitea-Tools``). Explicit caller org/repo always win and are validated
@@ -3414,6 +3420,7 @@ def _resolve(remote: str, host: str | None, org: str | None, repo: str | None):
         # Workspace-filled sides are intentional alignment for #530.
         org_explicit=org_explicit or filled_org,
         repo_explicit=repo_explicit or filled_repo,
+        for_mutation=for_mutation,
     )
     return (resolved_host, resolved_org, resolved_repo)
 
@@ -3475,8 +3482,9 @@ def _enforce_remote_repo_guard(
     *,
     org_explicit: bool,
     repo_explicit: bool,
+    for_mutation: bool = False,
 ) -> None:
-    """Fail closed on a remote/repo mismatch vs. the local git remote (#530).
+    """Fail closed on a remote/repo mismatch vs. the local git remote (#530/#707).
 
     Best-effort: bypassed under pytest unless ``GITEA_FORCE_REMOTE_REPO_CHECK`` is
     set, so the unit suite (which calls tools with bare remotes against mocked APIs)
@@ -3488,6 +3496,12 @@ def _enforce_remote_repo_guard(
     ):
         return
     local_remote_url = _local_git_remote_url(remote)
+    primary_org = None
+    primary_repo = None
+    ctx = session_ctx.get_session_context()
+    if ctx:
+        primary_org = ctx.get("org")
+        primary_repo = ctx.get("repository")
     assessment = remote_repo_guard.assess_remote_repo_match(
         remote=remote,
         resolved_org=resolved_org,
@@ -3495,6 +3509,9 @@ def _enforce_remote_repo_guard(
         local_remote_url=local_remote_url,
         org_explicit=org_explicit,
         repo_explicit=repo_explicit,
+        for_mutation=for_mutation,
+        primary_org=primary_org,
+        primary_repo=primary_repo,
     )
     if assessment["block"]:
         raise RuntimeError(remote_repo_guard.format_remote_repo_guard_error(assessment))
@@ -4063,7 +4080,7 @@ def gitea_lock_issue(
     resolved_worktree = issue_lock_worktree.resolve_author_worktree_path(
         worktree_path, _canonical_local_git_root()
     )
-    h, o, r = _resolve(remote, host, org, repo)
+    h, o, r = _resolve(remote, host, org, repo, for_mutation=True)
     existing_issue_lock = _load_existing_issue_lock(
         remote=remote, org=o, repo=r, issue_number=issue_number
     )
@@ -5239,7 +5256,7 @@ def gitea_create_pr(
             org=org,
             repo=repo,
         )
-    h, o, r = _resolve(remote, host, org, repo)
+    h, o, r = _resolve(remote, host, org, repo, for_mutation=True)
 
     # ── Issue Lock Validation (Issue #194 / #196 / #443) ──
     lock_data = _resolve_issue_lock_for_pr(remote=remote, org=o, repo=r, head=head)
@@ -9560,7 +9577,7 @@ def gitea_edit_pr(
                 required_permission="gitea.pr.close",
             )
 
-    h, o, r = _resolve(remote, host, org, repo)
+    h, o, r = _resolve(remote, host, org, repo, for_mutation=True)
     auth = _auth(h)
     url = f"{repo_api_url(h, o, r)}/pulls/{pr_number}"
 
@@ -9844,7 +9861,7 @@ def gitea_commit_files(
     verify_preflight_purity(remote=remote, worktree_path=worktree_path, task="commit_files", org=org, repo=repo)
     processed_files, source_proofs = _prepare_commit_payload_files(files)
 
-    h, o, r = _resolve(remote, host, org, repo)
+    h, o, r = _resolve(remote, host, org, repo, for_mutation=True)
     auth = _auth(h)
     url = f"{repo_api_url(h, o, r)}/contents"
 
@@ -9980,7 +9997,7 @@ def gitea_publish_unpublished_issue_branch(
         repo=repo,
     )
 
-    h, o, r = _resolve(remote, host, org, repo)
+    h, o, r = _resolve(remote, host, org, repo, for_mutation=True)
     git_remote = (git_remote_name or remote or "").strip()
 
     existing_lock = issue_lock_store.load_issue_lock(
@@ -10183,7 +10200,7 @@ def gitea_bootstrap_author_issue_worktree(
         repo=repo,
     )
 
-    h, o, r = _resolve(remote, host, org, repo)
+    h, o, r = _resolve(remote, host, org, repo, for_mutation=True)
     canonical_root = _canonical_local_git_root()
 
     import author_issue_bootstrap
@@ -12131,7 +12148,7 @@ def gitea_reconcile_merged_cleanups(
                 "blocker_kind": "invalid_pr_number",
             }
 
-    h, o, r = _resolve(remote, host, org, repo)
+    h, o, r = _resolve(remote, host, org, repo, for_mutation=True)
     auth = _auth(h)
     base = repo_api_url(h, o, r)
 
@@ -12521,7 +12538,7 @@ def gitea_assess_already_landed_reconciliation(
             "permission_report": _permission_block_report("gitea.read"),
         }
 
-    h, o, r = _resolve(remote, host, org, repo)
+    h, o, r = _resolve(remote, host, org, repo, for_mutation=True)
     auth = _auth(h)
     pr = api_request("GET", f"{repo_api_url(h, o, r)}/pulls/{pr_number}", auth)
 
@@ -14585,56 +14602,6 @@ def _session_context_mutation_block(
     return blocked
 
 
-def _is_client_managed_process() -> bool:
-    """Check whether the current MCP server process has client-managed launch provenance (#686)."""
-    val = (
-        os.environ.get("GITEA_CLIENT_MANAGED")
-        or os.environ.get("GITEA_MCP_CLIENT_MANAGED")
-        or os.environ.get("GITEA_SERVER_PROVENANCE")
-        or os.environ.get("GITEA_FORCE_CLIENT_MANAGED")
-        or ""
-    ).strip().lower()
-
-    if val in ("0", "false", "no", "manual", "manual_launch"):
-        return False
-
-    if val in ("1", "true", "yes", "client_managed"):
-        return True
-
-    # A terminal launch has an active TTY on stdin
-    try:
-        if sys.stdin and sys.stdin.isatty():
-            return False
-    except Exception:
-        pass
-
-    # Standard client launch or test runner with stdio pipe and profile env
-    if "GITEA_MCP_CONFIG" in os.environ or "GITEA_MCP_PROFILE" in os.environ or "GITEA_PROFILE_NAME" in os.environ:
-        return True
-
-    return False
-
-
-def _provenance_mutation_block(**extra_fields) -> dict | None:
-    """Refuse mutating tool calls on processes lacking client-managed launch provenance (#686)."""
-    if _is_client_managed_process():
-        return None
-    unconsumed = gitea_config.get_unconsumed_gitea_env_overrides()
-    blocked = {
-        "success": False,
-        "performed": False,
-        "blocker_kind": "unsupported_manual_launch",
-        "reasons": [
-            "mutation denied: server process was launched manually from a terminal without client-managed provenance (fail closed). Manually launched mcp_server.py processes cannot receive IDE stdio or serve workflow mutations."
-        ],
-        "exact_next_action": "BLOCKED + RECONNECT: Reconnect the IDE/client-managed MCP server namespace instead of an ad hoc terminal launch. Hand-launched processes and mcp_config.json hand-edits are classified as workflow contamination.",
-        "provenance": "manual_launch",
-        "unconsumed_gitea_env": unconsumed,
-    }
-    blocked.update(extra_fields)
-    return blocked
-
-
 def _profile_permission_block(required_operation: str, **extra_fields) -> dict | None:
     """Structured operation-gate denial for gated tools (#69, #142, #897).
 
@@ -14651,10 +14618,6 @@ def _profile_permission_block(required_operation: str, **extra_fields) -> dict |
     # #714: evaluate active profile only — never auto-switch.
     _ensure_matching_profile(required_operation, req_role, extra_fields.get("remote"))
 
-    prov_block = _provenance_mutation_block(**extra_fields)
-    if prov_block is not None:
-        return prov_block
-
     reasons = _profile_operation_gate(required_operation)
     if reasons:
         return _build_operation_gate_refusal(
@@ -14688,10 +14651,6 @@ def _namespace_mutation_block(mutation_task: str, **extra_fields) -> dict | None
     # #714: evaluate active profile only — never auto-switch.
     _ensure_matching_profile(required_permission, required_role, extra_fields.get("remote"))
 
-    prov_block = _provenance_mutation_block(**extra_fields)
-    if prov_block is not None:
-        return prov_block
-
     try:
         profile = get_profile()
     except Exception as exc:
@@ -18139,9 +18098,6 @@ def gitea_get_runtime_context(
         source="gitea_get_runtime_context",
     )
 
-    is_client_managed = _is_client_managed_process()
-    unconsumed_env = gitea_config.get_unconsumed_gitea_env_overrides()
-
     result = {
         "active_profile": profile["profile_name"],
         "authenticated_username": username,
@@ -18158,9 +18114,6 @@ def gitea_get_runtime_context(
         "review_merge_blocked_reasons": blocked_reasons,
         "suggested_fix": suggested_fix,
         "safe_next_action": safe_next_action,
-        "server_provenance": "client_managed" if is_client_managed else "manual_launch",
-        "is_client_managed": is_client_managed,
-        "unconsumed_gitea_env": unconsumed_env,
         "preflight_ready": preflight["preflight_ready"],
         "preflight_block_reasons": preflight["preflight_block_reasons"],
         "preflight_workspace": preflight.get("preflight_workspace"),
@@ -18174,13 +18127,6 @@ def gitea_get_runtime_context(
             PROJECT_ROOT),
     }
 
-    if not is_client_managed:
-        result["safe_next_action"] = (
-            "BLOCKED + RECONNECT: Serving process lacks client-managed launch provenance (manual launch). "
-            "Reconnect the IDE/client-managed MCP server namespace instead of an ad hoc terminal launch."
-        )
-
-
     # #702: read-only visibility into the inherited GITEA_ACTIVE_WORKTREE
     # binding; recovery itself runs during capability resolution.
     try:
@@ -20638,9 +20584,7 @@ def _check_mcp_runtimes_diagnostics(task: str, matching_profiles: list[str]) ->
     self_pid = os.getpid()
     self_stale = False
 
-    all_profile_procs: dict[str, list[dict]] = {}
-    unsupported_env_found = set()
-
+    running_profiles = {}
     for line in proc.stdout.splitlines()[1:]:
         line = line.strip()
         if not line or "mcp_server.py" not in line:
@@ -20672,55 +20616,16 @@ def _check_mcp_runtimes_diagnostics(task: str, matching_profiles: list[str]) ->
         if match:
             profile = match.group(1)
 
-        is_client_managed = bool(
-            re.search(r'\bGITEA_CLIENT_MANAGED=(1|true|yes|client_managed)\b', env_out, re.IGNORECASE)
-            or re.search(r'\bGITEA_MCP_CLIENT_MANAGED=(1|true|yes|client_managed)\b', env_out, re.IGNORECASE)
-            or re.search(r'\bGITEA_SERVER_PROVENANCE=client_managed\b', env_out, re.IGNORECASE)
-        )
-
-        for env_match in re.finditer(r'\b(GITEA_[A-Z0-9_]+)=([^\s]+)', env_out):
-            k, v = env_match.group(1), env_match.group(2)
-            if k not in gitea_config.RECOGNIZED_GITEA_ENV_KEYS and not any(k.startswith(p) for p in gitea_config.RECOGNIZED_GITEA_ENV_PREFIXES):
-                unsupported_env_found.add(f"{k}={v}")
-
         is_stale = (start_time < code_mtime) or git_stale
         if pid == self_pid and is_stale:
             self_stale = True
 
-        proc_info = {
-            "pid": pid,
-            "start_time": start_time,
-            "is_stale": is_stale,
-            "is_client_managed": is_client_managed,
-        }
-        if profile not in all_profile_procs:
-            all_profile_procs[profile] = []
-        all_profile_procs[profile].append(proc_info)
-
-    running_profiles = {}
-    for profile, procs in all_profile_procs.items():
-        if len(procs) > 1:
-            pids_str = ", ".join(str(p["pid"]) for p in procs)
-            reasons.append(
-                f"stale-runtime: Duplicate MCP server process(es) detected for profile '{profile}' (PIDs: {pids_str}). "
-                "Manual or duplicate launches defeat staleness detection and cannot receive client stdio."
-            )
-        client_procs = [p for p in procs if p["is_client_managed"]]
-        if client_procs:
-            client_procs.sort(key=lambda p: p["start_time"], reverse=True)
-            running_profiles[profile] = client_procs[0]
-        else:
-            pids_str = ", ".join(str(p["pid"]) for p in procs)
-            reasons.append(
-                f"stale-runtime: Manually launched MCP process(es) detected without client-managed provenance for profile '{profile}' (PIDs: {pids_str}). "
-                "Manual launches cannot serve client stdio and are ignored for runtime freshness."
-            )
-
-    if unsupported_env_found:
-        reasons.append(
-            f"unsupported-env: Unsupported GITEA_* environment variable override(s) detected: {', '.join(sorted(unsupported_env_found))}. "
-            "Unknown env overrides are unsupported."
-        )
+        if profile not in running_profiles or start_time > running_profiles[profile]["start_time"]:
+            running_profiles[profile] = {
+                "pid": pid,
+                "start_time": start_time,
+                "is_stale": is_stale
+            }
 
     if self_stale:
         # #685: report-only — no config utime, no thread, no os._exit.
@@ -20755,7 +20660,6 @@ def _check_mcp_runtimes_diagnostics(task: str, matching_profiles: list[str]) ->
     return reasons
 
 
-
 @mcp.tool()
 def gitea_resolve_task_capability(
     task: str,

mcp_namespace_health.py

@@ -51,14 +51,6 @@ EOF_PATTERNS = (
     "eof",
 )
 
-ERROR_CONNECTED_NAMESPACES_MISSING = "mcp_connected_namespaces_missing"
-
-UNSAFE_FALLBACK_WARNING = (
-    "Workflow Safety Hard Stop (#708): Connected-but-namespaces-missing recovery must "
-    "NEVER use direct imports, Gitea API mutations, profile hopping, session-state "
-    "overrides, PID kills, or config mtime touches. Use client reconnect only."
-)
-
 SAFE_ENV_KEYS = (
     "GITEA_MCP_PROFILE",
     "GITEA_PROFILE_NAME",
@@ -68,83 +60,6 @@ SAFE_ENV_KEYS = (
 )
 
 
-def assess_connected_namespace_attachment(
-    *,
-    connected_servers: list[str] | tuple[str, ...] | set[str] | None = None,
-    attached_session_namespaces: list[str] | tuple[str, ...] | set[str] | None = None,
-    required_namespaces: list[str] | tuple[str, ...] | set[str] | None = None,
-) -> dict[str, Any]:
-    """Assess whether host-connected MCP servers have attached tool namespaces in the active session (#708).
-
-    Addresses the Connected-but-namespaces-missing defect: CLI/host status may report Connected
-    while the active LLM session tool surface exposes 0 attached tool namespaces.
-
-    Returns structured telemetry and detection details.
-    """
-    connected = [str(s).strip() for s in (connected_servers or []) if str(s).strip()]
-    attached = set(str(ns).strip() for ns in (attached_session_namespaces or []) if str(ns).strip())
-    req = [str(r).strip() for r in (required_namespaces or DEFAULT_NAMESPACES) if str(r).strip()]
-
-    proof: dict[str, dict[str, bool]] = {}
-    missing: list[str] = []
-
-    for s in connected:
-        is_attached = s in attached
-        proof[s] = {"connected": True, "attached": is_attached}
-        if not is_attached and s in req:
-            missing.append(s)
-
-    for r in req:
-        if r not in proof:
-            proof[r] = {"connected": r in connected, "attached": r in attached}
-            if r in connected and r not in attached and r not in missing:
-                missing.append(r)
-
-    attachment_healthy = len(missing) == 0 and len(connected) > 0
-    discovery_status = (
-        "namespaces_attached" if attachment_healthy
-        else ("connected_but_namespaces_missing" if len(connected) > 0 else "disconnected")
-    )
-
-    reasons: list[str] = []
-    remediation: list[str] = []
-    if missing:
-        reasons.append(
-            f"MCP server(s) {missing} report Connected at host/CLI layer but tool namespaces "
-            f"are missing from active session attached tools (Connected ≠ attached tools, #708)."
-        )
-        remediation.append(
-            "Reconnect the IDE/client MCP session to attach namespaces to the active session. "
-            "Do not use direct imports, CLI API mutations, profile hopping, or session file overrides."
-        )
-    elif not connected:
-        reasons.append("No MCP servers reported Connected.")
-        remediation.append("Start or reconnect Gitea MCP servers in client config.")
-    else:
-        reasons.append("All connected MCP server namespaces are attached to the active session.")
-
-    return {
-        "success": attachment_healthy,
-        "attachment_healthy": attachment_healthy,
-        "discovery_status": discovery_status,
-        "connected_servers": connected,
-        "attached_session_namespaces": list(attached),
-        "missing_namespaces": missing,
-        "proof_of_connected_vs_attached": proof,
-        "error_type": None if attachment_healthy else ERROR_CONNECTED_NAMESPACES_MISSING,
-        "reasons": reasons,
-        "remediation": remediation,
-        "exact_next_action": (
-            "Reconnect the IDE/client MCP session so tool namespaces attach to the active session. "
-            "Do not use direct imports, CLI API mutations, profile hopping, or session-state overrides."
-            if not attachment_healthy
-            else "None; session tool namespaces attached."
-        ),
-        "unsafe_fallback_policy": UNSAFE_FALLBACK_WARNING,
-    }
-
-
-
 def _as_list(value: Any) -> list[str] | None:
     if value is None:
         return None
@@ -310,16 +225,6 @@ def classify_namespace_probe(
         # on bad data without treating success as IDE proof).
         blocks = namespace_health_blocks_task("merge_pr", healthy)
 
-    import gitea_config
-    raw_env = process.get("env") if isinstance(process, dict) else None
-    unconsumed_env = gitea_config.get_unconsumed_gitea_env_overrides(raw_env)
-    is_client_managed = bool(
-        env_summary.get("GITEA_CLIENT_MANAGED") in ("1", "true", "yes", "client_managed")
-        or env_summary.get("GITEA_MCP_CLIENT_MANAGED") in ("1", "true", "yes", "client_managed")
-        or env_summary.get("GITEA_SERVER_PROVENANCE") == "client_managed"
-    )
-    provenance = "client_managed" if is_client_managed else "manual_launch"
-
     return {
         "success": healthy,
         "healthy": healthy,
@@ -335,9 +240,6 @@ def classify_namespace_probe(
         "error_message": error_message or None,
         "reasons": reasons,
         "remediation": remediation,
-        "provenance": provenance,
-        "is_client_managed": is_client_managed,
-        "unconsumed_gitea_env": unconsumed_env,
         "diagnostics": {
             "namespace": ns,
             "required_tool": tool,
@@ -346,9 +248,6 @@ def classify_namespace_probe(
             "env": env_summary,
             "config_path": config_path,
             "probe_source": source,
-            "provenance": provenance,
-            "is_client_managed": is_client_managed,
-            "unconsumed_gitea_env": unconsumed_env,
         },
         "blocks_merge_workflow": blocks,
     }

remote_repo_guard.py

@@ -54,20 +54,62 @@ def assess_remote_repo_match(
     local_remote_url: str | None,
     org_explicit: bool,
     repo_explicit: bool,
+    for_mutation: bool = False,
+    primary_org: str | None = None,
+    primary_repo: str | None = None,
 ) -> dict:
     """Fail closed when the resolved org/repo disagrees with the local git remote.
 
-    The guard is intentionally conservative:
+    The guard enforces two protection levels:
 
-    * When the caller passed both ``org`` and ``repo`` explicitly, their intent is
-      authoritative and the guard never blocks.
-    * When the local git remote URL is unavailable (``None``/empty), corroboration
-      is impossible, so the guard does not block (best-effort only).
-    * Otherwise, the resolved ``org/repo`` slug must appear in the local remote URL
-      (case-insensitive); if it does not, the guard blocks.
+    1. Cross-Project Mutation Boundary (#707):
+       When ``for_mutation`` is True (codebase mutation operation: branch, commit, PR,
+       merge, branch deletion), the target repository (``resolved_org/resolved_repo``)
+       MUST match the primary authorized project context (``primary_org/primary_repo``
+       or parsed from ``local_remote_url``). Any attempt to mutate a different project
+       fails closed, even if explicit org/repo were passed. Metadata operations (such
+       as creating issues or commenting on issues) across projects remain allowed.
+
+    2. Workspace Mismatch Guard (#530):
+       When ``for_mutation`` is False (or targets match), bare remotes must resolve
+       to an org/repo slug present in the local git remote URL. When explicit org/repo
+       are supplied for non-mutation operations, the caller's intent is authoritative.
     """
     reasons: list[str] = []
 
+    eff_primary_org = primary_org
+    eff_primary_repo = primary_repo
+    if not eff_primary_org or not eff_primary_repo:
+        parsed_primary = parse_org_repo_from_remote_url(local_remote_url)
+        if parsed_primary:
+            eff_primary_org = eff_primary_org or parsed_primary[0]
+            eff_primary_repo = eff_primary_repo or parsed_primary[1]
+
+    # #707: Enforce cross-project codebase mutation boundary if for_mutation is True
+    if for_mutation and eff_primary_org and eff_primary_repo:
+        if (
+            resolved_org.lower() != eff_primary_org.lower()
+            or resolved_repo.lower() != eff_primary_repo.lower()
+        ):
+            reasons.append(
+                f"Cross-project mutation guard (#707): Attempted codebase mutation targeting "
+                f"'{resolved_org}/{resolved_repo}' outside of primary authorized project "
+                f"context '{eff_primary_org}/{eff_primary_repo}'"
+            )
+            return {
+                "proven": False,
+                "block": True,
+                "cross_project_mutation_block": True,
+                "reasons": reasons,
+                "remote": remote,
+                "resolved_org": resolved_org,
+                "resolved_repo": resolved_repo,
+                "primary_org": eff_primary_org,
+                "primary_repo": eff_primary_repo,
+                "local_remote_url": local_remote_url,
+                "remediation": f"Cross-project codebase work is forbidden. Create an issue in the target repository ('{resolved_org}/{resolved_repo}') instead.",
+            }
+
     if org_explicit and repo_explicit:
         return _assessment(True, reasons, remote, resolved_org, resolved_repo, local_remote_url)
 
@@ -88,6 +130,16 @@ def assess_remote_repo_match(
 
 def format_remote_repo_guard_error(assessment: dict) -> str:
     """Single RuntimeError message for the MCP resolver gate."""
+    if assessment.get("cross_project_mutation_block"):
+        resolved = f"{assessment.get('resolved_org')}/{assessment.get('resolved_repo')}"
+        primary = f"{assessment.get('primary_org')}/{assessment.get('primary_repo')}"
+        return (
+            f"Cross-project mutation guard (#707): Attempted codebase mutation targeting '{resolved}' "
+            f"outside of primary authorized project context '{primary}'. "
+            f"Cross-project codebase work (creating branches, committing files, creating PRs) is forbidden; "
+            f"create an issue in the target repository ('{resolved}') instead."
+        )
+
     reasons = "; ".join(
         assessment.get("reasons") or ["remote/repo resolution mismatch"]
     )
@@ -120,3 +172,4 @@ def _assessment(
         "local_remote_url": local_remote_url,
         "remediation": REMEDIATION,
     }
+

tests/conftest.py

@@ -44,8 +44,6 @@ def _reset_mutation_authority(monkeypatch):
     ]:
         monkeypatch.delenv(env_key, raising=False)
 
-    monkeypatch.setenv("GITEA_CLIENT_MANAGED", "1")
-
     # Isolate durable session-state files so tests never share host cache (#559).
     import tempfile
 

tests/test_commit_files_gate.py

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

tests/test_config_menu.py

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

tests/test_cross_project_mutation_boundary.py

@@ -0,0 +1,94 @@
+import os
+import sys
+import unittest
+from unittest import mock
+
+import remote_repo_guard
+import gitea_mcp_server as server
+from session_context_binding import _reset_session_context_for_testing, bind_session_context
+
+
+LOCAL_GITEA_TOOLS_URL = "https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools.git"
+
+
+class TestCrossProjectMutationBoundary(unittest.TestCase):
+    def setUp(self):
+        os.environ["PYTEST_CURRENT_TEST"] = "test"
+        _reset_session_context_for_testing()
+
+    def tearDown(self):
+        _reset_session_context_for_testing()
+        os.environ.pop("PYTEST_CURRENT_TEST", None)
+
+    def test_cross_project_codebase_mutation_blocked(self):
+        """Codebase mutations targeting another project must fail closed (#707)."""
+        assessment = remote_repo_guard.assess_remote_repo_match(
+            remote="prgs",
+            resolved_org="Other-Org",
+            resolved_repo="Other-Repo",
+            local_remote_url=LOCAL_GITEA_TOOLS_URL,
+            org_explicit=True,
+            repo_explicit=True,
+            for_mutation=True,
+        )
+        self.assertTrue(assessment["block"])
+        self.assertTrue(assessment.get("cross_project_mutation_block"))
+        self.assertEqual(assessment["primary_org"], "Scaled-Tech-Consulting")
+        self.assertEqual(assessment["primary_repo"], "Gitea-Tools")
+
+        err_msg = remote_repo_guard.format_remote_repo_guard_error(assessment)
+        self.assertIn("Cross-project mutation guard (#707)", err_msg)
+        self.assertIn("Attempted codebase mutation targeting 'Other-Org/Other-Repo'", err_msg)
+        self.assertIn("outside of primary authorized project context 'Scaled-Tech-Consulting/Gitea-Tools'", err_msg)
+        self.assertIn("create an issue in the target repository ('Other-Org/Other-Repo') instead", err_msg)
+
+    def test_same_project_codebase_mutation_allowed(self):
+        """Codebase mutations targeting the primary project context are allowed."""
+        assessment = remote_repo_guard.assess_remote_repo_match(
+            remote="prgs",
+            resolved_org="Scaled-Tech-Consulting",
+            resolved_repo="Gitea-Tools",
+            local_remote_url=LOCAL_GITEA_TOOLS_URL,
+            org_explicit=True,
+            repo_explicit=True,
+            for_mutation=True,
+        )
+        self.assertFalse(assessment["block"])
+        self.assertFalse(assessment.get("cross_project_mutation_block", False))
+
+    def test_cross_project_metadata_operation_allowed(self):
+        """Metadata operations (issue creation, comments) across project boundaries are allowed."""
+        assessment = remote_repo_guard.assess_remote_repo_match(
+            remote="prgs",
+            resolved_org="Other-Org",
+            resolved_repo="Other-Repo",
+            local_remote_url=LOCAL_GITEA_TOOLS_URL,
+            org_explicit=True,
+            repo_explicit=True,
+            for_mutation=False,
+        )
+        self.assertTrue(assessment["proven"])
+        self.assertFalse(assessment["block"])
+
+    @mock.patch.dict(os.environ, {"GITEA_FORCE_REMOTE_REPO_CHECK": "1"})
+    @mock.patch("gitea_mcp_server._local_git_remote_url", return_value=LOCAL_GITEA_TOOLS_URL)
+    def test_mcp_server_resolve_cross_project_mutation_fails_closed(self, mock_remote):
+        """_resolve with for_mutation=True blocks cross-project targets."""
+        with self.assertRaises(RuntimeError) as ctx:
+            server._resolve("prgs", None, "Other-Org", "Other-Repo", for_mutation=True)
+        
+        err = str(ctx.exception)
+        self.assertIn("Cross-project mutation guard (#707)", err)
+        self.assertIn("create an issue in the target repository", err)
+
+    @mock.patch.dict(os.environ, {"GITEA_FORCE_REMOTE_REPO_CHECK": "1"})
+    @mock.patch("gitea_mcp_server._local_git_remote_url", return_value=LOCAL_GITEA_TOOLS_URL)
+    def test_mcp_server_resolve_cross_project_metadata_succeeds(self, mock_remote):
+        """_resolve with for_mutation=False allows explicit cross-project target."""
+        host, org, repo = server._resolve("prgs", None, "Other-Org", "Other-Repo", for_mutation=False)
+        self.assertEqual(org, "Other-Org")
+        self.assertEqual(repo, "Other-Repo")
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_issue_686_manual_mcp_provenance.py

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

tests/test_issue_708_mcp_namespace_attachment.py

@@ -1,69 +0,0 @@
-"""Unit regression tests for Issue #708: Connected-but-namespaces-missing detection and attachment safety."""
-
-import mcp_namespace_health
-
-
-def test_assess_connected_namespace_attachment_success():
-    connected = ["gitea-author", "gitea-reviewer", "gitea-merger"]
-    attached = ["gitea-author", "gitea-reviewer", "gitea-merger"]
-    res = mcp_namespace_health.assess_connected_namespace_attachment(
-        connected_servers=connected,
-        attached_session_namespaces=attached,
-    )
-    assert res["success"] is True
-    assert res["attachment_healthy"] is True
-    assert res["discovery_status"] == "namespaces_attached"
-    assert res["error_type"] is None
-    assert res["missing_namespaces"] == []
-    assert res["exact_next_action"] == "None; session tool namespaces attached."
-
-
-def test_assess_connected_namespace_attachment_missing():
-    connected = ["gitea-author", "gitea-reviewer", "gitea-merger"]
-    attached = ["gitea-author"]
-    res = mcp_namespace_health.assess_connected_namespace_attachment(
-        connected_servers=connected,
-        attached_session_namespaces=attached,
-    )
-    assert res["success"] is False
-    assert res["attachment_healthy"] is False
-    assert res["discovery_status"] == "connected_but_namespaces_missing"
-    assert res["error_type"] == "mcp_connected_namespaces_missing"
-    assert "gitea-reviewer" in res["missing_namespaces"]
-    assert "gitea-merger" in res["missing_namespaces"]
-    assert "Reconnect the IDE/client MCP session" in res["exact_next_action"]
-
-
-def test_assess_connected_namespace_attachment_disconnected():
-    res = mcp_namespace_health.assess_connected_namespace_attachment(
-        connected_servers=[],
-        attached_session_namespaces=[],
-    )
-    assert res["success"] is False
-    assert res["attachment_healthy"] is False
-    assert res["discovery_status"] == "disconnected"
-
-
-def test_proof_of_connected_vs_attached_mapping():
-    connected = ["gitea-author", "gitea-reviewer"]
-    attached = ["gitea-author"]
-    res = mcp_namespace_health.assess_connected_namespace_attachment(
-        connected_servers=connected,
-        attached_session_namespaces=attached,
-    )
-    proof = res["proof_of_connected_vs_attached"]
-    assert proof["gitea-author"] == {"connected": True, "attached": True}
-    assert proof["gitea-reviewer"] == {"connected": True, "attached": False}
-
-
-def test_unsafe_fallback_policy_enforcement():
-    res = mcp_namespace_health.assess_connected_namespace_attachment(
-        connected_servers=["gitea-author"],
-        attached_session_namespaces=[],
-    )
-    policy = res["unsafe_fallback_policy"]
-    assert "Workflow Safety Hard Stop (#708)" in policy
-    assert "direct imports" in policy
-    assert "API mutations" in policy
-    assert "profile hopping" in policy
-    assert "session-state overrides" in policy

tests/test_mcp_stale_runtime.py

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

tests/test_runtime_clarity.py

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