Merge pull request 'fix(guardrail): detect and reject manually launched duplicate MCP role servers (#686)' (#925) from fix/issue-686-detect-reject-manual-mcp into master

docs/mcp-namespace-eof-recovery.md

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

gitea_auth.py

@@ -22,62 +22,8 @@ import gitea_config
 
 PROJECT_ROOT = os.path.dirname(os.path.abspath(__file__))
 
-# Reserved runtime-control / workspace-binding environment variables (#704).
-# Repository .env files MUST NOT populate or override any of these keys.
-RESERVED_WORKTREE_ENV_KEYS: frozenset[str] = frozenset({
-    "GITEA_ACTIVE_WORKTREE",
-    "GITEA_AUTHOR_WORKTREE",
-    "GITEA_REVIEWER_WORKTREE",
-    "GITEA_MERGER_WORKTREE",
-    "GITEA_RECONCILER_WORKTREE",
-})
-
-
-def is_reserved_worktree_env_key(key: str | None) -> bool:
-    """Return True if *key* is a reserved runtime workspace binding variable (#704)."""
-    if not key:
-        return False
-    k = str(key).upper().strip()
-    return k in RESERVED_WORKTREE_ENV_KEYS or (k.startswith("GITEA_") and k.endswith("_WORKTREE"))
-
-
-def load_env_file_sanitized(
-    env_path: str,
-    *,
-    target_env: dict | os._Environ | None = None,
-) -> list[str]:
-    """Load a .env file without populating or overriding reserved workspace keys (#704).
-
-    Pre-existing process environment values retain their precedence. Reserved
-    runtime-control keys found in repository files are ignored (without logging
-    their values). Returns a list of sanitized rejection reasons.
-    """
-    if target_env is None:
-        target_env = os.environ
-    if not os.path.exists(env_path) or os.path.isdir(env_path):
-        return []
-
-    rejection_reasons: list[str] = []
-    try:
-        file_vals = dotenv_values(env_path)
-        for key, val in file_vals.items():
-            if not key or val is None:
-                continue
-            if is_reserved_worktree_env_key(key):
-                filename = os.path.basename(env_path)
-                rejection_reasons.append(
-                    f"Ignored reserved runtime workspace key '{key}' from repository {filename}"
-                )
-                continue
-            if key not in target_env:
-                target_env[key] = val
-    except Exception:
-        pass
-    return rejection_reasons
-
-
-# Load standard .env if present (sanitized to prevent repo workspace binding contamination #704)
-load_env_file_sanitized(os.path.join(PROJECT_ROOT, ".env"))
+# Load standard .env if present
+load_dotenv(os.path.join(PROJECT_ROOT, ".env"))
 
 # Dictionary to store configurations parsed dynamically from .env.* files
 DYNAMIC_CONFIGS = {}
@@ -91,13 +37,9 @@ for env_path in glob.glob(os.path.join(PROJECT_ROOT, ".env*")):
         continue
     try:
         config_vals = dotenv_values(env_path)
-        # Filter out reserved workspace keys from dynamic configs (#704)
-        sanitized_config = {
-            k: v for k, v in config_vals.items() if not is_reserved_worktree_env_key(k)
-        }
-        site = sanitized_config.get("GITEA_SITE") or sanitized_config.get("GITEA_HOST")
+        site = config_vals.get("GITEA_SITE") or config_vals.get("GITEA_HOST")
         if site:
-            DYNAMIC_CONFIGS[site.lower().strip()] = sanitized_config
+            DYNAMIC_CONFIGS[site.lower().strip()] = config_vals
     except Exception:
         pass
 

gitea_config.py

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

gitea_mcp_server.py

@@ -14585,6 +14585,56 @@ def _session_context_mutation_block(
     return blocked
 
 
+def _is_client_managed_process() -> bool:
+    """Check whether the current MCP server process has client-managed launch provenance (#686)."""
+    val = (
+        os.environ.get("GITEA_CLIENT_MANAGED")
+        or os.environ.get("GITEA_MCP_CLIENT_MANAGED")
+        or os.environ.get("GITEA_SERVER_PROVENANCE")
+        or os.environ.get("GITEA_FORCE_CLIENT_MANAGED")
+        or ""
+    ).strip().lower()
+
+    if val in ("0", "false", "no", "manual", "manual_launch"):
+        return False
+
+    if val in ("1", "true", "yes", "client_managed"):
+        return True
+
+    # A terminal launch has an active TTY on stdin
+    try:
+        if sys.stdin and sys.stdin.isatty():
+            return False
+    except Exception:
+        pass
+
+    # Standard client launch or test runner with stdio pipe and profile env
+    if "GITEA_MCP_CONFIG" in os.environ or "GITEA_MCP_PROFILE" in os.environ or "GITEA_PROFILE_NAME" in os.environ:
+        return True
+
+    return False
+
+
+def _provenance_mutation_block(**extra_fields) -> dict | None:
+    """Refuse mutating tool calls on processes lacking client-managed launch provenance (#686)."""
+    if _is_client_managed_process():
+        return None
+    unconsumed = gitea_config.get_unconsumed_gitea_env_overrides()
+    blocked = {
+        "success": False,
+        "performed": False,
+        "blocker_kind": "unsupported_manual_launch",
+        "reasons": [
+            "mutation denied: server process was launched manually from a terminal without client-managed provenance (fail closed). Manually launched mcp_server.py processes cannot receive IDE stdio or serve workflow mutations."
+        ],
+        "exact_next_action": "BLOCKED + RECONNECT: Reconnect the IDE/client-managed MCP server namespace instead of an ad hoc terminal launch. Hand-launched processes and mcp_config.json hand-edits are classified as workflow contamination.",
+        "provenance": "manual_launch",
+        "unconsumed_gitea_env": unconsumed,
+    }
+    blocked.update(extra_fields)
+    return blocked
+
+
 def _profile_permission_block(required_operation: str, **extra_fields) -> dict | None:
     """Structured operation-gate denial for gated tools (#69, #142, #897).
 
@@ -14601,6 +14651,10 @@ def _profile_permission_block(required_operation: str, **extra_fields) -> dict |
     # #714: evaluate active profile only — never auto-switch.
     _ensure_matching_profile(required_operation, req_role, extra_fields.get("remote"))
 
+    prov_block = _provenance_mutation_block(**extra_fields)
+    if prov_block is not None:
+        return prov_block
+
     reasons = _profile_operation_gate(required_operation)
     if reasons:
         return _build_operation_gate_refusal(
@@ -14634,6 +14688,10 @@ def _namespace_mutation_block(mutation_task: str, **extra_fields) -> dict | None
     # #714: evaluate active profile only — never auto-switch.
     _ensure_matching_profile(required_permission, required_role, extra_fields.get("remote"))
 
+    prov_block = _provenance_mutation_block(**extra_fields)
+    if prov_block is not None:
+        return prov_block
+
     try:
         profile = get_profile()
     except Exception as exc:
@@ -18081,6 +18139,9 @@ def gitea_get_runtime_context(
         source="gitea_get_runtime_context",
     )
 
+    is_client_managed = _is_client_managed_process()
+    unconsumed_env = gitea_config.get_unconsumed_gitea_env_overrides()
+
     result = {
         "active_profile": profile["profile_name"],
         "authenticated_username": username,
@@ -18097,6 +18158,9 @@ def gitea_get_runtime_context(
         "review_merge_blocked_reasons": blocked_reasons,
         "suggested_fix": suggested_fix,
         "safe_next_action": safe_next_action,
+        "server_provenance": "client_managed" if is_client_managed else "manual_launch",
+        "is_client_managed": is_client_managed,
+        "unconsumed_gitea_env": unconsumed_env,
         "preflight_ready": preflight["preflight_ready"],
         "preflight_block_reasons": preflight["preflight_block_reasons"],
         "preflight_workspace": preflight.get("preflight_workspace"),
@@ -18110,6 +18174,13 @@ 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:
@@ -20567,7 +20638,9 @@ def _check_mcp_runtimes_diagnostics(task: str, matching_profiles: list[str]) ->
     self_pid = os.getpid()
     self_stale = False
 
-    running_profiles = {}
+    all_profile_procs: dict[str, list[dict]] = {}
+    unsupported_env_found = set()
+
     for line in proc.stdout.splitlines()[1:]:
         line = line.strip()
         if not line or "mcp_server.py" not in line:
@@ -20599,16 +20672,55 @@ def _check_mcp_runtimes_diagnostics(task: str, matching_profiles: list[str]) ->
         if match:
             profile = match.group(1)
 
+        is_client_managed = bool(
+            re.search(r'\bGITEA_CLIENT_MANAGED=(1|true|yes|client_managed)\b', env_out, re.IGNORECASE)
+            or re.search(r'\bGITEA_MCP_CLIENT_MANAGED=(1|true|yes|client_managed)\b', env_out, re.IGNORECASE)
+            or re.search(r'\bGITEA_SERVER_PROVENANCE=client_managed\b', env_out, re.IGNORECASE)
+        )
+
+        for env_match in re.finditer(r'\b(GITEA_[A-Z0-9_]+)=([^\s]+)', env_out):
+            k, v = env_match.group(1), env_match.group(2)
+            if k not in gitea_config.RECOGNIZED_GITEA_ENV_KEYS and not any(k.startswith(p) for p in gitea_config.RECOGNIZED_GITEA_ENV_PREFIXES):
+                unsupported_env_found.add(f"{k}={v}")
+
         is_stale = (start_time < code_mtime) or git_stale
         if pid == self_pid and is_stale:
             self_stale = True
 
-        if profile not in running_profiles or start_time > running_profiles[profile]["start_time"]:
-            running_profiles[profile] = {
-                "pid": pid,
-                "start_time": start_time,
-                "is_stale": is_stale
-            }
+        proc_info = {
+            "pid": pid,
+            "start_time": start_time,
+            "is_stale": is_stale,
+            "is_client_managed": is_client_managed,
+        }
+        if profile not in all_profile_procs:
+            all_profile_procs[profile] = []
+        all_profile_procs[profile].append(proc_info)
+
+    running_profiles = {}
+    for profile, procs in all_profile_procs.items():
+        if len(procs) > 1:
+            pids_str = ", ".join(str(p["pid"]) for p in procs)
+            reasons.append(
+                f"stale-runtime: Duplicate MCP server process(es) detected for profile '{profile}' (PIDs: {pids_str}). "
+                "Manual or duplicate launches defeat staleness detection and cannot receive client stdio."
+            )
+        client_procs = [p for p in procs if p["is_client_managed"]]
+        if client_procs:
+            client_procs.sort(key=lambda p: p["start_time"], reverse=True)
+            running_profiles[profile] = client_procs[0]
+        else:
+            pids_str = ", ".join(str(p["pid"]) for p in procs)
+            reasons.append(
+                f"stale-runtime: Manually launched MCP process(es) detected without client-managed provenance for profile '{profile}' (PIDs: {pids_str}). "
+                "Manual launches cannot serve client stdio and are ignored for runtime freshness."
+            )
+
+    if unsupported_env_found:
+        reasons.append(
+            f"unsupported-env: Unsupported GITEA_* environment variable override(s) detected: {', '.join(sorted(unsupported_env_found))}. "
+            "Unknown env overrides are unsupported."
+        )
 
     if self_stale:
         # #685: report-only — no config utime, no thread, no os._exit.
@@ -20643,6 +20755,7 @@ 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

@@ -225,6 +225,16 @@ def classify_namespace_probe(
         # on bad data without treating success as IDE proof).
         blocks = namespace_health_blocks_task("merge_pr", healthy)
 
+    import gitea_config
+    raw_env = process.get("env") if isinstance(process, dict) else None
+    unconsumed_env = gitea_config.get_unconsumed_gitea_env_overrides(raw_env)
+    is_client_managed = bool(
+        env_summary.get("GITEA_CLIENT_MANAGED") in ("1", "true", "yes", "client_managed")
+        or env_summary.get("GITEA_MCP_CLIENT_MANAGED") in ("1", "true", "yes", "client_managed")
+        or env_summary.get("GITEA_SERVER_PROVENANCE") == "client_managed"
+    )
+    provenance = "client_managed" if is_client_managed else "manual_launch"
+
     return {
         "success": healthy,
         "healthy": healthy,
@@ -240,6 +250,9 @@ def classify_namespace_probe(
         "error_message": error_message or None,
         "reasons": reasons,
         "remediation": remediation,
+        "provenance": provenance,
+        "is_client_managed": is_client_managed,
+        "unconsumed_gitea_env": unconsumed_env,
         "diagnostics": {
             "namespace": ns,
             "required_tool": tool,
@@ -248,6 +261,9 @@ def classify_namespace_probe(
             "env": env_summary,
             "config_path": config_path,
             "probe_source": source,
+            "provenance": provenance,
+            "is_client_managed": is_client_managed,
+            "unconsumed_gitea_env": unconsumed_env,
         },
         "blocks_merge_workflow": blocks,
     }

tests/conftest.py

@@ -44,6 +44,8 @@ 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": ["Example-Org/Example-Repo"],
+        "allowed_repositories": ["Scaled-Tech-Consulting/Gitea-Tools", "Example-Org/Example-Repo"],
         },
         "reviewer-no-commit": {
             "enabled": True,
@@ -50,7 +50,7 @@ CONFIG = {
                 "gitea.repo.commit", "gitea.pr.create", "gitea.branch.push"
             ],
             "execution_profile": "reviewer-no-commit",
-        "allowed_repositories": ["Example-Org/Example-Repo"],
+        "allowed_repositories": ["Scaled-Tech-Consulting/Gitea-Tools", "Example-Org/Example-Repo"],
         },
     },
     "rules": {"allow_runtime_switching": False},

tests/test_config_menu.py

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

tests/test_issue_686_manual_mcp_provenance.py

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

tests/test_issue_704_prevent_env_workspace_bindings.py

@@ -1,166 +0,0 @@
-"""Tests for Issue #704: Preventing repository .env files from injecting workspace bindings.
-
-Acceptance Criteria (#704):
-1. Repository .env loading cannot populate or override GITEA_ACTIVE_WORKTREE or any role-specific GITEA_*_WORKTREE runtime-binding variable.
-2. Runtime workspace bindings are accepted only from sanctioned managed-launch/session mechanisms.
-3. Pre-existing sanctioned process environment values retain their intended precedence.
-4. Importing gitea_auth or related modules does not mutate workspace-binding state from repository files.
-5. Reserved runtime-control keys found in .env are ignored or rejected with a sanitized actionable reason; their values are never logged.
-6. The protection applies consistently to author, reviewer, merger, and reconciler namespaces.
-7. Comprehensive test coverage for stale worktree, missing worktree, task-specific, role-specific, launcher binding, repeated imports, namespace isolation, precedence, and absence of secret leakage.
-8. Dirty-state and workspace-preflight gates cannot be bypassed by an injected missing-path binding.
-9. No environment, dotenv, offline-import, or caller-controlled path can forge native transport or mutation provenance.
-10. Cross-linked with #702, PR #703, #510.
-11. Required immediate follow-up to Issue #702 / PR #703.
-"""
-
-from __future__ import annotations
-
-import os
-import sys
-import tempfile
-import importlib
-import unittest
-from pathlib import Path
-
-sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
-
-import gitea_auth
-from gitea_auth import is_reserved_worktree_env_key, load_env_file_sanitized
-
-
-class TestIssue704PreventEnvWorkspaceBindings(unittest.TestCase):
-    """Test suite verifying .env workspace-binding injection prevention (#704)."""
-
-    def setUp(self):
-        self.tmpdir = tempfile.TemporaryDirectory()
-        self.addCleanup(self.tmpdir.cleanup)
-        self.env_dir = Path(self.tmpdir.name)
-
-    def test_is_reserved_worktree_env_key(self):
-        """Verify key classification for all role namespaces (#704 AC6)."""
-        reserved_keys = [
-            "GITEA_ACTIVE_WORKTREE",
-            "GITEA_AUTHOR_WORKTREE",
-            "GITEA_REVIEWER_WORKTREE",
-            "GITEA_MERGER_WORKTREE",
-            "GITEA_RECONCILER_WORKTREE",
-            "gitea_active_worktree",
-            "GITEA_CUSTOM_ROLE_WORKTREE",
-        ]
-        for key in reserved_keys:
-            self.assertTrue(
-                is_reserved_worktree_env_key(key),
-                f"Expected {key} to be recognized as a reserved worktree key",
-            )
-
-        unreserved_keys = [
-            "GITEA_USER",
-            "GITEA_PASS",
-            "GITEA_TOKEN",
-            "GITEA_HOST",
-            "PATH",
-        ]
-        for key in unreserved_keys:
-            self.assertFalse(
-                is_reserved_worktree_env_key(key),
-                f"Expected {key} to NOT be recognized as a reserved worktree key",
-            )
-
-    def test_load_env_file_sanitized_ignores_reserved_keys(self):
-        """Verify .env loading ignores GITEA_ACTIVE_WORKTREE and role-specific keys (#704 AC1)."""
-        env_file = self.env_dir / ".env"
-        stale_path = "/tmp/stale-worktree-path-1234"
-        env_file.write_text(
-            f"GITEA_USER=testuser\n"
-            f"GITEA_ACTIVE_WORKTREE={stale_path}\n"
-            f"GITEA_AUTHOR_WORKTREE={stale_path}\n"
-            f"GITEA_REVIEWER_WORKTREE={stale_path}\n"
-            f"GITEA_MERGER_WORKTREE={stale_path}\n"
-            f"GITEA_RECONCILER_WORKTREE={stale_path}\n"
-        )
-
-        test_env = {}
-        reasons = load_env_file_sanitized(str(env_file), target_env=test_env)
-
-        # Unreserved key loaded
-        self.assertEqual(test_env.get("GITEA_USER"), "testuser")
-
-        # Reserved keys ignored
-        self.assertNotIn("GITEA_ACTIVE_WORKTREE", test_env)
-        self.assertNotIn("GITEA_AUTHOR_WORKTREE", test_env)
-        self.assertNotIn("GITEA_REVIEWER_WORKTREE", test_env)
-        self.assertNotIn("GITEA_MERGER_WORKTREE", test_env)
-        self.assertNotIn("GITEA_RECONCILER_WORKTREE", test_env)
-
-        # Rejection reasons populated without leaking the secret value (#704 AC5)
-        self.assertTrue(len(reasons) >= 5)
-        for r in reasons:
-            self.assertNotIn(stale_path, r, "Secret/path value must not leak into rejection reason")
-
-    def test_preexisting_sanctioned_launcher_env_retained(self):
-        """Sanctioned launcher values in process env are retained (#704 AC2, AC3)."""
-        sanctioned_path = "/tmp/sanctioned-launcher-worktree"
-        test_env = {"GITEA_ACTIVE_WORKTREE": sanctioned_path}
-
-        env_file = self.env_dir / ".env"
-        env_file.write_text("GITEA_ACTIVE_WORKTREE=/tmp/injected-repo-worktree\n")
-
-        load_env_file_sanitized(str(env_file), target_env=test_env)
-
-        # Pre-existing value retained, not overwritten by .env
-        self.assertEqual(test_env.get("GITEA_ACTIVE_WORKTREE"), sanctioned_path)
-
-    def test_stale_or_missing_worktree_in_env_ignored(self):
-        """Stale or non-existent worktree path in .env file is ignored (#704 AC7)."""
-        nonexistent_path = "/nonexistent/branches/stale-issue-999"
-        env_file = self.env_dir / ".env"
-        env_file.write_text(f"GITEA_ACTIVE_WORKTREE={nonexistent_path}\n")
-
-        test_env = {}
-        load_env_file_sanitized(str(env_file), target_env=test_env)
-
-        self.assertNotIn("GITEA_ACTIVE_WORKTREE", test_env)
-
-    def test_repeated_module_import_does_not_mutate_workspace_env(self):
-        """Repeated imports of gitea_auth leave os.environ un-contaminated (#704 AC4, AC7)."""
-        # Ensure no active worktree env exists initially
-        original_val = os.environ.pop("GITEA_ACTIVE_WORKTREE", None)
-        try:
-            importlib.reload(gitea_auth)
-            self.assertNotIn("GITEA_ACTIVE_WORKTREE", os.environ)
-
-            importlib.reload(gitea_auth)
-            self.assertNotIn("GITEA_ACTIVE_WORKTREE", os.environ)
-        finally:
-            if original_val is not None:
-                os.environ["GITEA_ACTIVE_WORKTREE"] = original_val
-
-    def test_namespace_isolation_all_roles_protected(self):
-        """Verify protection across author, reviewer, merger, reconciler (#704 AC6)."""
-        env_file = self.env_dir / ".env"
-        env_file.write_text(
-            "GITEA_AUTHOR_WORKTREE=/bad/author\n"
-            "GITEA_REVIEWER_WORKTREE=/bad/reviewer\n"
-            "GITEA_MERGER_WORKTREE=/bad/merger\n"
-            "GITEA_RECONCILER_WORKTREE=/bad/reconciler\n"
-        )
-        test_env = {}
-        load_env_file_sanitized(str(env_file), target_env=test_env)
-
-        self.assertEqual(test_env, {})
-
-    def test_absence_of_secret_leakage(self):
-        """Rejection reasons contain key names but never secret path values (#704 AC5)."""
-        sensitive_path = "/Users/secret/path/private_repo"
-        env_file = self.env_dir / ".env"
-        env_file.write_text(f"GITEA_ACTIVE_WORKTREE={sensitive_path}\n")
-
-        test_env = {}
-        reasons = load_env_file_sanitized(str(env_file), target_env=test_env)
-        for reason in reasons:
-            self.assertNotIn(sensitive_path, reason)
-
-
-if __name__ == "__main__":
-    unittest.main()

tests/test_mcp_stale_runtime.py

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

tests/test_runtime_clarity.py

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