feat(routing): enforce role alignment in automatic dispatch profile switching

gitea_config.py

@@ -205,7 +205,7 @@ def selected_profile_name():
 
 
 def is_runtime_switching_enabled(path=None):
-    """Check if runtime profile switching is enabled in config."""
+    """Check if runtime profile switching is explicitly enabled in config."""
     try:
         config = load_config(path)
     except Exception:
@@ -213,18 +213,10 @@ def is_runtime_switching_enabled(path=None):
     if not config:
         return False
     rules = config.get("rules") or {}
-    if rules.get("allow_runtime_switching") is False:
-        return False
-    if config.get("allow_runtime_switching") is False:
-        return False
     if rules.get("allow_runtime_switching") is True:
         return True
     if config.get("allow_runtime_switching") is True:
         return True
-    # Default to True if multiple profiles exist in the config
-    profiles = config.get("profiles") or {}
-    if len(profiles) > 1:
-        return True
     return False
 
 

gitea_mcp_server.py

@@ -164,194 +164,81 @@ _preflight_capability_called = False
 _preflight_whoami_violation = False
 _preflight_capability_violation = False
 _preflight_resolved_role = None
-_process_start_porcelain: str | None = None
-_preflight_whoami_baseline_porcelain: str | None = None
-_preflight_capability_baseline_porcelain: str | None = None
-_preflight_whoami_violation_files: list[str] = []
-_preflight_capability_violation_files: list[str] = []
-_preflight_reviewer_violation_files: list[str] = []
-
-
-def _preflight_in_test_mode() -> bool:
-    return "pytest" in sys.modules or "unittest" in sys.modules
-
-
-def _ensure_process_start_porcelain() -> str:
-    """Capture the shared-worktree baseline once per MCP process (#252)."""
-    global _process_start_porcelain
-    if _process_start_porcelain is None:
-        _process_start_porcelain = _get_workspace_porcelain()
-    return _process_start_porcelain
-
-
-def _get_workspace_porcelain() -> str:
-    """Return tracked-workspace porcelain for pre-flight attribution."""
-    if os.environ.get("GITEA_TEST_FORCE_DIRTY"):
-        return " M __gitea_test_force_dirty__.py\n"
-    override = os.environ.get("GITEA_TEST_PORCELAIN")
-    if override is not None:
-        return override
-    try:
-        res = subprocess.run(
-            ["git", "status", "--porcelain"],
-            capture_output=True,
-            text=True,
-            cwd=PROJECT_ROOT,
-        )
-        return res.stdout or ""
-    except Exception:
-        return ""
-
-
-def _parse_porcelain_entries(porcelain: str) -> dict[str, str]:
-    """Map tracked path -> full porcelain line (untracked ``??`` ignored)."""
-    entries: dict[str, str] = {}
-    for line in (porcelain or "").splitlines():
-        if not line or len(line) < 4 or line.startswith("??"):
-            continue
-        path = line[3:].strip()
-        if " -> " in path:
-            path = path.split(" -> ", 1)[1].strip()
-        if path:
-            entries[path] = line
-    return entries
-
-
-def _new_tracked_changes_since(baseline: str, current: str) -> list[str]:
-    """Tracked paths that are new or changed since *baseline* porcelain."""
-    base = _parse_porcelain_entries(baseline)
-    cur = _parse_porcelain_entries(current)
-    changed = [
-        path for path, line in cur.items()
-        if path not in base or base[path] != line
-    ]
-    return sorted(changed)
-
-
-def _format_preflight_files(files: list[str]) -> str:
-    if not files:
-        return "(none)"
-    return ", ".join(files)
-
-
-def assess_preflight_status() -> dict:
-    """Non-throwing pre-flight readiness for runtime-context alignment (#252)."""
-    reasons: list[str] = []
-    if not _preflight_whoami_called:
-        reasons.append(
-            "Identity (gitea_whoami) has not been verified"
-        )
-    if not _preflight_capability_called:
-        reasons.append(
-            "Task capability (gitea_resolve_task_capability) has not been resolved"
-        )
-    if _preflight_whoami_violation:
-        reasons.append(
-            "Workspace file edits occurred before gitea_whoami verification "
-            f"(offending files: {_format_preflight_files(_preflight_whoami_violation_files)})"
-        )
-    if _preflight_capability_violation:
-        reasons.append(
-            "Workspace file edits occurred before gitea_resolve_task_capability verification "
-            f"(offending files: {_format_preflight_files(_preflight_capability_violation_files)})"
-        )
-    if _preflight_reviewer_violation_files:
-        reasons.append(
-            "Reviewer profile modified tracked workspace files after capability resolution "
-            f"(offending files: {_format_preflight_files(_preflight_reviewer_violation_files)})"
-        )
-    return {
-        "preflight_ready": not reasons,
-        "preflight_block_reasons": reasons,
-        "preflight_whoami_verified": _preflight_whoami_called,
-        "preflight_capability_resolved": _preflight_capability_called,
-        "preflight_whoami_violation_files": list(_preflight_whoami_violation_files),
-        "preflight_capability_violation_files": list(_preflight_capability_violation_files),
-        "preflight_reviewer_violation_files": list(_preflight_reviewer_violation_files),
-    }
-
 
 def record_preflight_check(type_name: str, resolved_role: str | None = None):
-    """Record a pre-flight check (whoami or capability) with session-scoped deltas."""
+    """Record a pre-flight check (whoami or capability) and check for workspace edits."""
     global _preflight_whoami_called, _preflight_capability_called
     global _preflight_whoami_violation, _preflight_capability_violation
     global _preflight_resolved_role
-    global _preflight_whoami_baseline_porcelain, _preflight_capability_baseline_porcelain
-    global _preflight_whoami_violation_files, _preflight_capability_violation_files
-    global _preflight_reviewer_violation_files
 
-    current = _get_workspace_porcelain()
+    in_test = "pytest" in sys.modules or "unittest" in sys.modules
+    if os.environ.get("GITEA_TEST_FORCE_DIRTY"):
+        is_dirty = True
+    elif in_test:
+        is_dirty = False
+    else:
+        is_dirty = False
+        try:
+            res = subprocess.run(
+                ["git", "status", "--porcelain"],
+                capture_output=True, text=True, cwd=PROJECT_ROOT
+            )
+            for line in res.stdout.splitlines():
+                if line and not line.startswith("??"):
+                    is_dirty = True
+                    break
+        except Exception:
+            pass
+
+    if is_dirty:
+        if type_name == "whoami" and not _preflight_whoami_called:
+            _preflight_whoami_violation = True
+        if type_name == "capability" and not _preflight_capability_called:
+            _preflight_capability_violation = True
 
     if type_name == "whoami":
-        # Fresh whoami restarts the capability step and re-evaluates violations
-        # instead of replaying a sticky record (#252).
-        _preflight_capability_called = False
-        _preflight_capability_violation = False
-        _preflight_capability_violation_files = []
-        _preflight_capability_baseline_porcelain = None
-        _preflight_reviewer_violation_files = []
-
-        process_start = _ensure_process_start_porcelain()
-        whoami_delta = _new_tracked_changes_since(process_start, current)
-        _preflight_whoami_violation = bool(whoami_delta)
-        _preflight_whoami_violation_files = whoami_delta
-        _preflight_whoami_baseline_porcelain = current
         _preflight_whoami_called = True
     elif type_name == "capability":
-        baseline = _preflight_whoami_baseline_porcelain or ""
-        capability_delta = _new_tracked_changes_since(baseline, current)
-        _preflight_capability_violation = bool(capability_delta)
-        _preflight_capability_violation_files = capability_delta
-        _preflight_capability_baseline_porcelain = current
         _preflight_capability_called = True
         if resolved_role:
             _preflight_resolved_role = resolved_role
 
-
 def verify_preflight_purity(remote: str | None = None):
-    """Verify that identity and capability were verified prior to session edits."""
+    """Verify that identity and capability were verified prior to edits, and that reviewers made no edits."""
-    global _preflight_reviewer_violation_files
+    in_test = "pytest" in sys.modules or "unittest" in sys.modules
-
+    if in_test and not os.environ.get("GITEA_TEST_FORCE_DIRTY"):
-    in_test = _preflight_in_test_mode()
-    if in_test and not (
-        os.environ.get("GITEA_TEST_FORCE_DIRTY")
-        or os.environ.get("GITEA_TEST_PORCELAIN") is not None
-    ):
         return
 
     if not _preflight_whoami_called:
-        raise RuntimeError(
+        raise RuntimeError("Pre-flight order violation: Identity (gitea_whoami) has not been verified (fail closed)")
-            "Pre-flight order violation: Identity (gitea_whoami) has not been verified (fail closed)"
-        )
     if not _preflight_capability_called:
-        raise RuntimeError(
+        raise RuntimeError("Pre-flight order violation: Task capability (gitea_resolve_task_capability) has not been resolved (fail closed)")
-            "Pre-flight order violation: Task capability (gitea_resolve_task_capability) has not been resolved (fail closed)"
-        )
 
     if _preflight_whoami_violation:
-        raise RuntimeError(
+        raise RuntimeError("Pre-flight order violation: Workspace file edits occurred before gitea_whoami verification (fail closed)")
-            "Pre-flight order violation: Workspace file edits occurred before "
-            f"gitea_whoami verification (fail closed). Offending files: "
-            f"{_format_preflight_files(_preflight_whoami_violation_files)}"
-        )
     if _preflight_capability_violation:
-        raise RuntimeError(
+        raise RuntimeError("Pre-flight order violation: Workspace file edits occurred before gitea_resolve_task_capability verification (fail closed)")
-            "Pre-flight order violation: Workspace file edits occurred before "
-            f"gitea_resolve_task_capability verification (fail closed). Offending files: "
-            f"{_format_preflight_files(_preflight_capability_violation_files)}"
-        )
 
-    if _preflight_resolved_role == "reviewer":
+    if os.environ.get("GITEA_TEST_FORCE_DIRTY"):
-        current = _get_workspace_porcelain()
+        is_dirty = True
-        baseline = _preflight_capability_baseline_porcelain or ""
+    elif in_test:
-        reviewer_delta = _new_tracked_changes_since(baseline, current)
+        is_dirty = False
-        _preflight_reviewer_violation_files = reviewer_delta
+    else:
-        if reviewer_delta:
+        is_dirty = False
-            raise RuntimeError(
+        try:
-                "Reviewer role violation: Reviewer profile is forbidden from modifying "
+            res = subprocess.run(
-                "tracked workspace files (fail closed). Offending files: "
+                ["git", "status", "--porcelain"],
-                f"{_format_preflight_files(reviewer_delta)}"
+                capture_output=True, text=True, cwd=PROJECT_ROOT
             )
+            for line in res.stdout.splitlines():
+                if line and not line.startswith("??"):
+                    is_dirty = True
+                    break
+        except Exception:
+            pass
+
+    if _preflight_resolved_role == "reviewer" and is_dirty:
+        raise RuntimeError("Reviewer role violation: Reviewer profile is forbidden from modifying tracked workspace files (fail closed)")
 
 from mcp.server.fastmcp import FastMCP  # noqa: E402
 
@@ -551,6 +438,9 @@ def _ensure_matching_profile(required_permission: str, required_role: str, remot
             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_role = p_data.get("role") or _role_kind(p_allowed, p_forbidden)
+                if required_role and p_role != required_role:
+                    continue
                 p_allowed_n = []
                 for op in p_allowed:
                     try:
@@ -3061,76 +2951,6 @@ def _permission_block_report(required_operation: str,
     return report
 
 
-def _role_for_operation(op: str) -> str | None:
-    # Normalize op first
-    try:
-        op_n = gitea_config.normalize_operation(op)
-    except Exception:
-        op_n = op
-    if op_n in ("gitea.pr.approve", "gitea.pr.merge", "gitea.pr.review", "gitea.pr.request_changes"):
-        return "reviewer"
-    if op_n in (
-        "gitea.issue.create",
-        "gitea.issue.comment",
-        "gitea.issue.close",
-        "gitea.branch.create",
-        "gitea.branch.push",
-        "gitea.pr.create",
-        "gitea.pr.comment",
-        "gitea.pr.close",
-        "gitea.branch.delete",
-        "gitea.repo.commit"
-    ):
-        return "author"
-    return 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.
-
-    If found, switch to it, clear identity cache, and return True.
-    Otherwise return False.
-    """
-    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
-    return False
-
-
 def _profile_operation_gate(op: str) -> list[str]:
     """Profile permission check for a single gated operation (#126, #216).
 
@@ -3148,17 +2968,6 @@ def _profile_operation_gate(op: str) -> list[str]:
         op, profile["allowed_operations"], profile["forbidden_operations"])
     if op_ok:
         return []
-
-    if _try_auto_switch_for_operation(op):
-        try:
-            profile = get_profile()
-            op_ok, op_reason = gitea_config.check_operation(
-                op, profile["allowed_operations"], profile["forbidden_operations"])
-            if op_ok:
-                return []
-        except Exception as exc:
-            return [f"profile could not be resolved (fail closed): {_redact(str(exc))}"]
-
     if op_reason == "no-allowed-operations":
         return ["profile has no configured allowed operations (fail closed)"]
     if op_reason == "forbidden":
@@ -4161,14 +3970,6 @@ def gitea_get_runtime_context(
             "or ask the operator to update GITEA_MCP_PROFILE to a reviewer profile."
         )
 
-    preflight = assess_preflight_status()
-    if not preflight["preflight_ready"]:
-        safe_next_action = (
-            "Complete pre-flight verification before mutating: call gitea_whoami, then "
-            "gitea_resolve_task_capability for the intended task. "
-            f"Blocked: {'; '.join(preflight['preflight_block_reasons'])}"
-        )
-
     result = {
         "active_profile": profile["profile_name"],
         "authenticated_username": username,
@@ -4183,8 +3984,6 @@ def gitea_get_runtime_context(
         "review_merge_blocked_reasons": blocked_reasons,
         "suggested_fix": suggested_fix,
         "safe_next_action": safe_next_action,
-        "preflight_ready": preflight["preflight_ready"],
-        "preflight_block_reasons": preflight["preflight_block_reasons"],
     }
 
     if reveal and h:
@@ -4832,12 +4631,6 @@ def gitea_resolve_task_capability(
         required_permission, active_allowed, active_forbidden
     )
 
-    switching = gitea_config.is_runtime_switching_enabled()
-    available_in_session = allowed_in_current_session
-    configured = False
-    restart_required = False
-    reason_msg = None
-
     # Find matching configured profiles
     matching_profiles = []
     if config and "profiles" in config:
@@ -4862,19 +4655,17 @@ def gitea_resolve_task_capability(
 
     configured = len(matching_profiles) > 0
     available_in_session = allowed_in_current_session
+    restart_required = False
+    reason = ""
 
     if not allowed_in_current_session:
-        if configured and switching:
+        if configured:
             restart_required = True
-            available_in_session = False
+            reason = f"Reviewer profile exists but MCP server was added after session startup and is not attached."
-            reason_msg = (
+        else:
-                f"{required_role.capitalize()} profile exists but MCP server "
+            reason = f"No profile configured with permission '{required_permission}'."
-                "was added after session startup and is not attached."
+
-            )
+    switching = gitea_config.is_runtime_switching_enabled()
-        elif not configured:
-            reason_msg = (
-                f"No profile configured with permission '{required_permission}'."
-            )
     different_namespace_required = False
     next_safe_action = "None; ready for operations."
 
@@ -4952,19 +4743,17 @@ def gitea_resolve_task_capability(
         "active_identity": username,
         "active_profile_allowed_operations": active_allowed,
         "allowed_in_current_session": allowed_in_current_session,
-        "available_in_session": available_in_session,
+        "stop_required": stop_required,
-        "configured": configured,
-        "restart_required": restart_required,
-        "stop_required": stop_required or restart_required,
         "task_role_guidance": task_role_guidance,
         "matching_configured_profile": matching_profiles,
         "runtime_switching_supported": switching,
         "different_mcp_namespace_required": different_namespace_required,
         "exact_safe_next_action": next_safe_action,
+        "available_in_session": available_in_session,
+        "configured": configured,
+        "restart_required": restart_required,
+        "reason": reason,
     }
-    if reason_msg:
-        result["reason"] = reason_msg
-    role_session_router.sync_route_from_capability(result)
     was_terminal = capability_stop_terminal.is_active()
     terminal = capability_stop_terminal.sync_from_capability_result(result)
     if terminal:
@@ -4972,7 +4761,7 @@ def gitea_resolve_task_capability(
         result["terminal_report"] = (
             capability_stop_terminal.build_terminal_report(result)
         )
-    elif was_terminal and not (stop_required or restart_required):
+    elif was_terminal and not stop_required:
         result["cleared_stale_denial"] = True
     return result
 

mcp_discoverability.py

@@ -1,259 +0,0 @@
-#!/usr/bin/env python3
-"""MCP Discoverability Validation Tool for external servers (Issue #155)."""
-import os
-import sys
-import json
-import argparse
-import subprocess
-
-EXPECTED_JENKINS_TOOLS = {
-    "jenkins_whoami",
-    "jenkins_list_jobs",
-    "jenkins_latest_build",
-    "jenkins_build_status",
-    "jenkins_get_build",
-}
-
-EXPECTED_GLITCHTIP_TOOLS = {
-    "glitchtip_whoami",
-    "glitchtip_list_projects",
-    "glitchtip_list_unresolved",
-    "glitchtip_get_issue",
-    "glitchtip_recent_events",
-    "glitchtip_search",
-}
-
-RELOAD_INSTRUCTIONS = """
-=== MCP CLIENT RELOAD/RECONNECT RUNBOOK ===
-After registering or changing external MCP servers, reload your client to discover the new tools:
-- Codex: Click 'Reload Developer Tools' or restart the editor.
-- Gemini / Grok / ChatGPT Desktop: Restart the client or run the reload slash command if available.
-- Claude Desktop: Use 'Developer -> Reload' or restart the app.
-- General MCP Clients: Restart the process or reload the server config.
--------------------------------------------
-"""
-
-def parse_gitea_mcp_config(path):
-    if not path or not os.path.exists(path):
-        return {}
-    with open(path, "r", encoding="utf-8") as fh:
-        try:
-            return json.load(fh)
-        except Exception:
-            return {}
-
-def read_json_rpc_response(proc, req_id):
-    import time
-    start_time = time.time()
-    while time.time() - start_time < 5.0:
-        line = proc.stdout.readline()
-        if not line:
-            break
-        try:
-            data = json.loads(line)
-            if data.get("id") == req_id:
-                return data
-        except json.JSONDecodeError:
-            continue
-    return None
-
-def query_live_tools(command, args, env):
-    run_env = os.environ.copy()
-    if env:
-        run_env.update(env)
-
-    proc = subprocess.Popen(
-        [command] + args,
-        stdin=subprocess.PIPE,
-        stdout=subprocess.PIPE,
-        stderr=subprocess.PIPE,
-        env=run_env,
-        text=True,
-        bufsize=1
-    )
-
-    tools = []
-    try:
-        # 1. Send initialize
-        init_req = {
-            "jsonrpc": "2.0",
-            "id": 1,
-            "method": "initialize",
-            "params": {
-                "protocolVersion": "2024-11-05",
-                "capabilities": {},
-                "clientInfo": {"name": "mcp-discoverability-check", "version": "1.0.0"}
-            }
-        }
-        proc.stdin.write(json.dumps(init_req) + "\n")
-        proc.stdin.flush()
-
-        # Read init response
-        init_resp = read_json_rpc_response(proc, 1)
-        if init_resp:
-            # Send initialized notification
-            init_notif = {
-                "jsonrpc": "2.0",
-                "method": "notifications/initialized"
-            }
-            proc.stdin.write(json.dumps(init_notif) + "\n")
-            proc.stdin.flush()
-
-            # 2. Send tools/list
-            tools_req = {
-                "jsonrpc": "2.0",
-                "id": 2,
-                "method": "tools/list",
-                "params": {}
-            }
-            proc.stdin.write(json.dumps(tools_req) + "\n")
-            proc.stdin.flush()
-
-            tools_resp = read_json_rpc_response(proc, 2)
-            if tools_resp and "result" in tools_resp and "tools" in tools_resp["result"]:
-                for t in tools_resp["result"]["tools"]:
-                    tools.append(t["name"])
-    except Exception as e:
-        print(f"Error querying live tools: {e}", file=sys.stderr)
-    finally:
-        proc.terminate()
-        try:
-            proc.wait(timeout=2)
-        except Exception:
-            proc.kill()
-
-    return set(tools)
-
-def validate_mcp_client_config(client_config_path, gitea_config_path=None, live=False):
-    if not client_config_path or not os.path.exists(client_config_path):
-        print(f"SKIPPED: MCP client config not found at '{client_config_path}'", file=sys.stderr)
-        return True
-
-    with open(client_config_path, "r", encoding="utf-8") as fh:
-        try:
-            config_data = json.load(fh)
-        except Exception as e:
-            print(f"Error parsing client config: {e}", file=sys.stderr)
-            return False
-
-    mcp_servers = config_data.get("mcpServers", {})
-
-    # Check for stale server names
-    stale_names = {"jenkins-readonly", "glitchtip-readonly"}
-    for name in mcp_servers:
-        if name in stale_names:
-            print(f"ERROR: Stale server name '{name}' configured. Use canonical names 'jenkins-mcp' or 'glitchtip-mcp'.", file=sys.stderr)
-            return False
-
-    gitea_data = parse_gitea_mcp_config(gitea_config_path)
-    enabled_services = set()
-    contexts = gitea_data.get("contexts", {})
-
-    profile_name = os.environ.get("GITEA_MCP_PROFILE")
-    if profile_name and "profiles" in gitea_data:
-        profile = gitea_data["profiles"].get(profile_name)
-        if profile and "context" in profile:
-            ctx_name = profile["context"]
-            ctx = contexts.get(ctx_name, {})
-            if ctx.get("enabled"):
-                services = ctx.get("services", {})
-                for s_name, s_data in services.items():
-                    if s_data.get("enabled"):
-                        enabled_services.add(s_name)
-    else:
-        for ctx_name, ctx in contexts.items():
-            if ctx.get("enabled"):
-                services = ctx.get("services", {})
-                for s_name, s_data in services.items():
-                    if s_data.get("enabled"):
-                        enabled_services.add(s_name)
-
-    if not enabled_services:
-        print("No external services enabled in Gitea contexts. Discoverability check complete.", file=sys.stderr)
-        return True
-
-    success = True
-    for service in enabled_services:
-        canonical_name = f"{service}-mcp"
-        if canonical_name not in mcp_servers:
-            stale_match = f"{service}-readonly"
-            if stale_match in mcp_servers:
-                print(f"ERROR: Server '{canonical_name}' references stale name '{stale_match}' (fail closed).", file=sys.stderr)
-                success = False
-                continue
-            print(f"ERROR: Enabled service '{service}' is not registered under canonical name '{canonical_name}' in client config.", file=sys.stderr)
-            success = False
-            continue
-
-        server_conf = mcp_servers[canonical_name]
-        command = server_conf.get("command")
-        args = server_conf.get("args") or []
-        env = server_conf.get("env") or {}
-
-        if not command:
-            print(f"ERROR: Server '{canonical_name}' has no command configured.", file=sys.stderr)
-            success = False
-            continue
-
-        expected_module = f"{service}_mcp"
-        if "-m" not in args or expected_module not in args:
-            print(f"ERROR: Server '{canonical_name}' args do not point to expected module '{expected_module}' (args: {args}).", file=sys.stderr)
-            success = False
-            continue
-
-        profile_var = f"{service.upper()}_MCP_PROFILE"
-        config_var = f"{service.upper()}_MCP_CONFIG"
-        if profile_var not in env or config_var not in env:
-            print(f"ERROR: Server '{canonical_name}' env is missing required variables '{profile_var}' or '{config_var}'.", file=sys.stderr)
-            success = False
-            continue
-
-        if live:
-            tools = query_live_tools(command, args, env)
-            if not tools:
-                print("SKIPPED: server enabled but no usable tools visible", file=sys.stdout)
-                success = False
-                continue
-
-            expected = EXPECTED_JENKINS_TOOLS if service == "jenkins" else EXPECTED_GLITCHTIP_TOOLS
-            missing = expected - tools
-            if missing:
-                print(f"ERROR: Server '{canonical_name}' is missing expected tools: {', '.join(missing)}", file=sys.stderr)
-                success = False
-            else:
-                print(f"SUCCESS: Server '{canonical_name}' discoverability verified.", file=sys.stderr)
-        else:
-            print(f"SUCCESS: Server '{canonical_name}' static registration verified.", file=sys.stderr)
-
-    return success
-
-def main():
-    parser = argparse.ArgumentParser(description="MCP client registration discoverability checks.")
-    parser.add_argument("--client-config", help="Path to MCP client config JSON file.")
-    parser.add_argument("--gitea-config", help="Path to Gitea MCP config JSON file.")
-    parser.add_argument("--live", action="store_true", help="Perform live stdio checks on configured servers.")
-    parser.add_argument("--runbook", action="store_true", help="Print reload/reconnect guide runbook instructions.")
-
-    args = parser.parse_args()
-
-    if args.runbook:
-        print(RELOAD_INSTRUCTIONS)
-        return 0
-
-    if not args.client_config:
-        print("ERROR: --client-config must be specified.", file=sys.stderr)
-        return 2
-
-    ok = validate_mcp_client_config(
-        client_config_path=args.client_config,
-        gitea_config_path=args.gitea_config,
-        live=args.live
-    )
-
-    if not ok:
-        print(RELOAD_INSTRUCTIONS, file=sys.stderr)
-        return 1
-    return 0
-
-if __name__ == "__main__":
-    sys.exit(main())

mcp_server.py

@@ -6,12 +6,16 @@ Runs over stdio. All tools authenticate via macOS keychain (git credential fill)
 import os
 import sys
 
-from role_session_router import python_bytes_have_conflict_markers
-
-
 # Startup health check: scan all python files in the Gitea-Tools directory for unresolved conflict markers.
 def check_conflict_markers():
     dir_path = os.path.dirname(os.path.abspath(__file__))
+    # Construct conflict patterns dynamically so the loader does not match itself
+    conflict_patterns = [
+        b"<" * 7 + b" ",
+        b"=" * 7 + b"\n",
+        b"=" * 7 + b"\r\n",
+        b">" * 7 + b" "
+    ]
     for root, dirs, files in os.walk(dir_path):
         if any(p in root for p in ("venv", ".git", ".pytest_cache", "branches")):
             continue
@@ -20,7 +24,8 @@ def check_conflict_markers():
                 file_path = os.path.join(root, file)
                 try:
                     with open(file_path, "rb") as f:
-                        if python_bytes_have_conflict_markers(f.read()):
+                        content = f.read()
+                        if any(pattern in content for pattern in conflict_patterns):
                             rel_path = os.path.relpath(file_path, dir_path)
                             print(
                                 f"infra_stop: Unresolved merge conflict detected in {rel_path}. "

role_session_router.py

@@ -14,24 +14,6 @@ ROUTE_TO_REVIEWER = "route_to_reviewer_session"
 ROUTE_AMBIGUOUS = "ambiguous_task_stop"
 ROUTE_INFRA_STOP = "infra_stop"
 
-_CONFLICT_HEAD = b"<" * 7 + b" "
-_CONFLICT_TAIL = b">" * 7 + b" "
-_CONFLICT_SEPARATOR = b"=" * 7
-
-
-def python_bytes_have_conflict_markers(content: bytes) -> bool:
-    """Return True when *content* contains git merge-conflict marker lines."""
-    for line in content.splitlines():
-        stripped = line.rstrip(b"\r\n")
-        if stripped.startswith(_CONFLICT_HEAD):
-            return True
-        if stripped.startswith(_CONFLICT_TAIL):
-            return True
-        if stripped == _CONFLICT_SEPARATOR:
-            return True
-    return False
-
-
 REVIEWER_TASKS = frozenset({
     "review_pr",
     "merge_pr",
@@ -212,56 +194,11 @@ def _record_route(result: dict):
     _session_last_route = dict(result)
 
 
-def sync_route_from_capability(capability: dict) -> None:
-    """Align sticky route state with operation-scoped capability resolution (#228)."""
-    capability = capability or {}
-    task = (capability.get("requested_task") or "").strip()
-    required_role = capability.get("required_role_kind")
-    if not task or not required_role:
-        return
-    if capability.get("allowed_in_current_session"):
-        _record_route({
-            "task_type": task,
-            "required_role": required_role,
-            "active_role": required_role,
-            "active_profile": capability.get("active_profile"),
-            "route_result": ROUTE_ALLOWED,
-            "downstream_allowed": True,
-            "reasons": [],
-            "message": (
-                f"Operation-scoped task '{task}' resolved for current session; "
-                "proceed."
-            ),
-        })
-        return
-    if required_role == "reviewer" and capability.get("stop_required"):
-        _record_route({
-            "task_type": task,
-            "required_role": required_role,
-            "active_role": capability.get("required_role_kind"),
-            "active_profile": capability.get("active_profile"),
-            "route_result": ROUTE_WRONG_ROLE,
-            "downstream_allowed": False,
-            "reasons": [WRONG_ROLE_REVIEWER_MSG],
-            "message": WRONG_ROLE_REVIEWER_MSG,
-        })
-
-
 def check_author_mutation_after_reviewer_stop(mutation_task: str) -> tuple[bool, list[str]]:
-    """Block author-side fallback after a reviewer wrong_role_stop (#206).
+    """Block author-side fallback after a reviewer wrong_role_stop (#206)."""
-
-    An explicit operation-scoped author capability resolution for the same
-    *mutation_task* clears the sticky reviewer denial (#228).
-    """
     last = _session_last_route
     if not last:
         return True, []
-    if (
-        last.get("route_result") == ROUTE_ALLOWED
-        and last.get("task_type") == mutation_task
-        and last.get("required_role") == "author"
-    ):
-        return True, []
     if last.get("route_result") != ROUTE_WRONG_ROLE:
         return True, []
     if last.get("required_role") != "reviewer":
@@ -270,32 +207,13 @@ def check_author_mutation_after_reviewer_stop(mutation_task: str) -> tuple[bool,
         return False, [
             WRONG_ROLE_REVIEWER_MSG,
             "Author-side mutations are blocked after a reviewer-task "
-            "wrong_role_stop unless the operator explicitly resolves the "
+            "wrong_role_stop unless the operator explicitly changes the "
-            "author task via gitea_resolve_task_capability.",
+            "task and relaunches an author MCP session.",
             f"Attempted fallback mutation: {mutation_task}",
         ]
     return True, []
 
 
-def first_conflict_marker_path(project_root: str | None = None) -> str | None:
-    """Return the first .py path containing a git conflict marker, or None."""
-    root_dir = project_root or os.path.dirname(os.path.abspath(__file__))
-    for root, dirs, files in os.walk(root_dir):
-        if any(p in root for p in ("venv", ".git", ".pytest_cache", "branches")):
-            continue
-        for file in files:
-            if not file.endswith(".py"):
-                continue
-            file_path = os.path.join(root, file)
-            try:
-                with open(file_path, "rb") as f:
-                    if python_bytes_have_conflict_markers(f.read()):
-                        return file_path
-            except OSError:
-                pass
-    return None
-
-
 def check_mid_merge() -> bool:
     """Return True if the repository is mid-merge, mid-rebase, or has conflict markers."""
     project_root = os.path.dirname(os.path.abspath(__file__))
@@ -306,4 +224,24 @@ def check_mid_merge() -> bool:
                 or os.path.exists(os.path.join(git_dir, "rebase-apply"))):
             return True
 
-    return first_conflict_marker_path(project_root) is not None
+    # Scan python files for conflict markers
+    conflict_patterns = [
+        b"<" * 7 + b" ",
+        b"=" * 7 + b"\n",
+        b"=" * 7 + b"\r\n",
+        b">" * 7 + b" "
+    ]
+    for root, dirs, files in os.walk(project_root):
+        if any(p in root for p in ("venv", ".git", ".pytest_cache", "branches")):
+            continue
+        for file in files:
+            if file.endswith(".py"):
+                file_path = os.path.join(root, file)
+                try:
+                    with open(file_path, "rb") as f:
+                        content = f.read()
+                        if any(pattern in content for pattern in conflict_patterns):
+                            return True
+                except Exception:
+                    pass
+    return False

tests/test_health.py

@@ -5,7 +5,6 @@ import unittest
 from unittest.mock import patch
 
 import role_session_router
-from role_session_router import python_bytes_have_conflict_markers
 from mcp_server import gitea_route_task_session, gitea_resolve_task_capability
 
 _HEALTH_SUBPROCESS_TIMEOUT_SEC = 30
@@ -81,20 +80,6 @@ class TestMCPHealth(unittest.TestCase):
             with self.assertRaises(unittest.SkipTest):
                 _health_test_python()
 
-    def test_conflict_marker_helper_ignores_decorative_equals_border(self):
-        sample = b'banner = """\n===========================================\n"""\n'
-        self.assertFalse(python_bytes_have_conflict_markers(sample))
-
-    def test_conflict_marker_helper_detects_real_markers(self):
-        sample = (
-            b"<" * 7 + b" HEAD\n"
-            b"print('hello')\n"
-            b"=" * 7 + b"\n"
-            b"print('world')\n"
-            b">" * 7 + b" main\n"
-        )
-        self.assertTrue(python_bytes_have_conflict_markers(sample))
-
     def test_startup_conflict_detection(self):
         # Create a Python file with conflict markers constructed dynamically
         with open(self.temp_file, "w") as f:

tests/test_mcp_discoverability.py

@@ -1,260 +0,0 @@
-"""Tests for MCP discoverability validation (Issue #155)."""
-import os
-import sys
-import json
-import tempfile
-import unittest
-from unittest.mock import patch, MagicMock
-
-sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
-
-from mcp_discoverability import (
-    validate_mcp_client_config,
-    RELOAD_INSTRUCTIONS,
-    EXPECTED_JENKINS_TOOLS,
-    EXPECTED_GLITCHTIP_TOOLS,
-)
-
-class TestMcpDiscoverability(unittest.TestCase):
-    def setUp(self):
-        self.tmp_dir = tempfile.TemporaryDirectory()
-        self.gitea_config_path = os.path.join(self.tmp_dir.name, "gitea-mcp.json")
-        self.client_config_path = os.path.join(self.tmp_dir.name, "claude_desktop_config.json")
-
-    def tearDown(self):
-        self.tmp_dir.cleanup()
-
-    def _write_json(self, path, data):
-        with open(path, "w", encoding="utf-8") as fh:
-            json.dump(data, fh)
-
-    def _v2_gitea_config(self, jenkins_enabled=True, glitchtip_enabled=False):
-        return {
-            "version": 2,
-            "contexts": {
-                "prod": {
-                    "enabled": True,
-                    "services": {
-                        "jenkins": {
-                            "enabled": jenkins_enabled,
-                            "kind": "jenkins",
-                            "capabilities": ["read"]
-                        },
-                        "glitchtip": {
-                            "enabled": glitchtip_enabled,
-                            "kind": "glitchtip",
-                            "capabilities": ["read"]
-                        }
-                    }
-                }
-            }
-        }
-
-    def test_static_validation_success(self):
-        g_cfg = self._v2_gitea_config(jenkins_enabled=True, glitchtip_enabled=True)
-        self._write_json(self.gitea_config_path, g_cfg)
-
-        c_cfg = {
-            "mcpServers": {
-                "jenkins-mcp": {
-                    "command": "python3",
-                    "args": ["-m", "jenkins_mcp"],
-                    "env": {
-                        "JENKINS_MCP_PROFILE": "jenkins-readonly",
-                        "JENKINS_MCP_CONFIG": "/path/to/config.json"
-                    }
-                },
-                "glitchtip-mcp": {
-                    "command": "python3",
-                    "args": ["-m", "glitchtip_mcp"],
-                    "env": {
-                        "GLITCHTIP_MCP_PROFILE": "glitchtip-readonly",
-                        "GLITCHTIP_MCP_CONFIG": "/path/to/config.json"
-                    }
-                }
-            }
-        }
-        self._write_json(self.client_config_path, c_cfg)
-
-        res = validate_mcp_client_config(
-            client_config_path=self.client_config_path,
-            gitea_config_path=self.gitea_config_path,
-            live=False
-        )
-        self.assertTrue(res)
-
-    def test_stale_server_name_rejected(self):
-        g_cfg = self._v2_gitea_config(jenkins_enabled=True)
-        self._write_json(self.gitea_config_path, g_cfg)
-
-        # Uses stale key in client config
-        c_cfg = {
-            "mcpServers": {
-                "jenkins-readonly": {
-                    "command": "python3",
-                    "args": ["-m", "jenkins_mcp"],
-                    "env": {
-                        "JENKINS_MCP_PROFILE": "jenkins-readonly",
-                        "JENKINS_MCP_CONFIG": "/path/to/config.json"
-                    }
-                }
-            }
-        }
-        self._write_json(self.client_config_path, c_cfg)
-
-        res = validate_mcp_client_config(
-            client_config_path=self.client_config_path,
-            gitea_config_path=self.gitea_config_path,
-            live=False
-        )
-        self.assertFalse(res)
-
-    def test_incorrect_arguments_rejected(self):
-        g_cfg = self._v2_gitea_config(jenkins_enabled=True)
-        self._write_json(self.gitea_config_path, g_cfg)
-
-        # Missing -m or wrong module
-        c_cfg = {
-            "mcpServers": {
-                "jenkins-mcp": {
-                    "command": "python3",
-                    "args": ["wrong_runner.py"],
-                    "env": {
-                        "JENKINS_MCP_PROFILE": "jenkins-readonly",
-                        "JENKINS_MCP_CONFIG": "/path/to/config.json"
-                    }
-                }
-            }
-        }
-        self._write_json(self.client_config_path, c_cfg)
-
-        res = validate_mcp_client_config(
-            client_config_path=self.client_config_path,
-            gitea_config_path=self.gitea_config_path,
-            live=False
-        )
-        self.assertFalse(res)
-
-    def test_missing_required_env_rejected(self):
-        g_cfg = self._v2_gitea_config(jenkins_enabled=True)
-        self._write_json(self.gitea_config_path, g_cfg)
-
-        # Missing JENKINS_MCP_PROFILE env variable
-        c_cfg = {
-            "mcpServers": {
-                "jenkins-mcp": {
-                    "command": "python3",
-                    "args": ["-m", "jenkins_mcp"],
-                    "env": {
-                        "JENKINS_MCP_CONFIG": "/path/to/config.json"
-                    }
-                }
-            }
-        }
-        self._write_json(self.client_config_path, c_cfg)
-
-        res = validate_mcp_client_config(
-            client_config_path=self.client_config_path,
-            gitea_config_path=self.gitea_config_path,
-            live=False
-        )
-        self.assertFalse(res)
-
-    @patch("subprocess.Popen")
-    def test_live_check_verification_success(self, mock_popen):
-        g_cfg = self._v2_gitea_config(jenkins_enabled=True)
-        self._write_json(self.gitea_config_path, g_cfg)
-
-        c_cfg = {
-            "mcpServers": {
-                "jenkins-mcp": {
-                    "command": "python3",
-                    "args": ["-m", "jenkins_mcp"],
-                    "env": {
-                        "JENKINS_MCP_PROFILE": "jenkins-readonly",
-                        "JENKINS_MCP_CONFIG": "/path/to/config.json"
-                    }
-                }
-            }
-        }
-        self._write_json(self.client_config_path, c_cfg)
-
-        # Mock stdout lines for JSON-RPC
-        mock_proc = MagicMock()
-        mock_popen.return_value = mock_proc
-
-        init_resp = json.dumps({"jsonrpc": "2.0", "id": 1, "result": {"protocolVersion": "2024-11-05"}})
-        tools_resp = json.dumps({
-            "jsonrpc": "2.0",
-            "id": 2,
-            "result": {
-                "tools": [{"name": tool} for tool in EXPECTED_JENKINS_TOOLS]
-            }
-        })
-
-        mock_proc.stdout.readline.side_effect = [
-            init_resp + "\n",
-            tools_resp + "\n",
-            ""
-        ]
-
-        res = validate_mcp_client_config(
-            client_config_path=self.client_config_path,
-            gitea_config_path=self.gitea_config_path,
-            live=True
-        )
-        self.assertTrue(res)
-
-    @patch("subprocess.Popen")
-    def test_live_check_empty_toolset_rejected(self, mock_popen):
-        g_cfg = self._v2_gitea_config(jenkins_enabled=True)
-        self._write_json(self.gitea_config_path, g_cfg)
-
-        c_cfg = {
-            "mcpServers": {
-                "jenkins-mcp": {
-                    "command": "python3",
-                    "args": ["-m", "jenkins_mcp"],
-                    "env": {
-                        "JENKINS_MCP_PROFILE": "jenkins-readonly",
-                        "JENKINS_MCP_CONFIG": "/path/to/config.json"
-                    }
-                }
-            }
-        }
-        self._write_json(self.client_config_path, c_cfg)
-
-        mock_proc = MagicMock()
-        mock_popen.return_value = mock_proc
-
-        init_resp = json.dumps({"jsonrpc": "2.0", "id": 1, "result": {}})
-        tools_resp = json.dumps({
-            "jsonrpc": "2.0",
-            "id": 2,
-            "result": {
-                "tools": []  # empty tool list
-            }
-        })
-
-        mock_proc.stdout.readline.side_effect = [
-            init_resp + "\n",
-            tools_resp + "\n",
-            ""
-        ]
-
-        # Should return False (coverage fails) when no tools are exposed
-        res = validate_mcp_client_config(
-            client_config_path=self.client_config_path,
-            gitea_config_path=self.gitea_config_path,
-            live=True
-        )
-        self.assertFalse(res)
-
-    def test_runbook_instructions_content(self):
-        self.assertIn("MCP CLIENT RELOAD/RECONNECT RUNBOOK", RELOAD_INSTRUCTIONS)
-        self.assertIn("Codex", RELOAD_INSTRUCTIONS)
-        self.assertIn("Gemini", RELOAD_INSTRUCTIONS)
-        self.assertIn("Claude Desktop", RELOAD_INSTRUCTIONS)
-
-if __name__ == "__main__":
-    unittest.main()

tests/test_mcp_server.py

@@ -3158,12 +3158,6 @@ class TestPreflightVerification(unittest.TestCase):
         self.orig_whoami_violation = mcp_server._preflight_whoami_violation
         self.orig_capability_violation = mcp_server._preflight_capability_violation
         self.orig_resolved_role = mcp_server._preflight_resolved_role
-        self.orig_process_start = mcp_server._process_start_porcelain
-        self.orig_whoami_baseline = mcp_server._preflight_whoami_baseline_porcelain
-        self.orig_capability_baseline = mcp_server._preflight_capability_baseline_porcelain
-        self.orig_whoami_files = mcp_server._preflight_whoami_violation_files
-        self.orig_capability_files = mcp_server._preflight_capability_violation_files
-        self.orig_reviewer_files = mcp_server._preflight_reviewer_violation_files
 
         # Reset state for each test
         mcp_server._preflight_whoami_called = False
@@ -3171,13 +3165,6 @@ class TestPreflightVerification(unittest.TestCase):
         mcp_server._preflight_whoami_violation = False
         mcp_server._preflight_capability_violation = False
         mcp_server._preflight_resolved_role = None
-        mcp_server._process_start_porcelain = ""
-        mcp_server._preflight_whoami_baseline_porcelain = None
-        mcp_server._preflight_capability_baseline_porcelain = None
-        mcp_server._preflight_whoami_violation_files = []
-        mcp_server._preflight_capability_violation_files = []
-        mcp_server._preflight_reviewer_violation_files = []
-        os.environ["GITEA_TEST_PORCELAIN"] = ""
 
     def tearDown(self):
         # Restore real global variables
@@ -3187,42 +3174,30 @@ class TestPreflightVerification(unittest.TestCase):
         mcp_server._preflight_whoami_violation = self.orig_whoami_violation
         mcp_server._preflight_capability_violation = self.orig_capability_violation
         mcp_server._preflight_resolved_role = self.orig_resolved_role
-        mcp_server._process_start_porcelain = self.orig_process_start
+        if "GITEA_TEST_FORCE_DIRTY" in os.environ:
-        mcp_server._preflight_whoami_baseline_porcelain = self.orig_whoami_baseline
+            del os.environ["GITEA_TEST_FORCE_DIRTY"]
-        mcp_server._preflight_capability_baseline_porcelain = self.orig_capability_baseline
-        mcp_server._preflight_whoami_violation_files = self.orig_whoami_files
-        mcp_server._preflight_capability_violation_files = self.orig_capability_files
-        mcp_server._preflight_reviewer_violation_files = self.orig_reviewer_files
-        for key in ("GITEA_TEST_FORCE_DIRTY", "GITEA_TEST_PORCELAIN"):
-            if key in os.environ:
-                del os.environ[key]
 
     def test_preflight_whoami_violation(self):
         import mcp_server
         os.environ["GITEA_TEST_FORCE_DIRTY"] = "1"
+        mcp_server._preflight_capability_called = True
         mcp_server.record_preflight_check("whoami")
         self.assertTrue(mcp_server._preflight_whoami_violation)
-        mcp_server.record_preflight_check("capability", resolved_role="author")
 
         with self.assertRaises(RuntimeError) as ctx:
             mcp_server.verify_preflight_purity()
         self.assertIn("Workspace file edits occurred before gitea_whoami verification", str(ctx.exception))
-        self.assertIn("Offending files:", str(ctx.exception))
 
     def test_preflight_capability_violation(self):
         import mcp_server
-        mcp_server.record_preflight_check("whoami")
         os.environ["GITEA_TEST_FORCE_DIRTY"] = "1"
+        mcp_server._preflight_whoami_called = True
         mcp_server.record_preflight_check("capability", resolved_role="author")
         self.assertTrue(mcp_server._preflight_capability_violation)
 
         with self.assertRaises(RuntimeError) as ctx:
             mcp_server.verify_preflight_purity()
-        self.assertIn(
+        self.assertIn("Workspace file edits occurred before gitea_resolve_task_capability verification", str(ctx.exception))
-            "Workspace file edits occurred before gitea_resolve_task_capability verification",
-            str(ctx.exception),
-        )
-        self.assertIn("Offending files:", str(ctx.exception))
 
     def test_preflight_not_called_fails_closed(self):
         import mcp_server
@@ -3238,48 +3213,16 @@ class TestPreflightVerification(unittest.TestCase):
 
     def test_preflight_reviewer_mutation_violation(self):
         import mcp_server
-        mcp_server.record_preflight_check("whoami")
+        mcp_server._preflight_whoami_called = True
-        mcp_server.record_preflight_check("capability", resolved_role="reviewer")
+        mcp_server._preflight_capability_called = True
+        mcp_server._preflight_resolved_role = "reviewer"
 
-        # Session-owned reviewer edits after capability are blocked.
+        # When dirty, reviewer edits are blocked
-        os.environ["GITEA_TEST_PORCELAIN"] = " M reviewer_edit.py\n"
+        os.environ["GITEA_TEST_FORCE_DIRTY"] = "1"
         with self.assertRaises(RuntimeError) as ctx:
             mcp_server.verify_preflight_purity()
         self.assertIn("Reviewer profile is forbidden from modifying tracked workspace files", str(ctx.exception))
-        self.assertIn("reviewer_edit.py", str(ctx.exception))
-
-        # Foreign pre-existing dirty state does not block when unchanged.
-        os.environ["GITEA_TEST_PORCELAIN"] = ""
-        mcp_server.verify_preflight_purity()
-
-    def test_foreign_workspace_edits_do_not_block_clean_reviewer(self):
-        """#252: concurrent-session dirt in the shared worktree is not attributed."""
-        import mcp_server
-        mcp_server._process_start_porcelain = " M foreign_author.py\n"
-        os.environ["GITEA_TEST_PORCELAIN"] = " M foreign_author.py\n"
-        mcp_server.record_preflight_check("whoami")
-        self.assertFalse(mcp_server._preflight_whoami_violation)
-        mcp_server.record_preflight_check("capability", resolved_role="reviewer")
-        self.assertFalse(mcp_server._preflight_capability_violation)
-        mcp_server.verify_preflight_purity()
-
-    def test_fresh_whoami_clears_sticky_violation(self):
-        """#252: re-running whoami re-evaluates instead of replaying sticky state."""
-        import mcp_server
-        os.environ["GITEA_TEST_FORCE_DIRTY"] = "1"
-        mcp_server.record_preflight_check("whoami")
-        self.assertTrue(mcp_server._preflight_whoami_violation)
 
+        # When clean, reviewer is allowed
         del os.environ["GITEA_TEST_FORCE_DIRTY"]
-        os.environ["GITEA_TEST_PORCELAIN"] = ""
-        mcp_server.record_preflight_check("whoami")
-        self.assertFalse(mcp_server._preflight_whoami_violation)
-        mcp_server.record_preflight_check("capability", resolved_role="reviewer")
         mcp_server.verify_preflight_purity()
-
-    def test_runtime_context_matches_preflight_block(self):
-        """#252: safe_next_action must not claim ready while pre-flight blocks."""
-        import mcp_server
-        status = mcp_server.assess_preflight_status()
-        self.assertFalse(status["preflight_ready"])
-        self.assertIn("gitea_whoami", status["preflight_block_reasons"][0])

tests/test_operation_scoped_roles.py

@@ -1,181 +0,0 @@
-"""Tests for operation-scoped role selection and automatic dispatch switching (#228)."""
-import os
-import sys
-import json
-import tempfile
-import unittest
-from unittest.mock import patch, MagicMock
-
-sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
-
-import gitea_config
-import gitea_auth
-import mcp_server
-from reviewer_worktree import assess_author_worktree_continuity
-
-CONFIG_TEST = {
-    "version": 2,
-    "contexts": {
-        "ctx": {
-            "enabled": True,
-            "gitea": {
-                "enabled": True,
-                "base_url": "https://gitea.example.com"
-            }
-        }
-    },
-    "profiles": {
-        "author-profile": {
-            "enabled": True,
-            "context": "ctx",
-            "role": "author",
-            "username": "author-user",
-            "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"],
-            "execution_profile": "author-profile"
-        },
-        "reviewer-profile": {
-            "enabled": True,
-            "context": "ctx",
-            "role": "reviewer",
-            "username": "reviewer-user",
-            "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"],
-            "execution_profile": "reviewer-profile"
-        }
-    }
-}
-
-
-class TestOperationScopedRoles(unittest.TestCase):
-    def setUp(self):
-        self._remotes_patch = patch.dict(mcp_server.REMOTES, {
-            "dadeschools": {"host": "gitea.example.com", "org": "Example-Org", "repo": "Example-Repo"},
-            "prgs": {"host": "gitea.example.com", "org": "Example-Org", "repo": "Example-Repo"}
-        })
-        self._remotes_patch.start()
-        mcp_server._IDENTITY_CACHE.clear()
-        gitea_config._active_profile_override = None
-        mcp_server._MUTATION_AUTHORITY = None
-        self._dir = tempfile.TemporaryDirectory()
-        self.config_path = os.path.join(self._dir.name, "profiles.json")
-        self._write_config(CONFIG_TEST)
-
-    def tearDown(self):
-        self._remotes_patch.stop()
-        mcp_server._IDENTITY_CACHE.clear()
-        gitea_config._active_profile_override = None
-        mcp_server._MUTATION_AUTHORITY = None
-        self._dir.cleanup()
-
-    def _write_config(self, obj):
-        with open(self.config_path, "w", encoding="utf-8") as fh:
-            fh.write(json.dumps(obj))
-
-    def _env(self, profile="author-profile", with_reviewer_token=True):
-        env = {
-            "GITEA_MCP_CONFIG": self.config_path,
-            "GITEA_MCP_PROFILE": profile,
-            "GITEA_TOKEN_AUTHOR": "author-pass",
-        }
-        if with_reviewer_token:
-            env["GITEA_TOKEN_REVIEWER"] = "reviewer-pass"
-        return env
-
-    @patch("mcp_server.api_request")
-    def test_auto_switch_to_reviewer(self, mock_api):
-        # mock identity resolution to return username matching profile
-        mock_api.side_effect = lambda method, url, header: (
-            {"login": "reviewer-user"} if "reviewer-pass" in str(header) else {"login": "author-user"}
-        )
-        with patch.dict(os.environ, self._env("author-profile")):
-            # initially we are author-profile
-            self.assertEqual(gitea_config.selected_profile_name(), "author-profile")
-            self.assertEqual(mcp_server.get_profile()["profile_name"], "author-profile")
-
-            # resolve a reviewer task (review_pr)
-            res = mcp_server.gitea_resolve_task_capability(task="review_pr", remote="prgs")
-            
-            # 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")
-
-    @patch("mcp_server.api_request")
-    def test_auto_switch_to_author(self, mock_api):
-        mock_api.side_effect = lambda method, url, header: (
-            {"login": "reviewer-user"} if "reviewer-pass" in str(header) else {"login": "author-user"}
-        )
-        with patch.dict(os.environ, self._env("reviewer-profile")):
-            # initially we are reviewer-profile
-            self.assertEqual(gitea_config.selected_profile_name(), "reviewer-profile")
-
-            # resolve an author task (create_issue)
-            res = mcp_server.gitea_resolve_task_capability(task="create_issue", remote="prgs")
-            
-            # 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")
-
-    @patch("mcp_server.api_request")
-    def test_restart_required_when_unattached(self, mock_api):
-        mock_api.side_effect = lambda method, url, header: {"login": "author-user"}
-        # launch without reviewer token in env
-        with patch.dict(os.environ, self._env("author-profile", with_reviewer_token=False)):
-            self.assertEqual(gitea_config.selected_profile_name(), "author-profile")
-
-            # resolve a reviewer task (review_pr)
-            res = mcp_server.gitea_resolve_task_capability(task="review_pr", remote="prgs")
-            
-            # verify it did NOT switch and reports restart_required
-            self.assertFalse(res["allowed_in_current_session"])
-            self.assertFalse(res["available_in_session"])
-            self.assertTrue(res["configured"])
-            self.assertTrue(res["restart_required"])
-            self.assertTrue(res["stop_required"])
-            self.assertIn("Reviewer profile exists but MCP server was added after session startup and is not attached", res["reason"])
-
-    def test_author_continuity_dirty_worktree(self):
-        # author is allowed to keep dirty worktree
-        res = assess_author_worktree_continuity({
-            "task_role": "author",
-            "dirty_files": ["review_proofs.py"],
-        })
-        self.assertTrue(res["allowed"])
-
-        # reviewer is blocked by reviewer worktree proof
-        res2 = assess_author_worktree_continuity({
-            "task_role": "reviewer",
-            "worktree_path": "/repo",
-            "dirty_files": ["review_proofs.py"],
-            "pr_scope_files": ["docs/wiki/Repositories.md"],
-            "scratch_used": False,
-        })
-        self.assertFalse(res2["proven"])
-
-    @patch("mcp_server.api_request")
-    def test_mutating_actions_auto_switch(self, mock_api):
-        mock_api.side_effect = lambda method, url, header: (
-            {"login": "reviewer-user"} if "reviewer-pass" in str(header) else {"login": "author-user"}
-        )
-        with patch.dict(os.environ, self._env("author-profile")):
-            # verify we are author-profile
-            self.assertEqual(gitea_config.selected_profile_name(), "author-profile")
-
-            # 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)
-
-            # verify we dynamically switched to reviewer-profile
-            self.assertEqual(gitea_config.selected_profile_name(), "reviewer-profile")
-
-
-if __name__ == "__main__":
-    unittest.main()

tests/test_role_session_router.py

@@ -119,25 +119,6 @@ class TestRoleSessionRouter(unittest.TestCase):
         ]
         self.assertEqual(issue_posts, [])
 
-    @patch("mcp_server.api_get_all", return_value=[])
-    @patch("mcp_server.api_request", return_value={"number": 888})
-    @patch("mcp_server.get_auth_header", return_value="token author-pass")
-    def test_author_task_allowed_after_capability_resolve(
-        self, _auth, _api, _get_all
-    ):
-        """#228: explicit create_issue capability clears reviewer wrong_role_stop."""
-        with patch.dict(os.environ, self._env("prgs-author")):
-            mcp_server.gitea_route_task_session(task_type="review_pr", remote="prgs")
-            cap = mcp_server.gitea_resolve_task_capability(
-                task="create_issue", remote="prgs"
-            )
-            self.assertTrue(cap["allowed_in_current_session"])
-            result = mcp_server.gitea_create_issue(
-                title="operation-scoped author recovery",
-                remote="prgs",
-            )
-        self.assertEqual(result.get("number"), 888)
-
     @patch("mcp_server.api_get_all", return_value=[])
     @patch("mcp_server.api_request", return_value={"number": 999})
     @patch("mcp_server.get_auth_header", return_value="token author-pass")
@@ -187,6 +168,74 @@ class TestRoleSessionRouter(unittest.TestCase):
         self.assertFalse(result["success"])
         self.assertIn("switching is disabled", result["message"])
 
+    def _switching_env(self, profile):
+        switching_config = dict(CONFIG)
+        switching_config["rules"] = {"allow_runtime_switching": True}
+        switching_config_path = os.path.join(self._dir.name, "profiles_switching.json")
+        with open(switching_config_path, "w", encoding="utf-8") as fh:
+            fh.write(json.dumps(switching_config))
+        return {
+            "GITEA_MCP_CONFIG": switching_config_path,
+            "GITEA_MCP_PROFILE": profile,
+            "GITEA_TOKEN_AUTHOR": "author-pass",
+            "GITEA_TOKEN_REVIEWER": "reviewer-pass",
+        }
+
+    @patch("mcp_server.api_request")
+    @patch("mcp_server.get_auth_header", return_value="token reviewer-pass")
+    def test_dynamic_switch_to_reviewer_profile(self, _auth, mock_api):
+        mock_api.side_effect = lambda method, url, header: (
+            {"login": "sysadmin"} if "reviewer-pass" in str(header) else {"login": "jcwalker3"}
+        )
+        with patch.dict(os.environ, self._switching_env("prgs-author")):
+            self.assertEqual(gitea_config.selected_profile_name(), "prgs-author")
+            route = mcp_server.gitea_route_task_session(
+                task_type="review_pr", remote="prgs"
+            )
+            self.assertEqual(route["route_result"], role_session_router.ROUTE_ALLOWED)
+            self.assertTrue(route["downstream_allowed"])
+            self.assertEqual(gitea_config.selected_profile_name(), "prgs-reviewer")
+
+    @patch("mcp_server.api_request")
+    @patch("mcp_server.get_auth_header", return_value="token author-pass")
+    def test_dynamic_switch_to_author_profile(self, _auth, mock_api):
+        mock_api.side_effect = lambda method, url, header: (
+            {"login": "jcwalker3"} if "author-pass" in str(header) else {"login": "sysadmin"}
+        )
+        with patch.dict(os.environ, self._switching_env("prgs-reviewer")):
+            self.assertEqual(gitea_config.selected_profile_name(), "prgs-reviewer")
+            route = mcp_server.gitea_route_task_session(
+                task_type="create_issue", remote="prgs"
+            )
+            self.assertEqual(route["route_result"], role_session_router.ROUTE_ALLOWED)
+            self.assertTrue(route["downstream_allowed"])
+            self.assertEqual(gitea_config.selected_profile_name(), "prgs-author")
+
+    def test_dynamic_switch_blocked_for_wrong_role(self):
+        with patch.dict(os.environ, self._switching_env("prgs-author")):
+            config_data = dict(CONFIG)
+            config_data["rules"] = {"allow_runtime_switching": True}
+            config_data["profiles"]["prgs-author"]["allowed_operations"] = [
+                "gitea.read", "gitea.issue.create", "gitea.pr.create", "gitea.branch.push"
+            ]
+            custom_path = os.path.join(self._dir.name, "profiles_wrong_role.json")
+            with open(custom_path, "w", encoding="utf-8") as fh:
+                fh.write(json.dumps(config_data))
+
+            env = {
+                "GITEA_MCP_CONFIG": custom_path,
+                "GITEA_MCP_PROFILE": "prgs-author",
+                "GITEA_TOKEN_AUTHOR": "author-pass",
+                "GITEA_TOKEN_REVIEWER": "reviewer-pass",
+            }
+            with patch.dict(os.environ, env):
+                route = mcp_server.gitea_route_task_session(
+                    task_type="comment_issue", remote="prgs"
+                )
+                self.assertEqual(route["route_result"], role_session_router.ROUTE_TO_AUTHOR)
+                self.assertFalse(route["downstream_allowed"])
+                self.assertEqual(gitea_config.selected_profile_name(), "prgs-author")
+
     def test_handoff_requires_route_fields(self):
         incomplete = assess_role_route_handoff("Task done.")
         self.assertFalse(incomplete["complete"])
@@ -204,25 +253,5 @@ class TestRoleSessionRouter(unittest.TestCase):
         self.assertTrue(complete["complete"])
 
 
-class TestCheckMidMerge(unittest.TestCase):
-    def test_decorative_equals_banner_is_not_mid_merge(self):
-        self.assertFalse(role_session_router.check_mid_merge())
-
-    def test_python_bytes_have_conflict_markers_rejects_decorative_equals(self):
-        banner = b"===========================================\n"
-        self.assertFalse(role_session_router.python_bytes_have_conflict_markers(banner))
-
-    def test_python_bytes_have_conflict_markers_detects_real_markers(self):
-        self.assertTrue(
-            role_session_router.python_bytes_have_conflict_markers(b"<<<<<<< HEAD\n")
-        )
-        self.assertTrue(
-            role_session_router.python_bytes_have_conflict_markers(b"=======\n")
-        )
-        self.assertTrue(
-            role_session_router.python_bytes_have_conflict_markers(b">>>>>>> topic\n")
-        )
-
-
 if __name__ == "__main__":
     unittest.main()