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

docs/architecture/jenkins-readonly-build-status-design.md

@@ -5,7 +5,7 @@
 - **Related:** #77 (repo/branch/PR → job mapping, designed separately)
 - **Date:** 2026-07-02
 
-Note on naming: This design used historical `jenkins-readonly` skill name in Gitea-Tools. Actual package/server is `jenkins-mcp` (see mcp-control-plane registration in #55). The read server boundary remains read-only; gated triggers live on the separate `jenkins-write-mcp` / `jenkins_mcp.write_server` boundary (see #56 / #152).
+Note on naming: This design used historical `jenkins-readonly` skill name in Gitea-Tools. Actual package/server is `jenkins-mcp` (see mcp-control-plane registration in #55). The server boundary now contains gated trigger (see #56), but read tools remain as designed.
 Client registration and reload instructions live in
 [`../mcp-client-registration.md`](../mcp-client-registration.md).
 
@@ -18,9 +18,7 @@ detail (build URL, number, timing, result) to report or investigate.
 Phase 1 is **primarily read-only**, per ADR-0001
 ([`adr-0001-mcp-control-plane-boundaries.md`](adr-0001-mcp-control-plane-boundaries.md)):
 
-- Build triggers are outside this read-only surface and require the separate
+- Build triggers are gated behind dedicated profile + exact confirmation (landed in #4, boundary correction in #56).
-  `jenkins-write-mcp` boundary, a dedicated profile, exact confirmation, and
-  fail-closed mutation audit (landed in #4, boundary correction in #56 / #152).
 - **Excluded: deploy triggers.**
 - **Excluded: parameterized job launches.**
 - Excluded: job creation/deletion/config changes, queue manipulation, node
@@ -111,7 +109,7 @@ by #76):
   `forbidden_operations: ["jenkins.build.trigger", "jenkins.deploy", "jenkins.job.configure"]`
   as belt-and-braces even though no mutating tool exists.
 - Missing URL/user/token/profile ⇒ **fail closed** with a clear message.
-- Since every tool on `jenkins-mcp` is read-only, no confirmation gates are needed — but
+- Since every tool is read-only, no confirmation gates are needed — but
   identity (`jenkins_whoami`) must still work so workflows can prove which
   Jenkins account they act as.
 

docs/llm-workflow-runbooks.md

@@ -229,62 +229,6 @@ Legacy environment-only setups keep working unchanged until migrated.
 Each runbook names the **profile role** it runs under, the steps, and a safe
 prompt. Confirm the active profile first (`gitea_get_profile` / `gitea_whoami`).
 
-## Work Selection Rule for LLMs
-
-Before starting any issue or PR work, acquire or verify a work lease. Do not
-begin coding, reviewing, fixing, branching, committing, pushing, commenting,
-or creating a PR until you prove the target is not already being worked.
-
-Required checks:
-
-1. List open PRs.
-2. Search for PRs linked to the target issue.
-3. Search local and remote branches for the issue number.
-4. Search registered worktrees for the issue branch.
-5. Check dirty worktrees.
-6. Check active leases or recent handoffs.
-7. Check whether the issue was already completed by a merged PR.
-
-If another active LLM/session owns the lease, stop. Allowed responses:
-continue as the lease owner; review the existing PR if reviewer capability
-allows; produce a handoff; request takeover after lease expiry; stop with
-"work already claimed."
-
-Never create a parallel branch or PR for the same issue unless the old branch
-is proven abandoned and the takeover is recorded.
-
-Gitea-Tools lease gates: `gitea_lock_issue` (fail-closed before author
-mutations), `status:in-progress`, and claim comments. Full portable wording:
-[`skills/llm-project-workflow/SKILL.md`](../skills/llm-project-workflow/SKILL.md).
-
-## Global LLM Worktree Rule
-
-The main project checkout is a stable control checkout. It must stay on the
-configured stable branch: `master`, `main`, or `dev`.
-
-All LLM task work must happen inside the project's `branches/` directory.
-
-Before any mutation, prove:
-
-1. current project root
-2. current working directory
-3. current branch
-4. stable branch for the main checkout
-5. session-owned worktree path under `branches/`
-
-If `cwd` is not inside `branches/`, stop. Do not edit, create, delete, format,
-test-write, commit, merge, rebase, checkout task branches, resolve conflicts,
-or run cleanup.
-
-There are no exceptions for small fixes, docs, tests, cleanup, PR review fixes,
-conflict resolution, or emergencies.
-
-The main checkout may only be used for read-only inspection, fetching,
-stable-branch update after merged PRs, creating `branches/` worktrees, or
-explicit control-checkout repair.
-
-Portable wording: [`skills/llm-project-workflow/SKILL.md`](../skills/llm-project-workflow/SKILL.md).
-
 ## Branch worktree isolation
 
 All LLM implementation and review work happens in an isolated branch worktree

docs/mcp-client-registration.md

@@ -9,14 +9,9 @@ Use these exact MCP server names in clients:
 
 | Server name | Boundary | Default capability |
 |---|---|---|
-| `jenkins-mcp` | Jenkins CI inspection (read) | Read-only build/job inspection |
+| `jenkins-mcp` | Jenkins CI inspection | Read-only build/job inspection |
-| `jenkins-write-mcp` | Jenkins build trigger (write) | Gated `jenkins_trigger_build` only |
 | `glitchtip-mcp` | GlitchTip observability inspection | Read-only issue/event inspection |
 
-The write boundary (`jenkins-write-mcp`) is **not** registered by default (#152).
-It exposes a single mutating tool and requires operator approval of a dedicated
-trigger profile before any client config references it.
-
 Historical names such as `jenkins-readonly` and `glitchtip-readonly` are
 descriptive profile labels only. They are not the canonical MCP server names
 unless an operator intentionally creates aliases and documents them.
@@ -59,7 +54,7 @@ entry, reconnect or reload the MCP client before claiming the tools are usable.
 Before using either server in a task, prove the expected tools are visible in
 the client. It is not enough for the config entry to exist.
 
-Expected Jenkins read tools (`jenkins-mcp` only):
+Expected Jenkins tools:
 
 - `jenkins_whoami`
 - `jenkins_list_jobs`
@@ -67,10 +62,6 @@ Expected Jenkins read tools (`jenkins-mcp` only):
 - `jenkins_build_status`
 - `jenkins_get_build`
 
-`jenkins_trigger_build` must **not** appear on `jenkins-mcp`. When an operator
-explicitly enables the write boundary, the only expected tool on
-`jenkins-write-mcp` is `jenkins_trigger_build`.
-
 Expected GlitchTip tools:
 
 - `glitchtip_whoami`
@@ -87,13 +78,8 @@ back to shell commands, raw service APIs, or unrelated MCP servers.
 ## Boundary Rules
 
 - `jenkins-mcp` read profiles must not expose build trigger tools.
-- Build triggers live on the separate `jenkins-write-mcp` server
+- Jenkins build triggers require a separately named write profile, exact
-  (`jenkins_mcp.write_server`), not on `jenkins-mcp`.
+  confirmation, and fail-closed mutation audit.
-- Jenkins build triggers require a dedicated trigger profile with
-  `jenkins.build.trigger` allowed, exact confirmation
-  (`TRIGGER BUILD <job-path>`), and fail-closed mutation audit.
-- Do not register `jenkins-write-mcp` until an operator approves a trigger
-  profile; no shipped profile carries trigger capability by default.
 - `glitchtip-mcp` remains read-only. It must not file or mutate Gitea issues.
 - GlitchTip-to-Gitea filing is a separate orchestrator that composes GlitchTip
   read tools with Gitea issue-write tools.

docs/safety-model.md

@@ -21,10 +21,5 @@ Note on naming: Historical design docs used `jenkins-readonly` / `glitchtip-read
 
 ## 5. Mutation Gating
 Any mutating action (e.g., Gitea issue creation from GlitchTip, or Jenkins builds) must be explicitly allowed by the execution profile.
-- **Jenkins build triggers** are gated on a separate write boundary
+- **Jenkins build triggers** exist in jenkins-mcp (landed #4) but require dedicated profile/identity and exact confirmation; not on standard reader profiles. See #56 for boundary correction.
-  (`jenkins-write-mcp` / `jenkins_mcp.write_server`), not on the read-only
+- **GlitchTip to Gitea issue filing** is documented as a gated, orchestrated workflow (not in glitchtip-mcp), currently partial (mocked, dedup not wired, audit missing). See #57.
-  `jenkins-mcp` surface. Triggers require a dedicated profile with
-  `jenkins.build.trigger`, exact confirmation, and fail-closed mutation audit.
-  No default profile carries trigger capability (#152 / mcp-control-plane #56).
-- **GlitchTip to Gitea issue filing** is a library-only orchestrator in
-  mcp-control-plane (not on `glitchtip-mcp`). See #153 / mcp-control-plane #57.

docs/wiki/Operator-Guide.md

@@ -13,7 +13,6 @@ Handbook for LLM operators and human developers using the Gitea-Tools MCP server
 3. **One unit of work per session** — Implement one claimed issue *or* review/merge one PR; do not mix author and reviewer mutations in one session.
 4. **No self-review / no self-merge** — The authenticated Gitea user must not approve or merge a PR they authored.
 5. **Follow the gates** — Prompts express intent; MCP tools enforce safety. Never bypass gates via prompt instructions.
-6. **Global LLM Worktree Rule** — Main checkout stays on `master`/`main`/`dev`; all mutations happen under `branches/`. Prove project root, `cwd`, branch, stable main-checkout branch, and session worktree path before editing. No exceptions.
 
 ## Supported Gitea instances
 

docs/wiki/Workflow.md

@@ -14,12 +14,6 @@
 
 ## Step 2: Implement issues (author profile)
 
-0. Work Selection Rule — verify a work lease before any mutations (open PRs,
-   issue-linked PRs, branches, worktrees, dirty worktrees, active leases/
-   handoffs, merged-PR completion). Stop if another session owns the lease.
-0b. Global LLM Worktree Rule — main checkout on `master`/`main`/`dev` only;
-   mutate only from a `branches/` worktree after proving root, cwd, branch,
-   stable main-checkout branch, and session worktree path (no exceptions).
 1. `gitea_resolve_task_capability` for the author task.
 2. `gitea_lock_issue` before implementation mutations.
 3. Claim with `gitea_mark_issue` / `status:in-progress` label.

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":
@@ -3460,40 +3269,11 @@ _GUIDE_RULES = {
         "Conflicting or stale PR state (e.g. prior said merged but live open, or head/updated mismatch) is a hard blocker: report it explicitly and stop until live state is unambiguous.",
         "After merge, re-list PRs and re-view the PR (plus verify master contains the merge) before claiming completion.",
     ],
-    "work_selection": (
-        "Before any issue or PR work, acquire or verify a work lease. Do not "
-        "code, review, branch, commit, push, comment, or open a PR until you "
-        "prove the target is not already being worked. Required checks: list "
-        "open PRs; search PRs linked to the issue; search local/remote "
-        "branches for the issue number; search worktrees for the issue "
-        "branch; check dirty worktrees; check active leases or recent "
-        "handoffs; check whether a merged PR already completed the issue. If "
-        "another session owns the lease, stop (continue as owner, review the "
-        "existing PR, hand off, request takeover after expiry, or report "
-        "'work already claimed'). Never create a parallel branch/PR unless "
-        "the old branch is proven abandoned and takeover is recorded. "
-        "gitea_lock_issue is the fail-closed author lease gate."),
-    "global_worktree": (
-        "Main checkout is a stable control checkout on master/main/dev only. "
-        "All LLM task work happens under branches/. Before any mutation, prove "
-        "project root, cwd, branch, main-checkout stable branch, and "
-        "session-owned branches/ worktree path. If cwd is not under "
-        "branches/, stop — no edit/create/delete/format/test-write/commit/"
-        "merge/rebase/checkout/cleanup, with no exceptions (docs, tests, "
-        "small fixes, review fixes, conflicts, emergencies). Main checkout: "
-        "read-only inspect, fetch, create worktrees, post-merge stable "
-        "update, explicit repair only."),
 }
 
 _COMMON_WORKFLOWS = [
     "task routing: resolve task capability via gitea_resolve_task_capability "
     "to check required permission/role kind and identify the safe next action.",
-    "work selection: acquire or verify a work lease before any issue/PR work "
-    "(open PRs, linked PRs, branches, worktrees, dirty worktrees, leases, "
-    "merged completion); stop if another session owns the lease.",
-    "global worktree: main checkout on master/main/dev only; mutate only "
-    "from a branches/ worktree after proving root, cwd, branch, stable "
-    "main-checkout branch, and session worktree path.",
     "issue authoring: verify identity, create/claim the issue, keep scope "
     "explicit (remote/org/repo).",
     "implementation: claim issue, branch from fresh master, implement only "
@@ -4190,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,
@@ -4212,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:
@@ -4861,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:
@@ -4891,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."
 
@@ -4981,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:
@@ -5001,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,24 +6,26 @@ 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,
-    skip_python_scan_walk_root,
-)
-
-
 # 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 skip_python_scan_walk_root(dir_path, 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:
-                        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}. "

review_proofs.py

@@ -1297,16 +1297,6 @@ HANDOFF_ROLE_FIELDS = {
         ("Session authored PR", ("session authored pr", "authored pr")),
         ("Why continuation allowed", ("why continuation", "continuation allowed")),
     ),
-    "issue_filing": (
-        ("Issue created or updated", (
-            "issue created or updated",
-            "issue created",
-            "issue updated",
-            "created issue",
-            "updated issue",
-        )),
-        ("Related issues", ("related issues",)),
-    ),
 }
 
 
@@ -2145,304 +2135,6 @@ def assess_issue_selection_final_report(
     }
 
 
-_SHA_EVIDENCE_LABEL = re.compile(
-    r"(?:\b(?:old|new)\s+head\b|\bhead\s+sha\b|\bcommit\s+sha\b|\bsha\b)"
-    r"\s*[:=]?\s*([0-9a-f]+)",
-    re.IGNORECASE,
-)
-
-
-def assess_issue_filing_sha_evidence(report_text: str) -> dict:
-    """Issue #191: commit evidence must use full 40-hex SHAs."""
-    text = report_text or ""
-    reasons = []
-    for match in _SHA_EVIDENCE_LABEL.finditer(text):
-        sha = match.group(1).strip().lower()
-        if sha and not _FULL_SHA.match(sha):
-            reasons.append(
-                f"abbreviated SHA in evidence ({len(sha)} hex chars); "
-                "full 40-character SHA required"
-            )
-    return {
-        "complete": not reasons,
-        "downgraded": bool(reasons),
-        "reasons": reasons,
-    }
-
-
-def assess_issue_filing_issue_reference(
-    report_text: str,
-    *,
-    issue_number: int | None = None,
-    issue_title: str | None = None,
-    action: str = "created",
-) -> dict:
-    """Issue #191: reports must cite exact issue number and title."""
-    text = (report_text or "").lower()
-    reasons = []
-    if issue_number is None:
-        reasons.append("issue number proof missing from assessment context")
-    else:
-        num_patterns = (f"#{issue_number}", f"issue #{issue_number}",
-                        f"issue {issue_number}")
-        if not any(p in text for p in num_patterns):
-            reasons.append(
-                f"report missing exact issue number #{issue_number}"
-            )
-    if issue_title:
-        title_fragment = issue_title.strip().lower()[:40]
-        if title_fragment and title_fragment not in text:
-            reasons.append("report missing exact issue title")
-    action = (action or "created").strip().lower()
-    if action == "created" and "creat" not in text:
-        reasons.append(
-            "report must distinguish new issue creation from update"
-        )
-    if action == "updated" and "updat" not in text:
-        reasons.append(
-            "report must distinguish issue update from new creation"
-        )
-    return {
-        "complete": not reasons,
-        "downgraded": bool(reasons),
-        "reasons": reasons,
-    }
-
-
-def assess_issue_filing_mutation_capability(
-    report_text: str,
-    mutations: list[str] | None,
-    resolved_capabilities: dict | None = None,
-) -> dict:
-    """Issue #191: every mutation needs exact capability proof in the report."""
-    import task_capability_map
-
-    text = (report_text or "").lower()
-    reasons = []
-    mutation_list = list(mutations or [])
-    if not mutation_list:
-        reasons.append("no mutations listed for capability proof")
-    cap_proof = assess_capability_proof(resolved_capabilities or {})
-    if not cap_proof.get("proven"):
-        reasons.extend(cap_proof.get("reasons") or [])
-
-    for task in mutation_list:
-        try:
-            permission = task_capability_map.required_permission(task)
-        except KeyError:
-            reasons.append(f"unknown mutation task '{task}'")
-            continue
-        if permission.lower() not in text and task.replace("_", " ") not in text:
-            reasons.append(
-                f"mutation '{task}' missing exact capability proof "
-                f"('{permission}')"
-            )
-        if task == "set_issue_labels":
-            label_proof = (
-                "label change",
-                "labels changed",
-                "set label",
-                "label mutation",
-                "issue labels",
-                "label:",
-            )
-            if not any(p in text for p in label_proof):
-                reasons.append(
-                    "label mutation missing label-specific capability proof"
-                )
-    return {
-        "complete": not reasons,
-        "downgraded": bool(reasons),
-        "reasons": reasons,
-    }
-
-
-def assess_issue_filing_mutation_scope(
-    report_text: str,
-    *,
-    performed_mutations: list[str] | None = None,
-    forbidden_mutations: list[str] | None = None,
-) -> dict:
-    """Issue #191: single-mutation runs must state scope and absent mutations."""
-    text = (report_text or "").lower()
-    performed = list(performed_mutations or [])
-    forbidden = list(forbidden_mutations or [
-        "label", "comment", "pr", "review", "merge", "close",
-    ])
-    reasons = []
-
-    if len(performed) == 1:
-        only_phrases = ("only mutation", "only mutations", "sole mutation")
-        if not any(p in text for p in only_phrases):
-            reasons.append(
-                "single-mutation run must state 'Only mutation(s): ...'"
-            )
-        for absent in forbidden:
-            if absent == "comment" and performed[0] in (
-                "comment_issue", "create_issue",
-            ):
-                continue
-            if absent in text and f"no {absent}" not in text:
-                if any(
-                    phrase in text
-                    for phrase in (
-                        f"no {absent}",
-                        f"no {absent}s",
-                        f"without {absent}",
-                        "none performed",
-                        "none.",
-                    )
-                ):
-                    continue
-        if performed == ["create_issue"]:
-            for check in ("label", "comment", "pr", "review", "merge"):
-                if check in text and f"no {check}" not in text:
-                    if not any(
-                        n in text
-                        for n in (
-                            f"no {check}",
-                            f"no {check}s",
-                            "none performed",
-                            "confirm no",
-                        )
-                    ):
-                        reasons.append(
-                            f"create-only run must confirm no {check} "
-                            "mutations were performed"
-                        )
-
-    return {
-        "complete": not reasons,
-        "downgraded": bool(reasons),
-        "reasons": reasons,
-    }
-
-
-def assess_issue_filing_duplicate_summary(
-    report_text: str,
-    *,
-    issues_searched: int | None = None,
-    closest_matches: list[dict] | None = None,
-    duplicate_result: dict | None = None,
-) -> dict:
-    """Issue #191: duplicate-check evidence must be summarized in the report."""
-    text = (report_text or "").lower()
-    reasons = []
-    dup = duplicate_result or {}
-
-    if issues_searched is not None:
-        count_tokens = (
-            str(issues_searched),
-            f"{issues_searched} open",
-            f"{issues_searched} issue",
-        )
-        if not any(t in text for t in count_tokens):
-            reasons.append(
-                "duplicate-check summary missing open-issues searched count"
-            )
-
-    matches = closest_matches if closest_matches is not None else dup.get("matches")
-    for match in matches or []:
-        num = match.get("number")
-        if num is not None and f"#{num}" not in text and str(num) not in text:
-            reasons.append(
-                f"duplicate-check summary missing closest issue #{num}"
-            )
-            break
-
-    justification_phrases = (
-        "why new",
-        "why update",
-        "rejected update",
-        "new issue justified",
-        "not a duplicate",
-        "no duplicate",
-        "justified",
-        "instead of expanding",
-    )
-    if not any(p in text for p in justification_phrases):
-        reasons.append(
-            "duplicate-check summary missing why update was rejected or "
-            "why a new issue was justified"
-        )
-
-    dup_proof = issue_duplicate_gate.assess_duplicate_search_proof(
-        report_text, matches or []
-    )
-    if not dup_proof.get("valid"):
-        reasons.extend(dup_proof.get("reasons") or [])
-
-    return {
-        "complete": not reasons,
-        "downgraded": bool(reasons),
-        "reasons": reasons,
-    }
-
-
-def assess_issue_filing_final_report(
-    report_text: str,
-    *,
-    issue_number: int | None = None,
-    issue_title: str | None = None,
-    action: str = "created",
-    mutations: list[str] | None = None,
-    resolved_capabilities: dict | None = None,
-    issues_searched: int | None = None,
-    closest_matches: list[dict] | None = None,
-    duplicate_result: dict | None = None,
-    performed_mutations: list[str] | None = None,
-) -> dict:
-    """Issue #191: composite A-bar for issue-filing final reports."""
-    checks = {
-        "controller_handoff": assess_controller_handoff(
-            report_text, role="issue_filing"
-        ),
-        "issue_reference": assess_issue_filing_issue_reference(
-            report_text,
-            issue_number=issue_number,
-            issue_title=issue_title,
-            action=action,
-        ),
-        "sha_evidence": assess_issue_filing_sha_evidence(report_text),
-        "mutation_capability": assess_issue_filing_mutation_capability(
-            report_text,
-            mutations or performed_mutations,
-            resolved_capabilities,
-        ),
-        "mutation_scope": assess_issue_filing_mutation_scope(
-            report_text,
-            performed_mutations=performed_mutations or mutations,
-        ),
-        "duplicate_summary": assess_issue_filing_duplicate_summary(
-            report_text,
-            issues_searched=issues_searched,
-            closest_matches=closest_matches,
-            duplicate_result=duplicate_result,
-        ),
-    }
-
-    reasons = []
-    downgraded = False
-    for name, result in checks.items():
-        if result.get("verdict") == "missing":
-            downgraded = True
-            reasons.extend(result.get("reasons") or [])
-        elif result.get("downgraded") or not result.get("complete", True):
-            downgraded = True
-            reasons.extend(
-                f"{name}: {r}" for r in (result.get("reasons") or [])
-            )
-
-    grade = "A" if not downgraded else "downgraded"
-    return {
-        "grade": grade,
-        "downgraded": downgraded,
-        "checks": checks,
-        "reasons": reasons,
-        "complete": not downgraded,
-    }
-
-
 def assess_duplicate_search_proof(report_text, matches):
     """#207: reject LLM duplicate summaries that omit known title matches."""
     return issue_duplicate_gate.assess_duplicate_search_proof(

role_session_router.py

@@ -14,45 +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
-
-
-def skip_python_scan_walk_root(project_root: str, walk_root: str) -> bool:
-    """Skip venv/git/cache and sibling worktrees under orchestration checkout.
-
-    When *project_root* is itself a worktree inside ``branches/``, still scan
-    that tree — do not treat the ``branches`` path segment as a skip signal.
-    """
-    rel = os.path.relpath(walk_root, project_root)
-    if rel == ".":
-        return False
-    head = rel.split(os.sep, 1)[0]
-    if head in ("venv", ".git", ".pytest_cache"):
-        return True
-    if head == "branches":
-        nested = os.path.join(project_root, "branches")
-        if os.path.isdir(nested) and (
-            walk_root == nested or walk_root.startswith(nested + os.sep)
-        ):
-            return True
-    return False
-
-
 REVIEWER_TASKS = frozenset({
     "review_pr",
     "merge_pr",
@@ -233,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":
@@ -291,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 skip_python_scan_walk_root(root_dir, root):
-            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__))
@@ -327,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

skills/llm-project-workflow/SKILL.md

@@ -53,73 +53,10 @@ Additional issue-first rules:
   owner decision.** Do not create a new repository or a new tracker unless
   explicitly approved by the owner.
 
-## Work Selection Rule for LLMs
-
-Before starting any issue or PR work, acquire or verify a work lease.
-
-Do not begin coding, reviewing, fixing, branching, committing, pushing,
-commenting, or creating a PR until you prove the target is not already being
-worked.
-
-Required checks:
-
-1. List open PRs.
-2. Search for PRs linked to the target issue.
-3. Search local and remote branches for the issue number.
-4. Search registered worktrees for the issue branch.
-5. Check dirty worktrees.
-6. Check active leases or recent handoffs.
-7. Check whether the issue was already completed by a merged PR.
-
-If another active LLM/session owns the lease, stop.
-
-Allowed responses:
-
-- continue as the lease owner,
-- review the existing PR if reviewer capability allows,
-- produce a handoff,
-- request takeover after lease expiry,
-- stop with "work already claimed."
-
-Never create a parallel branch or PR for the same issue unless the old branch
-is proven abandoned and the takeover is recorded.
-
-For Gitea-Tools: `gitea_lock_issue` is the fail-closed lease gate before author
-mutations; `status:in-progress` and claim comments are supporting lease signals.
-Use `review_proofs.classify_issue_for_selection` when reporting fresh issue
-selection (#188).
-
-## Global LLM Worktree Rule
-
-The main project checkout is a stable control checkout. It must stay on the
-configured stable branch: `master`, `main`, or `dev`.
-
-All LLM task work must happen inside the project's `branches/` directory.
-
-Before any mutation, prove:
-
-1. current project root
-2. current working directory
-3. current branch
-4. stable branch for the main checkout
-5. session-owned worktree path under `branches/`
-
-If `cwd` is not inside `branches/`, stop. Do not edit, create, delete, format,
-test-write, commit, merge, rebase, checkout task branches, resolve conflicts,
-or run cleanup.
-
-There are no exceptions for small fixes, docs, tests, cleanup, PR review fixes,
-conflict resolution, or emergencies.
-
-The main checkout may only be used for read-only inspection, fetching,
-stable-branch update after merged PRs, creating `branches/` worktrees, or
-explicit control-checkout repair.
-
 ## B. Isolated worktree rule
 
-**Never implement or review in the main checkout** (Global LLM Worktree Rule).
+**Never implement or review in the main checkout.** The main checkout is for
-The main checkout is for orchestration and status only (issue creation,
+orchestration and status only (issue creation, `git status`, creating worktrees).
-`git status`, creating worktrees) and must remain on the stable branch.
 
 - Each issue gets its own branch worktree under an ignored `branches/` directory.
 - Review work uses a **separate** review worktree, never the author's folder.
@@ -205,12 +142,7 @@ Worktree folder = branch with `/` replaced by `-`
 
 ## E. Start-work workflow
 
-0. Acquire or verify a work lease (Work Selection Rule) — complete all seven
+1. Verify the orchestration checkout (right repo, clean tree).
-   checks before any claim, branch, or PR work.
-0b. Global LLM Worktree Rule — prove project root, `cwd`, branch, main-checkout
-    stable branch, and session-owned `branches/` worktree path. If `cwd` is not
-    under `branches/`, stop before any mutation (no exceptions).
-1. Verify the orchestration checkout (right repo, clean tree, on stable branch).
 2. Fetch/prune: `git fetch <remote> --prune`.
 3. Confirm local `master` equals remote `master` (`git rev-list --left-right --count <remote>/master...master` → `0	0`).
 4. Create/claim the issue (§A).
@@ -462,12 +394,6 @@ Role-specific fields (append to the compact block):
 - review/merge tasks: `Selected PR:`, `Reviewer eligibility:`,
   `Pinned reviewed head:`, `Review decision:`, `Merge result:`,
   `Linked issue status:`, `Cleanup status:`
-- issue-filing tasks (#191): `Issue created or updated:`, `Related issues:`;
-  body must cite exact issue number/title, duplicate-search summary (issues
-  searched, closest matches, why update rejected / new issue justified), full
-  40-char SHAs when citing commits, exact mutation capability per change, and
-  `Only mutation(s):` when a single mutation was performed
-  (`review_proofs.assess_issue_filing_final_report`).
 - author tasks: `Selected issue:`, `Claim/comment status:`,
   `PR number opened:`, `No review/merge:` (explicit confirmation)
 - continuation tasks (#188): `Continuation mode:`, `Existing PR:`,

skills/llm-project-workflow/templates/start-issue.md

@@ -13,21 +13,6 @@ Rules (llm-project-workflow):
 - Do not self-review or self-merge.
 
 Steps:
-0. Work Selection Rule — before any claim, branch, or file edits, acquire or
-   verify a work lease. Required checks: list open PRs; search PRs linked to
-   the target issue; search local/remote branches for the issue number; search
-   registered worktrees for the issue branch; check dirty worktrees; check
-   active leases or recent handoffs; check whether a merged PR already
-   completed the issue. If another session owns the lease, stop (continue only
-   as lease owner, review the existing PR, hand off, request takeover after
-   expiry, or report "work already claimed"). Never open a parallel branch/PR
-   unless the old branch is proven abandoned and takeover is recorded.
-0b. Global LLM Worktree Rule — before any mutation, prove and state: project
-   root; cwd; current branch; stable branch for the main checkout (master/main/dev);
-   session-owned worktree path under branches/. If cwd is not inside branches/,
-   STOP (no exceptions — not for docs, tests, small fixes, review fixes, conflicts,
-   or cleanup). Main checkout is control-only: read-only inspect, fetch, create
-   worktrees, stable-branch update after merge, explicit repair.
 1. Identity Checklist: Before claiming work, verify and state:
    - Required identity/profile for this task: author (allowed to push branches / create PRs)
    - Current authenticated identity (from whoami): <username>

tests/test_external_mcp_registration_docs.py

@@ -1,4 +1,4 @@
-"""Documentation checks for external Jenkins/GlitchTip MCP registration (#151/#152)."""
+"""Documentation checks for external Jenkins/GlitchTip MCP registration (#151)."""
 from pathlib import Path
 
 REPO_ROOT = Path(__file__).resolve().parent.parent
@@ -52,20 +52,11 @@ def test_registration_doc_preserves_boundaries():
         "Do not add Jenkins or GlitchTip credentials to the Gitea MCP server",
         "do not add Gitea write credentials to the GlitchTip server",
         "must not expose build trigger tools",
-        "jenkins-write-mcp",
-        "jenkins_mcp.write_server",
         "remains read-only",
     ):
         assert phrase in text
 
 
-def test_registration_doc_separates_jenkins_trigger_from_read_surface():
-    text = _doc_text()
-    assert "jenkins_trigger_build" in text
-    assert "must **not** appear on `jenkins-mcp`" in text
-    assert "not" in text.lower() and "registered by default" in text.lower()
-
-
 def test_registration_doc_has_no_secret_material_or_live_urls():
     text = _doc_text()
     for marker in (
@@ -89,14 +80,3 @@ def test_related_docs_link_registration_doc():
         text = (REPO_ROOT / name).read_text(encoding="utf-8")
         assert "mcp-client-registration.md" in text, (
             f"{name} does not link docs/mcp-client-registration.md")
-
-
-def test_related_docs_keep_jenkins_trigger_off_read_surface():
-    for name in (
-        "docs/safety-model.md",
-        "docs/architecture/jenkins-readonly-build-status-design.md",
-    ):
-        text = (REPO_ROOT / name).read_text(encoding="utf-8")
-        assert "jenkins-write-mcp" in text
-        assert "jenkins_mcp.write_server" in text
-        assert "jenkins-mcp" in text

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_operator_guide.py

@@ -132,8 +132,7 @@ class TestControlPlaneGuide(GuideTestBase):
         rules = g["rules"]
         for key in ("hard_stops", "fail_closed", "head_sha_pinning",
                     "merge_confirmation", "redaction", "separation",
-                    "profile_switching", "identity_verification",
+                    "profile_switching", "identity_verification"):
-                    "work_selection", "global_worktree"):
             self.assertIn(key, rules)
         self.assertIn("MERGE PR", json.dumps(rules["merge_confirmation"]))
         self.assertTrue(rules["hard_stops"])

tests/test_review_proofs.py

@@ -33,9 +33,6 @@ from review_proofs import (  # noqa: E402
     assess_empty_queue_report,
     assess_fresh_issue_selection,
     assess_inventory_completeness,
-    assess_issue_filing_final_report,
-    assess_issue_filing_mutation_capability,
-    assess_issue_filing_sha_evidence,
     assess_issue_selection_final_report,
     assess_queue_target_final_report,
     assess_reviewer_queue_inventory,
@@ -1054,10 +1051,6 @@ class TestControllerHandoff(unittest.TestCase):
         self.assertIn("## Controller Handoff", skill)
         self.assertIn("assess_controller_handoff", skill)
         self.assertIn("issue #182", skill)
-        self.assertIn("## Work Selection Rule for LLMs", skill)
-        self.assertIn("work already claimed", skill)
-        self.assertIn("## Global LLM Worktree Rule", skill)
-        self.assertIn("branches/", skill)
 
     def test_handoff_rejects_none_workspace_mutations_when_local_edits_exist(self):
         # 1. Workspace mutations: none is rejected when local_edits is True
@@ -1932,115 +1925,5 @@ class TestIssueSelectionContinuation(unittest.TestCase):
         self.assertEqual(result["grade"], "A")
 
 
-class TestIssueFilingFinalReport(unittest.TestCase):
-    """Issue #191: issue-filing runs need A-bar final report proofs."""
-
-    ISSUE_TITLE = (
-        "Implement fail-closed continuation mode for issues already "
-        "represented by open PRs"
-    )
-    FULL_SHA = "a4060c5de00f2b1c9e88f4f6f0f3f9a7b2c1d0e9"
-    CAPABILITIES = {
-        "create_issue": {
-            "requested_task": "create_issue",
-            "required_operation_permission": "gitea.issue.create",
-            "allowed_in_current_session": True,
-        },
-    }
-    CLOSEST = [{"number": 183, "title": "Harden author-run reporting", "state": "open"}]
-
-    def _good_report(self, *, sha_line=None):
-        lines = [
-            "Created Gitea-Tools #189 — "
-            f"`{self.ISSUE_TITLE}`",
-            "Duplicate check: searched 12 open issues.",
-            "Closest existing: #183 — Harden author-run reporting "
-            "(update rejected — different scope).",
-            "Why new issue justified: continuation mode is distinct from #183.",
-            "Only mutation: issue creation (gitea.issue.create).",
-            "Confirm no labels, comments, PRs, reviews, merges, or closes.",
-            "## Controller Handoff",
-            "- Task: file issue",
-            "- Repo: Scaled-Tech-Consulting/Gitea-Tools",
-            "- Role: author",
-            "- Identity: prgs-author",
-            "- Issue/PR: #189",
-            "- Branch/SHA: n/a",
-            "- Files changed: none",
-            "- Validation: duplicate gate + create_issue capability resolved",
-            "- Mutations: create_issue via gitea.issue.create",
-            "- Workspace mutations: none",
-            "- Current status: issue created",
-            "- Blockers: none",
-            "- Next: implementation",
-            "- Safety: no review/merge",
-            "- Issue created or updated: created #189",
-            "- Related issues: #183, #188",
-        ]
-        if sha_line:
-            lines.insert(4, sha_line)
-        return "\n".join(lines)
-
-    def test_complete_issue_filing_report_earns_a(self):
-        result = assess_issue_filing_final_report(
-            self._good_report(),
-            issue_number=189,
-            issue_title=self.ISSUE_TITLE,
-            action="created",
-            mutations=["create_issue"],
-            resolved_capabilities=self.CAPABILITIES,
-            issues_searched=12,
-            closest_matches=self.CLOSEST,
-            performed_mutations=["create_issue"],
-        )
-        self.assertEqual(result["grade"], "A")
-        self.assertFalse(result["downgraded"])
-
-    def test_missing_controller_handoff_downgrades(self):
-        report = self._good_report().replace("## Controller Handoff", "")
-        result = assess_issue_filing_final_report(
-            report,
-            issue_number=189,
-            issue_title=self.ISSUE_TITLE,
-            mutations=["create_issue"],
-            resolved_capabilities=self.CAPABILITIES,
-            issues_searched=12,
-            performed_mutations=["create_issue"],
-        )
-        self.assertTrue(result["downgraded"])
-
-    def test_abbreviated_sha_downgrades(self):
-        result = assess_issue_filing_sha_evidence(
-            f"old head: {self.FULL_SHA[:7]}"
-        )
-        self.assertTrue(result["downgraded"])
-
-    def test_mutation_without_capability_proof_downgrades(self):
-        report = self._good_report().replace("gitea.issue.create", "allowed")
-        result = assess_issue_filing_mutation_capability(
-            report,
-            ["create_issue"],
-            self.CAPABILITIES,
-        )
-        self.assertTrue(result["downgraded"])
-
-    def test_label_mutation_without_label_proof_downgrades(self):
-        caps = {
-            "set_issue_labels": {
-                "requested_task": "set_issue_labels",
-                "required_operation_permission": "gitea.issue.comment",
-                "allowed_in_current_session": True,
-            },
-        }
-        report = "\n".join([
-            "Updated labels with gitea.issue.comment capability resolved.",
-            "Only mutations: set_issue_labels",
-        ])
-        result = assess_issue_filing_mutation_capability(
-            report, ["set_issue_labels"], caps
-        )
-        self.assertTrue(result["downgraded"])
-
-
 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,49 +253,5 @@ class TestRoleSessionRouter(unittest.TestCase):
         self.assertTrue(complete["complete"])
 
 
-class TestCheckMidMerge(unittest.TestCase):
-    def test_skip_scan_walk_root_skips_sibling_worktrees_only(self):
-        worktree_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
-        main_root = os.path.dirname(os.path.dirname(worktree_root))
-        self.assertFalse(
-            role_session_router.skip_python_scan_walk_root(
-                main_root, main_root
-            )
-        )
-        self.assertTrue(
-            role_session_router.skip_python_scan_walk_root(
-                main_root, os.path.join(main_root, "branches", "fix-issue-1")
-            )
-        )
-        self.assertFalse(
-            role_session_router.skip_python_scan_walk_root(
-                worktree_root, worktree_root
-            )
-        )
-        self.assertFalse(
-            role_session_router.skip_python_scan_walk_root(
-                worktree_root, os.path.join(worktree_root, "tests")
-            )
-        )
-
-    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()