feat: enforce capability stop terminal mode after reviewer denial (Issue #197 )

Enter terminal mode when review_pr/merge_pr capability is denied. Block reviewer queue tools (list_prs, eligibility checks) and add report-purity validators for forbidden PR selection, fallback, and empty-queue claims. Closes #197 Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
2026-07-05 17:17:16 -04:00
14 changed files with 25 additions and 1632 deletions
@@ -203,29 +203,6 @@ remote/org/repo arguments. Create operations are audit-logged
 redacted, and normal output contains no endpoint URLs
 (`GITEA_MCP_REVEAL_ENDPOINTS=1` is the local admin opt-in for web links).
 ## PR edits versus PR closure (#216)
 Editing a pull request and closing one are different capabilities:
 - **PR edits** (`gitea_edit_pr` with `title`/`body`/`base`, or reopening with
  `state="open"`) stay on the ordinary edit path and need no dedicated
  capability.
 - **PR closure** (`gitea_edit_pr` with `state="closed"`) requires the
  distinct `gitea.pr.close` operation. The resolver task is `close_pr`
  (`gitea_resolve_task_capability(task="close_pr")`, author-side). Without
  `gitea.pr.close` the close attempt fails closed — no API call, structured
  `permission_report` — so the broad edit path can never be used as an
  untracked close fallback.
 - Closures are audited as a distinct `close_pr` action with
  `required_permission: gitea.pr.close` in the request metadata, so final
  reports can prove exactly which mutation capability was exercised (#191).
 `gitea.pr.close` has no legacy alias; spell it canonically. It is not part of
 any default profile: the operator grants it deliberately (e.g. for an
 explicit operator-directed closure of a contaminated PR). If `close_pr` ever
 resolves as unknown, agents must fail closed rather than fall back to the
 edit path.
 ## Identity and fail-closed rules
 Before **any** mutating action, a workflow must know both:
@@ -1,170 +0,0 @@
 """Pre-create issue duplicate gate (#207)."""
 from __future__ import annotations
 import re
 import unicodedata
 VERDICT_NO_DUPLICATE = "no_duplicate_found"
 VERDICT_DUPLICATE = "duplicate_found"
 VERDICT_AMBIGUOUS = "ambiguous_duplicate_stop"
 _STOPWORDS = frozenset({"a", "an", "the", "and", "or", "for", "to", "of", "in", "on"})
 def normalize_issue_title(title: str) -> str:
    """Lowercase, punctuation-stripped, whitespace-collapsed title."""
    text = unicodedata.normalize("NFKC", (title or "").strip().lower())
    text = re.sub(r"[^\w\s]", " ", text)
    text = re.sub(r"\s+", " ", text).strip()
    return text
 def _title_tokens(title: str) -> set[str]:
    return {
        t for t in normalize_issue_title(title).split()
        if t and t not in _STOPWORDS
    }
 def titles_near_duplicate(proposed: str, existing: str) -> bool:
    """True when normalized titles match or are near-duplicates."""
    norm_a = normalize_issue_title(proposed)
    norm_b = normalize_issue_title(existing)
    if not norm_a or not norm_b:
        return False
    if norm_a == norm_b:
        return True
    if norm_a in norm_b or norm_b in norm_a:
        return True
    tokens_a = _title_tokens(proposed)
    tokens_b = _title_tokens(existing)
    if not tokens_a or not tokens_b:
        return False
    overlap = tokens_a & tokens_b
    union = tokens_a | tokens_b
    ratio = len(overlap) / len(union)
    if ratio >= 0.85:
        return True
    wall_pair = (
        {"hard", "wall"} <= tokens_a and {"wall"} <= tokens_b
    ) or (
        {"hard", "wall"} <= tokens_b and {"wall"} <= tokens_a
    )
    return wall_pair
 def assess_pre_create_duplicate(
    proposed_title: str,
    existing_issues: list[dict],
    *,
    duplicate_override_reason: str | None = None,
    split_from_issue: int | None = None,
 ) -> dict:
    """Evaluate duplicate risk immediately before issue creation."""
    proposed_title = (proposed_title or "").strip()
    if not proposed_title:
        return {
            "verdict": VERDICT_AMBIGUOUS,
            "performed": False,
            "reasons": ["issue title is required"],
            "matches": [],
        }
    override = (duplicate_override_reason or "").strip()
    if override and split_from_issue is not None:
        return {
            "verdict": VERDICT_NO_DUPLICATE,
            "performed": True,
            "override_applied": True,
            "split_from_issue": split_from_issue,
            "override_reason": override,
            "reasons": [],
            "matches": [],
        }
    matches = []
    for issue in existing_issues or []:
        existing_title = (issue.get("title") or "").strip()
        if not existing_title:
            continue
        if titles_near_duplicate(proposed_title, existing_title):
            matches.append({
                "number": issue.get("number"),
                "title": existing_title,
                "state": issue.get("state"),
            })
    if not matches:
        return {
            "verdict": VERDICT_NO_DUPLICATE,
            "performed": True,
            "normalized_title": normalize_issue_title(proposed_title),
            "reasons": [],
            "matches": [],
        }
    if len(matches) > 3:
        return {
            "verdict": VERDICT_AMBIGUOUS,
            "performed": False,
            "normalized_title": normalize_issue_title(proposed_title),
            "reasons": [
                "too many near-duplicate title matches; fail closed "
                "until operator clarifies"
            ],
            "matches": matches[:5],
        }
    return {
        "verdict": VERDICT_DUPLICATE,
        "performed": False,
        "normalized_title": normalize_issue_title(proposed_title),
        "reasons": [
            f"duplicate issue title blocked: matches existing "
            f"#{m['number']} ({m['state']})"
            for m in matches
        ],
        "matches": matches,
    }
 def pre_create_issue_duplicate_gate(
    proposed_title: str,
    existing_issues: list[dict],
    *,
    duplicate_override_reason: str | None = None,
    split_from_issue: int | None = None,
    allow_override: bool = False,
 ) -> dict:
    """Alias for ``assess_pre_create_duplicate`` (#207 suggested name)."""
    reason = (duplicate_override_reason or "").strip()
    if allow_override and not reason:
        reason = "operator-approved split after duplicate review"
    return assess_pre_create_duplicate(
        proposed_title,
        existing_issues,
        duplicate_override_reason=reason or None,
        split_from_issue=split_from_issue,
    )
 def assess_duplicate_search_proof(report_text: str, matches: list[dict]) -> dict:
    """Reject LLM duplicate summaries that omit known exact duplicates (#207)."""
    text = (report_text or "").lower()
    missing = []
    for match in matches or []:
        num = match.get("number")
        title = (match.get("title") or "").lower()
        if num is not None and f"#{num}" not in text and str(num) not in text:
            missing.append(f"issue #{num}")
        if title and title[:40] not in text:
            missing.append(f"title '{match.get('title')}'")
    if missing:
        return {
            "valid": False,
            "reasons": [
                "duplicate-search proof omitted required match: " + ", ".join(missing)
            ],
        }
    return {"valid": True, "reasons": []}
@@ -16,135 +16,10 @@ Configuration (mcp_config.json):
 import os
 import re
 import sys
 import json
 import functools
 import contextlib
 import subprocess
 # Mutation-authority record (#199, refs #194). Deliberately in-process, NOT a
 # file: a /tmp lock is host-global, writable (spoofable) by any local process,
 # goes silently stale across sessions, and races between concurrent agent
 # sessions. This record lives and dies with the MCP server process, so it can
 # never be forged from outside or leak between sessions. The CLI side-channel
 # (a subprocess overriding GITEA_MCP_PROFILE to escalate roles) is covered by
 # SESSION_PROFILE_LOCK_ENV below: the server exports its launch profile into
 # the environment, children inherit it, and reviewer CLIs (review_pr.py)
 # refuse to run under a different resolved profile.
 _MUTATION_AUTHORITY: dict | None = None
 SESSION_PROFILE_LOCK_ENV = "GITEA_SESSION_PROFILE_LOCK"
 def _export_session_profile_lock():
    """Export this process's launch profile for child CLI processes.
    setdefault: an already-locked environment (outer session) wins, so a
    nested launch cannot relabel the session.
    """
    try:
        name = (get_profile().get("profile_name") or "").strip()
        if name:
            os.environ.setdefault(SESSION_PROFILE_LOCK_ENV, name)
    except Exception:
        # Profile resolution problems surface loudly on the first real call;
        # the lock export must not mask them here at import time.
        pass
 def record_mutation_authority(profile_name: str | None, identity: str | None,
                              remote: str | None, task: str | None):
    """Record the resolved capability context for this process (fail-closed
    consumers in verify_mutation_authority)."""
    global _MUTATION_AUTHORITY
    _MUTATION_AUTHORITY = {
        "initial_profile": profile_name,
        "initial_identity": identity,
        "current_profile": profile_name,
        "current_identity": identity,
        "remote": remote,
        "task": task,
        "role_pivot_authorized": False,
        "role_pivot_record": None,
        "pid": os.getpid(),
    }
 def verify_mutation_authority(remote: str | None, host: str | None = None,
                              required_role: str = "reviewer",
                              active_identity: str | None = None):
    """Verify the current mutation matches this process's recorded authority.
    Fail-closed rules:
    - No recorded authority (or one from another process after a fork) is
      seeded from the live, config-resolved context — the approved preflight
      path (whoami → eligibility → mutation) therefore works without an
      explicit resolve call — but an unresolvable profile still fails closed.
    - GITEA_SESSION_PROFILE_LOCK (set by the launching session) must match
      the active profile: a mid-session GITEA_MCP_PROFILE override flips the
      active profile away from the lock and is refused.
    - Remote, profile, and identity must match the recorded authority.
    - An author→reviewer pivot requires an authorized pivot record
      (gitea_activate_profile in dynamic mode); it can never be improvised.
    """
    global _MUTATION_AUTHORITY
    profile = get_profile()
    active_profile = profile.get("profile_name")
    if active_identity is None:
        # Callers that already proved the identity (eligibility gate) pass it
        # in; otherwise resolve it here (cached, read-only).
        h = host or (REMOTES.get(remote, {}).get("host") if remote in REMOTES else None)
        active_identity = _authenticated_username(h) if h else None
    if not active_profile:
        raise RuntimeError(
            "Mutation authority unavailable: active profile unresolved (fail closed)"
        )
    session_lock = (os.environ.get(SESSION_PROFILE_LOCK_ENV) or "").strip()
    if session_lock and session_lock != active_profile:
        raise RuntimeError(
            f"Active profile '{active_profile}' does not match the session "
            f"profile lock '{session_lock}' — profile side-channel override "
            "rejected (fail closed)"
        )
    data = _MUTATION_AUTHORITY
    if data is None or data.get("pid") != os.getpid():
        # First mutation gate in this process (approved preflight path):
        # seed the authority from the live context, then verify against it.
        record_mutation_authority(
            active_profile, active_identity, remote, "seeded-at-mutation-gate"
        )
        data = _MUTATION_AUTHORITY
    if data.get("remote") != remote:
        raise RuntimeError(
            f"Mutation remote '{remote}' does not match locked remote "
            f"'{data.get('remote')}' (fail closed)"
        )
    locked_profile = data.get("current_profile")
    locked_identity = data.get("current_identity")
    if active_profile != locked_profile or active_identity != locked_identity:
        raise RuntimeError(
            f"Mutation profile '{active_profile}' or identity '{active_identity}' "
            f"does not match locked authority (profile: '{locked_profile}', "
            f"identity: '{locked_identity}') (fail closed)"
        )
    # Reviewer/author role pivot boundary: only an authorized pivot
    # (recorded by gitea_activate_profile) may cross author → reviewer.
    if (required_role == "reviewer"
            and "author" in str(data.get("initial_profile")).lower()
            and "reviewer" in str(active_profile).lower()):
        if not data.get("role_pivot_authorized"):
            raise RuntimeError(
                "Attempted reviewer mutation from author session without "
                "authorized role pivot (fail closed)"
            )
 # Resolve the project root. MCP clients must launch this script directly with
 # the venv interpreter (venv/bin/python3) — see the config example above. We do
 # NOT os.execv() to re-point the interpreter: replacing the process after the
@@ -171,8 +46,6 @@ from gitea_auth import (  # noqa: E402
 import gitea_audit  # noqa: E402
 import gitea_config  # noqa: E402
 import capability_stop_terminal  # noqa: E402
 import issue_duplicate_gate  # noqa: E402
 import role_session_router  # noqa: E402
 def _reveal_endpoints() -> bool:
@@ -438,8 +311,6 @@ def gitea_create_issue(
    host: str | None = None,
    org: str | None = None,
    repo: str | None = None,
    allow_duplicate_override: bool = False,
    split_from_issue: int | None = None,
 ) -> dict:
    """Create a new issue on a Gitea repository.
@@ -450,51 +321,15 @@ def gitea_create_issue(
        host: Override the Gitea host.
        org: Override the owner/organization.
        repo: Override the repository name.
        allow_duplicate_override: Operator-approved split after duplicate found.
        split_from_issue: Existing duplicate issue number when overriding.
    Returns:
        dict with 'number' of the created issue ('url' only with the reveal opt-in).
    """
    ok, block_reasons = role_session_router.check_author_mutation_after_reviewer_stop(
        "create_issue"
    )
    if not ok:
        return {
            "success": False,
            "performed": False,
            "number": None,
            "reasons": block_reasons,
        }
    h, o, r = _resolve(remote, host, org, repo)
    auth = _auth(h)
-    base = repo_api_url(h, o, r)
+    url = f"{repo_api_url(h, o, r)}/issues"
    open_issues = api_get_all(f"{base}/issues?state=open&type=issues", auth)
    closed_issues = api_get_all(
        f"{base}/issues?state=closed&type=issues", auth, limit=100
    )
    gate = issue_duplicate_gate.pre_create_issue_duplicate_gate(
        title,
        list(open_issues) + list(closed_issues),
        allow_override=allow_duplicate_override,
        split_from_issue=split_from_issue,
    )
    if not gate.get("performed"):
        return {
            "success": False,
            "performed": False,
            "number": None,
            "duplicate_gate": gate["verdict"],
            "matches": gate.get("matches", []),
            "reasons": gate.get("reasons", []),
        }
    issue_body = body
    if gate.get("override_applied") and split_from_issue is not None:
        rel = f"Operator-approved split from #{split_from_issue}."
        issue_body = f"{rel}\n\n{body}" if body else rel
    url = f"{base}/issues"
    try:
-        data = api_request("POST", url, auth, {"title": title, "body": issue_body})
+        data = api_request("POST", url, auth, {"title": title, "body": body})
    except Exception as exc:
        _audit("create_issue", host=h, remote=remote, org=o, repo=r,
               result=gitea_audit.FAILED, reason=_redact(str(exc)),
@@ -1023,7 +858,7 @@ def gitea_get_pr_review_feedback(
        'feedback_not_attempted' True, 'reasons', and 'permission_report' —
        deliberately distinct from a successful "no reviews yet" result.
    """
-    reasons = _profile_operation_gate("gitea.read")
+    reasons = _issue_comment_gate("gitea.read")
    if reasons:
        return {
            "success": False,
@@ -1225,17 +1060,6 @@ def gitea_submit_pr_review(
        reasons.append("PR head SHA unavailable (fail closed)")
        return result
    # Gate 5 — in-process mutation authority (#199): the last check before
    # the mutating POST, using the identity the eligibility gate proved.
    # A profile/identity flip or side-channel override between preflight
    # and mutation fails closed here.
    try:
        verify_mutation_authority(remote, host, required_role="reviewer",
                                  active_identity=auth_user)
    except RuntimeError as e:
        reasons.append(str(e))
        return result
    # All gates passed — perform the single mutating call.
    h, o, r = _resolve(remote, host, org, repo)
    try:
@@ -1266,20 +1090,11 @@ def gitea_edit_pr(
 ) -> dict:
    """Edit an existing pull request on a Gitea repository.
    Closing a PR (``state='closed'``) is a distinct capability from other
    edits (#216): it requires the ``gitea.pr.close`` operation (resolver
    task ``close_pr``), fails closed with a structured permission report
    when the active profile lacks it, and is audited as a distinct
    ``close_pr`` action. Title/body/base edits and reopening stay on the
    ordinary edit path, so the edit tool can never be used as an untracked
    close fallback.
    Args:
        pr_number: The pull request index/number (required).
        title: New PR title.
        body: New PR description.
-        state: New state — 'open' or 'closed'. 'closed' requires the
+        state: New state — 'open' or 'closed'.
            ``gitea.pr.close`` capability.
        base: Target branch name.
        remote: Known instance — 'dadeschools' or 'prgs'.
        host: Override the Gitea host.
@@ -1287,10 +1102,7 @@ def gitea_edit_pr(
        repo: Override the repository name.
    Returns:
-        dict with success status and details of the edited PR. A close
+        dict with success status and details of the edited PR.
        attempt without ``gitea.pr.close`` returns 'success'/'performed'
        False with 'reasons' and a structured 'permission_report' and makes
        no API call.
    """
    # Validate inputs BEFORE any auth/profile resolution or API setup: a
    # no-fields call is a pure validation error and must not depend on
@@ -1301,9 +1113,6 @@ def gitea_edit_pr(
    if body is not None:
        payload["body"] = body
    if state is not None:
        if state not in ("open", "closed"):
            raise ValueError(
                f"Invalid state {state!r}: must be 'open' or 'closed' (fail closed).")
        payload["state"] = state
    if base is not None:
        payload["base"] = base
@@ -1311,33 +1120,12 @@ def gitea_edit_pr(
    if not payload:
        raise ValueError("At least one field to edit (title, body, state, base) must be provided.")
    # PR closure is a first-class capability, distinct from retitling or
    # rebasing edits (#216). Gate BEFORE auth/API setup so a blocked close
    # never touches the network.
    closing = payload.get("state") == "closed"
    if closing:
        gate_reasons = _profile_operation_gate("gitea.pr.close")
        if gate_reasons:
            return {
                "success": False,
                "performed": False,
                "pr_number": pr_number,
                "requested_state": "closed",
                "required_permission": "gitea.pr.close",
                "reasons": gate_reasons,
                "permission_report": _permission_block_report("gitea.pr.close"),
            }
    h, o, r = _resolve(remote, host, org, repo)
    auth = _auth(h)
    url = f"{repo_api_url(h, o, r)}/pulls/{pr_number}"
-    request_metadata = {"fields": sorted(payload)}
+    with _audited("edit_pr", host=h, remote=remote, org=o, repo=r,
-    if closing:
+                  pr_number=pr_number, request_metadata={"fields": sorted(payload)}):
        request_metadata["required_permission"] = "gitea.pr.close"
    with _audited("close_pr" if closing else "edit_pr",
                  host=h, remote=remote, org=o, repo=r,
                  pr_number=pr_number, request_metadata=request_metadata):
        data = api_request("PATCH", url, auth, payload)
    cleanup_status = None
@@ -1618,17 +1406,6 @@ def gitea_merge_pr(
        reasons.append("self-merge blocked (authenticated user is PR author)")
        return result
    # Gate 7 — in-process mutation authority (#199): the last check before
    # the merge mutation, using the identity the eligibility gate proved.
    # A profile/identity flip or side-channel override between preflight
    # and merge fails closed here.
    try:
        verify_mutation_authority(remote, host, required_role="reviewer",
                                  active_identity=auth_user)
    except RuntimeError as e:
        reasons.append(str(e))
        return result
    # All gates passed — perform the single merge mutation.
    try:
        auth = _auth(h)
@@ -2109,14 +1886,13 @@ def _permission_block_report(required_operation: str,
    return report
-def _profile_operation_gate(op: str) -> list[str]:
+def _issue_comment_gate(op: str) -> list[str]:
-    """Profile permission check for a single gated operation (#126, #216).
+    """Profile permission check for issue-comment tools (#126).
    Issue discussion comments are gated separately from the gitea.pr.*
    review/merge family: listing requires ``gitea.read``, creating requires
-    ``gitea.issue.comment``. Closing a PR requires the distinct
+    ``gitea.issue.comment``. Returns a list of block reasons (empty = allowed);
-    ``gitea.pr.close`` (#216). Returns a list of block reasons (empty =
+    an unreadable profile fails closed.
    allowed); an unreadable profile fails closed.
    """
    try:
        profile = get_profile()
@@ -2168,7 +1944,7 @@ def gitea_list_issue_comments(
        'success' False, 'reasons', and a structured 'permission_report'
        (#142) with no API call made.
    """
-    reasons = _profile_operation_gate("gitea.read")
+    reasons = _issue_comment_gate("gitea.read")
    if reasons:
        return {"success": False, "issue_number": issue_number,
                "reasons": reasons,
@@ -2230,7 +2006,7 @@ def gitea_create_issue_comment(
        (permission blocks also carry a structured 'permission_report',
        #142).
    """
-    gate_reasons = _profile_operation_gate("gitea.issue.comment")
+    gate_reasons = _issue_comment_gate("gitea.issue.comment")
    reasons = list(gate_reasons)
    if not (body or "").strip():
        reasons.append("comment body must be a non-empty string")
@@ -3228,22 +3004,6 @@ def gitea_activate_profile(
    after_profile = get_profile()["profile_name"]
    after_identity = _authenticated_username(h) if h else None
    # 4.5 Record the authorized pivot in the in-process mutation authority
    # and keep the session profile lock in sync — this is the ONLY path that
    # may authorize an author→reviewer role pivot.
    if _MUTATION_AUTHORITY is not None:
        _MUTATION_AUTHORITY["current_profile"] = after_profile
        _MUTATION_AUTHORITY["current_identity"] = after_identity
        _MUTATION_AUTHORITY["role_pivot_authorized"] = True
        _MUTATION_AUTHORITY["role_pivot_record"] = {
            "from_profile": before_profile,
            "to_profile": after_profile,
            "from_identity": before_identity,
            "to_identity": after_identity,
        }
    if os.environ.get(SESSION_PROFILE_LOCK_ENV) and after_profile:
        os.environ[SESSION_PROFILE_LOCK_ENV] = after_profile
    # 5. Audit the switch if auditing is on
    _audit(
        "activate_profile",
@@ -3561,60 +3321,6 @@ def build_validation_report(commands: list[dict]) -> dict:
    }
@mcp.tool()
 def gitea_route_task_session(
    task_type: str,
    remote: str = "dadeschools",
    host: str | None = None,
 ) -> dict:
    """Pre-task role/session router (#206).
    Classify *task_type* against the active MCP profile before any mutation.
    Returns ``route_result`` — only ``allowed_current_session`` permits
    downstream tool use. Reviewer tasks under an author-bound session return
    ``wrong_role_stop`` with no fallback to author mutations.
    """
    task_type = (task_type or "").strip()
    profile = get_profile()
    allowed = profile.get("allowed_operations") or []
    forbidden = profile.get("forbidden_operations") or []
    active_role = _role_kind(allowed, forbidden)
    if not task_type:
        return role_session_router.route_task_session(
            "",
            active_profile=profile["profile_name"],
            active_role_kind=active_role,
            allowed_in_current_session=False,
        )
    capability = None
    try:
        capability = gitea_resolve_task_capability(
            task=task_type,
            remote=remote,
            host=host,
        )
    except ValueError:
        capability = None
    if capability is not None:
        return role_session_router.route_task_session(
            task_type,
            active_profile=capability["active_profile"],
            active_role_kind=active_role,
            allowed_in_current_session=capability["allowed_in_current_session"],
            runtime_switching_supported=capability["runtime_switching_supported"],
        )
    return role_session_router.route_task_session(
        task_type,
        active_profile=profile["profile_name"],
        active_role_kind=active_role,
        allowed_in_current_session=False,
    )
@mcp.tool()
 def gitea_resolve_task_capability(
    task: str,
@@ -3664,14 +3370,6 @@ def gitea_resolve_task_capability(
            "permission": "gitea.pr.comment",
            "role": "author",
        },
        # PR closure is a first-class capability (#216): distinct from
        # comment_pr (gitea.pr.comment) and from ordinary PR edits, which
        # need no dedicated capability. gitea_edit_pr(state='closed') is
        # gated on the same operation, so no edit-path fallback exists.
        "close_pr": {
            "permission": "gitea.pr.close",
            "role": "author",
        },
        "address_pr_change_requests": {
            "permission": "gitea.branch.push",
            "role": "author",
@@ -3684,18 +3382,6 @@ def gitea_resolve_task_capability(
            "permission": "gitea.pr.merge",
            "role": "reviewer",
        },
        "blind_pr_queue_review": {
            "permission": "gitea.pr.review",
            "role": "reviewer",
        },
        "request_changes_pr": {
            "permission": "gitea.pr.request_changes",
            "role": "reviewer",
        },
        "approve_pr": {
            "permission": "gitea.pr.approve",
            "role": "reviewer",
        },
        "delete_branch": {
            "permission": "gitea.branch.delete",
            "role": "author",
@@ -3807,8 +3493,6 @@ def gitea_resolve_task_capability(
            "STOP: the active profile cannot perform the requested task; "
            "follow exact_safe_next_action instead of improvising.")
    record_mutation_authority(profile["profile_name"], username, remote if remote in REMOTES else None, task)
    result = {
        "requested_task": task,
        "required_operation_permission": required_permission,
@@ -3856,8 +3540,4 @@ def gitea_capability_stop_terminal_report() -> dict:
 # ── Entry point ───────────────────────────────────────────────────────────────
 if __name__ == "__main__":
    # Lock this session's launch profile into the environment so child CLI
    # processes (e.g. review_pr.py) can detect and refuse profile
    # side-channel overrides (#199).
    _export_session_profile_lock()
    mcp.run(transport="stdio")
@@ -24,7 +24,7 @@ venv_python = os.path.join(PROJECT_ROOT, "venv", "bin", "python3")
 if os.path.exists(venv_python) and sys.executable != venv_python:
    os.execv(venv_python, [venv_python] + sys.argv)
-from gitea_auth import get_auth_header, resolve_remote, add_remote_args, api_request, repo_api_url, get_profile
+from gitea_auth import get_auth_header, resolve_remote, add_remote_args, api_request, repo_api_url
 def main(argv=None):
@@ -60,32 +60,6 @@ def main(argv=None):
    host, org, repo = resolve_remote(args)
    # ── Reviewer mutation side-channel wall (#199, refs #194) ──
    # The launching MCP session exports GITEA_SESSION_PROFILE_LOCK with the
    # profile it was started with; child processes inherit it. If this CLI
    # resolves a different profile — e.g. an ad-hoc GITEA_MCP_PROFILE
    # override escalating an author-bound session to reviewer — refuse
    # before any API call. No lock in the environment means no session
    # context (direct operator CLI use), which stays allowed. Unlike a /tmp
    # lock file, the environment is per-process-tree: other sessions cannot
    # spoof it and it cannot go stale across sessions.
    session_lock = (os.environ.get("GITEA_SESSION_PROFILE_LOCK") or "").strip()
    if session_lock:
        try:
            cli_profile = (get_profile().get("profile_name") or "").strip()
        except Exception as e:
            print(f"Mutation authority check failed: {e}", file=sys.stderr)
            return 3
        if cli_profile != session_lock:
            print(
                f"Mismatched active profile vs session profile lock "
                f"(CLI override rejected): CLI profile '{cli_profile}' does "
                f"not match locked session profile '{session_lock}' "
                f"(fail closed)",
                file=sys.stderr,
            )
            return 3
    body = args.body
    if args.body_file:
        if args.body_file == "-":
@@ -16,8 +16,6 @@ here weakens or replaces them.
 import re
 import issue_duplicate_gate
 _FULL_SHA = re.compile(r"^[0-9a-f]{40}$")
@@ -854,43 +852,6 @@ def assess_controller_handoff(report_text, role=None):
    }
 ROUTE_HANDOFF_FIELDS = (
    ("Task type", ("task type", "task_type")),
    ("Required role", ("required role", "required_role")),
    ("Active role", ("active role", "active_role")),
    ("Route result", ("route result", "route_result")),
 )
 def assess_role_route_handoff(report_text, route_result=None):
    """Issue #206: final handoff must record role routing verdict."""
    text = report_text or ""
    lower = text.lower()
    missing = []
    for name, aliases in ROUTE_HANDOFF_FIELDS:
        if not any(alias in lower for alias in aliases):
            missing.append(name)
    if route_result is not None:
        expected = str(route_result.get("route_result", "")).lower()
        if expected and expected not in lower:
            missing.append("route result value")
    if missing:
        return {
            "complete": False,
            "downgraded": True,
            "missing_fields": missing,
            "reasons": [
                f"handoff missing role-route field: {field}" for field in missing
            ],
        }
    return {
        "complete": True,
        "downgraded": False,
        "missing_fields": [],
        "reasons": [],
    }
 def assess_capability_stop_terminal_report(
    report_text,
    *,
@@ -999,10 +960,3 @@ def pr_inventory_trust_gate(
        "reasons": [],
        "corroborated": corroborated,
    }
 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(
        report_text, matches
    )
@@ -1,195 +0,0 @@
 """Pre-task role/session router (#206).
 Classifies a declared task type against the active MCP profile/session and
 returns a route result before any downstream mutation tools run.
 """
 from __future__ import annotations
 ROUTE_ALLOWED = "allowed_current_session"
 ROUTE_WRONG_ROLE = "wrong_role_stop"
 ROUTE_TO_AUTHOR = "route_to_author_session"
 ROUTE_TO_REVIEWER = "route_to_reviewer_session"
 ROUTE_AMBIGUOUS = "ambiguous_task_stop"
 REVIEWER_TASKS = frozenset({
    "review_pr",
    "merge_pr",
    "blind_pr_queue_review",
    "request_changes_pr",
    "approve_pr",
 })
 AUTHOR_TASKS = frozenset({
    "create_issue",
    "comment_issue",
    "close_issue",
    "claim_issue",
    "create_branch",
    "push_branch",
    "create_pr",
    "comment_pr",
    "address_pr_change_requests",
    "delete_branch",
 })
 TASK_REQUIRED_ROLE = {
    "create_issue": "author",
    "comment_issue": "author",
    "close_issue": "author",
    "claim_issue": "author",
    "create_branch": "author",
    "push_branch": "author",
    "create_pr": "author",
    "comment_pr": "author",
    "address_pr_change_requests": "author",
    "delete_branch": "author",
    "review_pr": "reviewer",
    "merge_pr": "reviewer",
    "blind_pr_queue_review": "reviewer",
    "request_changes_pr": "reviewer",
    "approve_pr": "reviewer",
 }
 WRONG_ROLE_REVIEWER_MSG = (
    "Wrong role/session for reviewer task. Launch reviewer MCP namespace."
 )
 _session_last_route: dict | None = None
 def required_role_for_task(task_type: str) -> str | None:
    return TASK_REQUIRED_ROLE.get((task_type or "").strip())
 def route_task_session(
    task_type: str,
    *,
    active_profile: str,
    active_role_kind: str,
    allowed_in_current_session: bool,
    runtime_switching_supported: bool = False,
 ) -> dict:
    """Return routing verdict for *task_type* under the active session."""
    task_type = (task_type or "").strip()
    required_role = required_role_for_task(task_type)
    if required_role is None:
        result = {
            "task_type": task_type,
            "required_role": None,
            "active_role": active_role_kind,
            "active_profile": active_profile,
            "route_result": ROUTE_AMBIGUOUS,
            "downstream_allowed": False,
            "reasons": [
                f"unknown task type '{task_type}'; cannot route session "
                "(fail closed)"
            ],
            "message": (
                "Ambiguous task type; relaunch with an explicit task before "
                "any tool use."
            ),
        }
        _record_route(result)
        return result
    if allowed_in_current_session:
        result = {
            "task_type": task_type,
            "required_role": required_role,
            "active_role": active_role_kind,
            "active_profile": active_profile,
            "route_result": ROUTE_ALLOWED,
            "downstream_allowed": True,
            "reasons": [],
            "message": "Task role matches active session; proceed.",
        }
        _record_route(result)
        return result
    if required_role == "reviewer":
        result = {
            "task_type": task_type,
            "required_role": required_role,
            "active_role": active_role_kind,
            "active_profile": active_profile,
            "route_result": ROUTE_WRONG_ROLE,
            "downstream_allowed": False,
            "reasons": [
                WRONG_ROLE_REVIEWER_MSG,
                "Reviewer tasks cannot run in author-bound sessions.",
                "Static-profile mode does not permit in-place role switching.",
            ],
            "message": WRONG_ROLE_REVIEWER_MSG,
            "runtime_switching_supported": runtime_switching_supported,
            "profile_switch_blocked": not runtime_switching_supported,
        }
        _record_route(result)
        return result
    if required_role == "author":
        route = ROUTE_TO_AUTHOR
        message = (
            "Wrong role/session for author task. Launch author MCP namespace."
        )
        result = {
            "task_type": task_type,
            "required_role": required_role,
            "active_role": active_role_kind,
            "active_profile": active_profile,
            "route_result": route,
            "downstream_allowed": False,
            "reasons": [message],
            "message": message,
            "runtime_switching_supported": runtime_switching_supported,
            "profile_switch_blocked": not runtime_switching_supported,
        }
        _record_route(result)
        return result
    result = {
        "task_type": task_type,
        "required_role": required_role,
        "active_role": active_role_kind,
        "active_profile": active_profile,
        "route_result": ROUTE_AMBIGUOUS,
        "downstream_allowed": False,
        "reasons": ["unable to classify task role (fail closed)"],
        "message": "Ambiguous task type; stop before any mutation.",
    }
    _record_route(result)
    return result
 def last_route() -> dict | None:
    return _session_last_route
 def clear_route_state():
    global _session_last_route
    _session_last_route = None
 def _record_route(result: dict):
    global _session_last_route
    _session_last_route = dict(result)
 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)."""
    last = _session_last_route
    if not last:
        return True, []
    if last.get("route_result") != ROUTE_WRONG_ROLE:
        return True, []
    if last.get("required_role") != "reviewer":
        return True, []
    if mutation_task in AUTHOR_TASKS:
        return False, [
            WRONG_ROLE_REVIEWER_MSG,
            "Author-side mutations are blocked after a reviewer-task "
            "wrong_role_stop unless the operator explicitly changes the "
            "task and relaunches an author MCP session.",
            f"Attempted fallback mutation: {mutation_task}",
        ]
    return True, []
@@ -1,38 +0,0 @@
 """Shared pytest fixtures for the Gitea-Tools test suite."""
 import sys
 import pytest
 sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
@pytest.fixture(autouse=True)
 def _reset_mutation_authority(monkeypatch):
    """Isolate the in-process mutation authority between tests (#199).
    The mutation-authority gate stays LIVE in every test — this fixture only
    clears the per-process record and the session profile lock so one test's
    seeded authority (or an intentionally mismatched one) cannot leak into
    the next test. It must never replace verify_mutation_authority with a
    no-op: individual tests that need a specific authority state set it up
    explicitly.
    """
    monkeypatch.delenv("GITEA_SESSION_PROFILE_LOCK", raising=False)
    try:
        import mcp_server
    except Exception:
        yield
        return
    monkeypatch.setattr(mcp_server, "_MUTATION_AUTHORITY", None)
    monkeypatch.setattr(mcp_server, "_IDENTITY_CACHE", {})
    try:
        import capability_stop_terminal
        capability_stop_terminal.clear()
    except Exception:
        pass
    yield
    try:
        import capability_stop_terminal
        capability_stop_terminal.clear()
    except Exception:
        pass
@@ -175,12 +175,9 @@ class _AuditWiringBase(unittest.TestCase):
 class TestSimpleToolAudit(_AuditWiringBase):
    @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
           return_value=(True, []))
    @patch("mcp_server.api_request")
    @patch("mcp_server.api_get_all", return_value=[])
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
-    def test_create_issue_success_audited(self, _auth, _get_all, mock_api, _role):
+    def test_create_issue_success_audited(self, _auth, mock_api):
        # 1: create POST result, 2: identity /user lookup for the audit record.
        mock_api.side_effect = [
            {"number": 11, "html_url": "https://gitea.prgs.cc/issues/11"},
@@ -199,12 +196,9 @@ class TestSimpleToolAudit(_AuditWiringBase):
        self.assertEqual(rec["issue_number"], 11)
        self.assertEqual(rec["request_metadata"]["title"], "Add thing")
    @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
           return_value=(True, []))
    @patch("mcp_server.api_request")
    @patch("mcp_server.api_get_all", return_value=[])
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
-    def test_create_issue_failure_audited(self, _auth, _get_all, mock_api, _role):
+    def test_create_issue_failure_audited(self, _auth, mock_api):
        mock_api.side_effect = [
            RuntimeError("HTTP 500: boom"),
            {"login": "author-bot"},  # identity lookup for the audit record
@@ -229,28 +223,19 @@ class TestSimpleToolAudit(_AuditWiringBase):
        self.assertEqual(recs[0]["issue_number"], 42)
        self.assertEqual(recs[0]["authenticated_username"], "mgr-bot")
    @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
           return_value=(True, []))
    @patch("mcp_server.api_request")
    @patch("mcp_server.api_get_all", return_value=[])
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
-    def test_disabled_writes_nothing_and_no_extra_call(self, _auth, _get_all, mock_api, _role):
+    def test_disabled_writes_nothing_and_no_extra_call(self, _auth, mock_api):
-        # No GITEA_AUDIT_LOG -> audit is a no-op: one create POST, no file.
+        # No GITEA_AUDIT_LOG -> audit is a no-op: exactly one API call, no file.
        mock_api.return_value = {"number": 1, "html_url": "http://x/1"}
        with patch.dict(os.environ, {"GITEA_PROFILE_NAME": "gitea-author"}, clear=True):
            gitea_create_issue(title="x", remote="prgs")
-        issue_posts = [
+        mock_api.assert_called_once()
            c for c in mock_api.call_args_list if c.args[0] == "POST"
        ]
        self.assertEqual(len(issue_posts), 1)
        self.assertEqual(self._records(), [])
    @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
           return_value=(True, []))
    @patch("mcp_server.api_request")
    @patch("mcp_server.api_get_all", return_value=[])
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
-    def test_secrets_never_written(self, _auth, _get_all, mock_api, _role):
+    def test_secrets_never_written(self, _auth, mock_api):
        mock_api.side_effect = [
            {"number": 3, "html_url": "http://x/3"},
            {"login": "author-bot"},
@@ -263,12 +248,9 @@ class TestSimpleToolAudit(_AuditWiringBase):
        for secret in ("super-secret-token", "authorization", "basic ", FAKE_AUTH.lower()):
            self.assertNotIn(secret, blob)
    @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
           return_value=(True, []))
    @patch("mcp_server.api_request")
    @patch("mcp_server.api_get_all", return_value=[])
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
-    def test_audit_failure_never_breaks_action(self, _auth, _get_all, mock_api, _role):
+    def test_audit_failure_never_breaks_action(self, _auth, mock_api):
        mock_api.side_effect = [
            {"number": 9, "html_url": "http://x/9"},
            {"login": "author-bot"},
@@ -1,154 +0,0 @@
 """Tests for the gated PR close path (Issue #216).
 ``gitea_edit_pr(state='closed')`` requires the distinct ``gitea.pr.close``
 capability: without it the close fails closed (no auth lookup, no API call,
 structured permission report). With it — the explicit operator-directed
 contaminated-PR closure path — the close proceeds and is audited as a
 distinct ``close_pr`` action carrying the required capability, so final
 reports can prove exactly which mutation capability was exercised.
 Ordinary edits (title/body/base) and reopening never require the close
 capability: PR comment, PR edit, and PR close remain distinct capabilities.
 """
 import sys
 import unittest
 from unittest.mock import patch
 sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
 import mcp_server
 from mcp_server import gitea_edit_pr
 FAKE_AUTH = "token fake"
 AUTHOR_NO_CLOSE = {
    "profile_name": "prgs-author",
    "allowed_operations": [
        "gitea.read", "gitea.pr.create", "gitea.pr.comment",
        "gitea.branch.push", "gitea.issue.comment",
    ],
    "forbidden_operations": ["gitea.pr.approve", "gitea.pr.merge"],
    "audit_label": "prgs-author",
 }
 AUTHOR_WITH_CLOSE = {
    "profile_name": "prgs-author-closer",
    "allowed_operations": AUTHOR_NO_CLOSE["allowed_operations"] + ["gitea.pr.close"],
    "forbidden_operations": ["gitea.pr.approve", "gitea.pr.merge"],
    "audit_label": "prgs-author-closer",
 }
 class TestEditPrCloseGate(unittest.TestCase):
    def setUp(self):
        self.mock_api = patch("mcp_server.api_request").start()
        self.mock_auth = patch("mcp_server.get_auth_header", return_value=FAKE_AUTH).start()
        # Deterministic permission reports: no real operator config, no switching.
        patch("gitea_config.load_config", return_value={}).start()
        patch("gitea_config.is_runtime_switching_enabled", return_value=False).start()
        patch("gitea_audit.audit_enabled", return_value=False).start()
        mcp_server._IDENTITY_CACHE.clear()
    def tearDown(self):
        patch.stopall()
        mcp_server._IDENTITY_CACHE.clear()
    def _set_profile(self, profile):
        patch("mcp_server.get_profile", return_value=profile).start()
    def test_close_blocked_without_close_capability(self):
        # Author profile without gitea.pr.close: the broad edit path must not
        # be usable as an untracked close fallback.
        self._set_profile(AUTHOR_NO_CLOSE)
        res = gitea_edit_pr(pr_number=205, state="closed", remote="prgs")
        self.assertFalse(res["success"])
        self.assertFalse(res["performed"])
        self.assertEqual(res["requested_state"], "closed")
        self.assertEqual(res["required_permission"], "gitea.pr.close")
        self.assertTrue(res["reasons"])
        report = res["permission_report"]
        self.assertEqual(report["required_permission"], "gitea.pr.close")
        self.assertEqual(report["active_profile"], "prgs-author")
        self.mock_api.assert_not_called()
        self.mock_auth.assert_not_called()
    def test_close_blocked_when_close_forbidden(self):
        profile = dict(AUTHOR_WITH_CLOSE)
        profile["forbidden_operations"] = ["gitea.pr.close"]
        self._set_profile(profile)
        res = gitea_edit_pr(pr_number=205, state="closed", remote="prgs")
        self.assertFalse(res["success"])
        self.assertIn("profile forbids 'gitea.pr.close'", res["reasons"])
        self.mock_api.assert_not_called()
    def test_close_blocked_when_profile_unresolvable(self):
        patch("mcp_server.get_profile", side_effect=RuntimeError("no profile")).start()
        res = gitea_edit_pr(pr_number=205, state="closed", remote="prgs")
        self.assertFalse(res["success"])
        self.assertTrue(any("fail closed" in r for r in res["reasons"]))
        self.mock_api.assert_not_called()
    def test_operator_directed_close_allowed_and_audited(self):
        # Explicit operator-directed contaminated-PR closure: the operator
        # granted gitea.pr.close, so the close proceeds and the audit trail
        # records a distinct close_pr action with the capability proof.
        self._set_profile(AUTHOR_WITH_CLOSE)
        patch("gitea_audit.audit_enabled", return_value=True).start()
        mock_write = patch("gitea_audit.write_event").start()
        def api_side_effect(method, url, auth, payload=None):
            if method == "GET" and "/user" in url:
                return {"login": "jcwalker3"}
            if method == "PATCH" and "pulls/205" in url:
                self.assertEqual(payload["state"], "closed")
                return {
                    "number": 205,
                    "title": "Contaminated PR",
                    "state": "closed",
                    "html_url": "url",
                    "body": "No issue link",
                    "head": {"ref": "feat/invalid-provenance"},
                }
            return {}
        self.mock_api.side_effect = api_side_effect
        res = gitea_edit_pr(pr_number=205, state="closed", remote="prgs")
        self.assertTrue(res["success"])
        self.assertEqual(res["state"], "closed")
        mock_write.assert_called()
        event = mock_write.call_args[0][0]
        self.assertEqual(event["action"], "close_pr")
        self.assertEqual(
            event["request_metadata"]["required_permission"], "gitea.pr.close")
    def test_title_edit_needs_no_close_capability(self):
        # PR edit and PR close are distinct capabilities: retitling stays on
        # the ordinary edit path.
        self._set_profile(AUTHOR_NO_CLOSE)
        self.mock_api.return_value = {
            "number": 7, "title": "Renamed", "state": "open",
            "body": "", "html_url": "u"}
        res = gitea_edit_pr(pr_number=7, title="Renamed", remote="prgs")
        self.assertTrue(res["success"])
    def test_reopen_needs_no_close_capability(self):
        self._set_profile(AUTHOR_NO_CLOSE)
        self.mock_api.return_value = {
            "number": 7, "title": "T", "state": "open",
            "body": "", "html_url": "u"}
        res = gitea_edit_pr(pr_number=7, state="open", remote="prgs")
        self.assertTrue(res["success"])
    def test_invalid_state_fails_closed_before_auth(self):
        # A case-variant state can neither bypass the close gate nor reach
        # the API: it is rejected as pure validation, before auth.
        self._set_profile(AUTHOR_WITH_CLOSE)
        with self.assertRaises(ValueError):
            gitea_edit_pr(pr_number=7, state="CLOSED", remote="prgs")
        self.mock_api.assert_not_called()
        self.mock_auth.assert_not_called()
 if __name__ == "__main__":
    unittest.main()
@@ -1,173 +0,0 @@
 """Tests for pre-create issue duplicate gate (Issue #207)."""
 import sys
 import unittest
 from unittest.mock import patch
 sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
 from issue_duplicate_gate import (  # noqa: E402
    VERDICT_AMBIGUOUS,
    VERDICT_DUPLICATE,
    VERDICT_NO_DUPLICATE,
    assess_duplicate_search_proof,
    assess_pre_create_duplicate,
    normalize_issue_title,
    pre_create_issue_duplicate_gate,
    titles_near_duplicate,
 )
 from mcp_server import gitea_create_issue  # noqa: E402
 from review_proofs import assess_duplicate_search_proof as proof_assess  # noqa: E402
 FAKE_AUTH = "Basic dGVzdDp0ZXN0"
 CANONICAL_TITLE = (
    "Add hard queue-target resolution wall before PR inventory "
    "or empty-queue claims"
 )
 class TestNormalizeAndNearDuplicate(unittest.TestCase):
    def test_normalize_strips_punctuation_and_case(self):
        norm = normalize_issue_title("  Hard-Wall: Queue Target!  ")
        self.assertEqual(norm, "hard wall queue target")
    def test_exact_title_match(self):
        self.assertTrue(titles_near_duplicate(CANONICAL_TITLE, CANONICAL_TITLE))
    def test_punctuation_and_capitalization_only_differs(self):
        proposed = CANONICAL_TITLE.upper().replace("-", " — ")
        self.assertTrue(titles_near_duplicate(proposed, CANONICAL_TITLE))
    def test_hard_wall_vs_wall_wording(self):
        self.assertTrue(
            titles_near_duplicate(
                "Add hard wall before PR inventory",
                "Add wall before PR inventory",
            )
        )
 class TestPreCreateGate(unittest.TestCase):
    def test_blocks_exact_duplicate_title(self):
        existing = [{"number": 201, "title": CANONICAL_TITLE, "state": "open"}]
        gate = assess_pre_create_duplicate(CANONICAL_TITLE, existing)
        self.assertEqual(gate["verdict"], VERDICT_DUPLICATE)
        self.assertFalse(gate["performed"])
        self.assertEqual(gate["matches"][0]["number"], 201)
    def test_final_pre_create_check_blocks_even_if_llm_search_missed(self):
        """TOCTOU: gate re-queries at create time with the live issue list."""
        stale_report = (
            "Searched 20 open issues (#194, #196, PR #195); no duplicate found."
        )
        live_existing = [{"number": 201, "title": CANONICAL_TITLE, "state": "open"}]
        proof = assess_duplicate_search_proof(stale_report, live_existing)
        self.assertFalse(proof["valid"])
        gate = assess_pre_create_duplicate(CANONICAL_TITLE, live_existing)
        self.assertFalse(gate["performed"])
    def test_invalid_proof_when_summary_omits_exact_duplicate(self):
        report = "Reviewed nearby issues #194, #196, and PR #195; no duplicate."
        matches = [{"number": 201, "title": CANONICAL_TITLE}]
        result = assess_duplicate_search_proof(report, matches)
        self.assertFalse(result["valid"])
        self.assertTrue(any("201" in r for r in result["reasons"]))
    def test_operator_split_override_allowed_with_relationship(self):
        existing = [{"number": 201, "title": CANONICAL_TITLE, "state": "open"}]
        gate = pre_create_issue_duplicate_gate(
            CANONICAL_TITLE,
            existing,
            allow_override=True,
            split_from_issue=201,
        )
        self.assertEqual(gate["verdict"], VERDICT_NO_DUPLICATE)
        self.assertTrue(gate["performed"])
        self.assertTrue(gate.get("override_applied"))
    def test_ambiguous_when_too_many_near_matches(self):
        base = "Add hard wall before PR inventory"
        existing = [
            {"number": n, "title": f"{base} variant {n}", "state": "open"}
            for n in range(1, 6)
        ]
        gate = assess_pre_create_duplicate(base, existing)
        self.assertEqual(gate["verdict"], VERDICT_AMBIGUOUS)
        self.assertFalse(gate["performed"])
    def test_no_duplicate_when_unique_title(self):
        gate = assess_pre_create_duplicate(
            "Brand new unique issue title",
            [{"number": 1, "title": "Unrelated backlog cleanup", "state": "open"}],
        )
        self.assertEqual(gate["verdict"], VERDICT_NO_DUPLICATE)
        self.assertTrue(gate["performed"])
 class TestCreateIssueMCPGate(unittest.TestCase):
    @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop")
    @patch("mcp_server.api_request")
    @patch("mcp_server.api_get_all")
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_blocks_normalized_title_match_without_override(
        self, _auth, mock_get_all, mock_api, mock_role_check
    ):
        mock_role_check.return_value = (True, [])
        mock_get_all.return_value = [
            {"number": 201, "title": CANONICAL_TITLE, "state": "open"},
        ]
        result = gitea_create_issue(title=CANONICAL_TITLE.upper(), remote="prgs")
        self.assertFalse(result["performed"])
        self.assertEqual(result["duplicate_gate"], VERDICT_DUPLICATE)
        mock_api.assert_not_called()
    @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop")
    @patch("mcp_server.api_request")
    @patch("mcp_server.api_get_all")
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_override_creates_with_split_relationship_in_body(
        self, _auth, mock_get_all, mock_api, mock_role_check
    ):
        mock_role_check.return_value = (True, [])
        mock_get_all.return_value = [
            {"number": 201, "title": CANONICAL_TITLE, "state": "open"},
        ]
        mock_api.return_value = {"number": 210, "html_url": "https://gitea.prgs.cc/issues/210"}
        result = gitea_create_issue(
            title=CANONICAL_TITLE,
            body="Follow-up scope",
            remote="prgs",
            allow_duplicate_override=True,
            split_from_issue=201,
        )
        self.assertEqual(result["number"], 210)
        payload = mock_api.call_args[0][3]
        self.assertIn("Operator-approved split from #201.", payload["body"])
        self.assertIn("Follow-up scope", payload["body"])
    @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop")
    @patch("mcp_server.api_request")
    @patch("mcp_server.api_get_all", return_value=[])
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
    def test_creates_when_no_duplicate(
        self, _auth, _get_all, mock_api, mock_role_check
    ):
        mock_role_check.return_value = (True, [])
        mock_api.return_value = {"number": 1, "html_url": "https://gitea.example.com/issues/1"}
        result = gitea_create_issue(title="Unique new issue", body="body text")
        self.assertEqual(result["number"], 1)
 class TestReviewProofsWrapper(unittest.TestCase):
    def test_review_proofs_reexports_duplicate_search_validator(self):
        report = "Checked open issues; nothing similar."
        matches = [{"number": 200, "title": CANONICAL_TITLE}]
        result = proof_assess(report, matches)
        self.assertFalse(result["valid"])
 if __name__ == "__main__":
    unittest.main()
@@ -37,8 +37,6 @@ from mcp_server import (  # noqa: E402
 from gitea_auth import get_profile  # noqa: E402
 import gitea_config  # noqa: E402
 import mcp_server
 FAKE_AUTH = "Basic dGVzdDp0ZXN0"
@@ -47,12 +45,9 @@ FAKE_AUTH = "Basic dGVzdDp0ZXN0"
 # ---------------------------------------------------------------------------
 class TestCreateIssue(unittest.TestCase):
    @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
           return_value=(True, []))
    @patch("mcp_server.api_request")
    @patch("mcp_server.api_get_all", return_value=[])
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
-    def test_creates_issue(self, _auth, _get_all, mock_api, _role):
+    def test_creates_issue(self, _auth, mock_api):
        mock_api.return_value = {"number": 1, "html_url": "https://gitea.example.com/issues/1"}
        with patch.dict(os.environ, {}, clear=True):
            result = gitea_create_issue(title="Test issue", body="body text")
@@ -64,23 +59,17 @@ class TestCreateIssue(unittest.TestCase):
        self.assertEqual(call_args[0][0], "POST")
        self.assertEqual(call_args[0][3]["title"], "Test issue")
    @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
           return_value=(True, []))
    @patch("mcp_server.api_request")
    @patch("mcp_server.api_get_all", return_value=[])
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
-    def test_create_issue_reveal_opt_in_includes_url(self, _auth, _get_all, mock_api, _role):
+    def test_create_issue_reveal_opt_in_includes_url(self, _auth, mock_api):
        mock_api.return_value = {"number": 1, "html_url": "https://gitea.example.com/issues/1"}
        with patch.dict(os.environ, {"GITEA_MCP_REVEAL_ENDPOINTS": "1"}, clear=True):
            result = gitea_create_issue(title="Test issue", body="body text")
        self.assertIn("issues/1", result["url"])
    @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
           return_value=(True, []))
    @patch("mcp_server.api_request")
    @patch("mcp_server.api_get_all", return_value=[])
    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
-    def test_creates_on_prgs(self, _auth, _get_all, mock_api, _role):
+    def test_creates_on_prgs(self, _auth, mock_api):
        mock_api.return_value = {"number": 5, "html_url": "https://gitea.prgs.cc/issues/5"}
        result = gitea_create_issue(title="Test", remote="prgs")
        self.assertEqual(result["number"], 5)
@@ -1632,8 +1621,7 @@ class TestTrackerHygieneCleanup(unittest.TestCase):
        self.mock_auth = patch("mcp_server.get_auth_header", return_value=FAKE_AUTH).start()
        patch("gitea_audit.audit_enabled", return_value=True).start()
        self.mock_audit = patch("gitea_audit.write_event").start()
-        # gitea.pr.close: closing a PR via gitea_edit_pr is capability-gated (#216).
+        patch("mcp_server.get_profile", return_value={"profile_name": "test", "allowed_operations": ["merge", "edit", "close"], "audit_label": "test", "forbidden_operations": []}).start()
        patch("mcp_server.get_profile", return_value={"profile_name": "test", "allowed_operations": ["merge", "edit", "close", "gitea.pr.close"], "audit_label": "test", "forbidden_operations": []}).start()
    def tearDown(self):
        patch.stopall()
@@ -2308,122 +2296,3 @@ class TestIssueCommentPermissionSeparation(unittest.TestCase):
            "gitea.issue.comment", reviewer["allowed_operations"],
            reviewer.get("forbidden_operations", []))
        self.assertTrue(ok)
 class TestVerifyMutationAuthority(unittest.TestCase):
    """In-process mutation authority (#199, refs #194).
    The authority record lives in mcp_server._MUTATION_AUTHORITY (per
    process, reset between tests by conftest); the CLI side-channel is
    covered by the GITEA_SESSION_PROFILE_LOCK environment lock. There is no
    lock file — nothing here touches /tmp.
    """
    def setUp(self):
        self.patch_profile = patch("mcp_server.get_profile")
        self.mock_profile = self.patch_profile.start()
        self.patch_username = patch("mcp_server._authenticated_username")
        self.mock_username = self.patch_username.start()
        self.mock_profile.return_value = {"profile_name": "prgs-reviewer"}
        self.mock_username.return_value = "sysadmin"
    def tearDown(self):
        self.patch_profile.stop()
        self.patch_username.stop()
    def _authority(self, **overrides):
        data = {
            "initial_profile": "prgs-reviewer",
            "initial_identity": "sysadmin",
            "current_profile": "prgs-reviewer",
            "current_identity": "sysadmin",
            "remote": "prgs",
            "task": "review_pr",
            "role_pivot_authorized": False,
            "role_pivot_record": None,
            "pid": os.getpid(),
        }
        data.update(overrides)
        mcp_server._MUTATION_AUTHORITY = data
    def test_missing_authority_seeds_from_live_context(self):
        # Approved preflight path (whoami → eligibility → mutation): the
        # first mutation gate seeds the authority instead of failing closed,
        # so the standard reviewer workflow keeps working.
        mcp_server._MUTATION_AUTHORITY = None
        mcp_server.verify_mutation_authority("prgs")
        seeded = mcp_server._MUTATION_AUTHORITY
        self.assertIsNotNone(seeded)
        self.assertEqual(seeded["current_profile"], "prgs-reviewer")
        self.assertEqual(seeded["current_identity"], "sysadmin")
        self.assertEqual(seeded["remote"], "prgs")
    def test_unresolved_profile_fails_closed(self):
        self.mock_profile.return_value = {}
        mcp_server._MUTATION_AUTHORITY = None
        with self.assertRaises(RuntimeError) as ctx:
            mcp_server.verify_mutation_authority("prgs")
        self.assertIn("profile unresolved", str(ctx.exception))
    def test_mismatched_remote_fails(self):
        self._authority(remote="dadeschools")
        with self.assertRaises(RuntimeError) as ctx:
            mcp_server.verify_mutation_authority("prgs")
        self.assertIn("does not match locked remote", str(ctx.exception))
    def test_profile_flip_after_record_fails(self):
        # Authority was recorded as author; the active profile now resolves
        # as reviewer (e.g. an env-var flip mid-session) — refuse.
        self._authority(
            initial_profile="prgs-author",
            initial_identity="jcwalker3",
            current_profile="prgs-author",
            current_identity="jcwalker3",
        )
        with self.assertRaises(RuntimeError) as ctx:
            mcp_server.verify_mutation_authority("prgs")
        self.assertIn("does not match locked authority", str(ctx.exception))
    def test_session_lock_env_mismatch_fails(self):
        # The launching session locked the environment to the author
        # profile; the active profile resolves as reviewer — side-channel
        # override rejected even with a matching in-process authority.
        self._authority()
        with patch.dict(os.environ,
                        {"GITEA_SESSION_PROFILE_LOCK": "prgs-author"}):
            with self.assertRaises(RuntimeError) as ctx:
                mcp_server.verify_mutation_authority("prgs")
        self.assertIn("side-channel override rejected", str(ctx.exception))
    def test_foreign_pid_authority_is_not_trusted(self):
        # An authority record from another process (fork leftovers) is
        # discarded and reseeded from the live context, never reused.
        self._authority(current_profile="prgs-author", pid=os.getpid() + 1)
        mcp_server.verify_mutation_authority("prgs")
        self.assertEqual(
            mcp_server._MUTATION_AUTHORITY["current_profile"], "prgs-reviewer"
        )
        self.assertEqual(mcp_server._MUTATION_AUTHORITY["pid"], os.getpid())
    def test_author_to_reviewer_pivot_blocked_without_authorization(self):
        self._authority(
            initial_profile="prgs-author",
            initial_identity="jcwalker3",
        )
        with self.assertRaises(RuntimeError) as ctx:
            mcp_server.verify_mutation_authority("prgs", required_role="reviewer")
        self.assertIn("without authorized role pivot", str(ctx.exception))
    def test_authorized_pivot_is_allowed(self):
        self._authority(
            initial_profile="prgs-author",
            initial_identity="jcwalker3",
            role_pivot_authorized=True,
        )
        mcp_server.verify_mutation_authority("prgs", required_role="reviewer")
    def test_allowed_when_match(self):
        self._authority()
        with patch.dict(os.environ,
                        {"GITEA_SESSION_PROFILE_LOCK": "prgs-reviewer"}):
            mcp_server.verify_mutation_authority("prgs")
@@ -203,67 +203,5 @@ class TestResolveTaskCapability(unittest.TestCase):
            reasons = res.get("reasons", [])
            self.assertTrue(any("author" in str(r).lower() for r in reasons) or len(reasons) > 0)
    # Issue #216: close_pr is a first-class resolver task gated on gitea.pr.close.
    @patch("mcp_server.api_request", return_value={"login": "author-user"})
    @patch("mcp_server.get_auth_header", return_value="token author-pass")
    def test_resolve_close_pr_author_profile_without_close_blocked(self, _auth, _api):
        # Default author profile has PR create/comment but not close: the
        # close capability must never be implied by broader author ops.
        with patch.dict(os.environ, self._env("author-profile")):
            res = mcp_server.gitea_resolve_task_capability(task="close_pr", remote="prgs")
            self.assertEqual(res["requested_task"], "close_pr")
            self.assertEqual(res["required_operation_permission"], "gitea.pr.close")
            self.assertEqual(res["required_role_kind"], "author")
            self.assertFalse(res["allowed_in_current_session"])
            self.assertTrue(res["stop_required"])
            self.assertEqual(res["matching_configured_profile"], [])
            self.assertTrue(res["different_mcp_namespace_required"])
    @patch("mcp_server.api_request", return_value={"login": "author-user"})
    @patch("mcp_server.get_auth_header", return_value="token author-pass")
    def test_resolve_close_pr_author_profile_with_close_allowed(self, _auth, _api):
        # Operator-granted close capability resolves cleanly under an author profile.
        config = json.loads(json.dumps(CONFIG_RESOLVER))
        config["profiles"]["author-closer"] = {
            "enabled": True,
            "context": "ctx",
            "role": "author",
            "username": "author-user",
            "auth": {"type": "env", "name": "GITEA_TOKEN_AUTHOR"},
            "allowed_operations": ["gitea.read", "gitea.pr.close"],
            "forbidden_operations": ["gitea.pr.approve", "gitea.pr.merge"],
            "execution_profile": "author-closer",
        }
        self._write_config(config)
        with patch.dict(os.environ, self._env("author-closer")):
            res = mcp_server.gitea_resolve_task_capability(task="close_pr", remote="prgs")
            self.assertTrue(res["allowed_in_current_session"])
            self.assertFalse(res["stop_required"])
            self.assertIn("author-closer", res["matching_configured_profile"])
            self.assertIn("ready for operations", res["exact_safe_next_action"])
    @patch("mcp_server.api_request", return_value={"login": "reviewer-user"})
    @patch("mcp_server.get_auth_header", return_value="token reviewer-pass")
    def test_resolve_close_pr_reviewer_profile_blocked(self, _auth, _api):
        # close_pr is author-side: a reviewer profile must be told to stop.
        with patch.dict(os.environ, self._env("reviewer-profile")):
            res = mcp_server.gitea_resolve_task_capability(task="close_pr", remote="prgs")
            self.assertFalse(res["allowed_in_current_session"])
            self.assertTrue(res["stop_required"])
            self.assertEqual(res["required_role_kind"], "author")
            self.assertIn("author", res["exact_safe_next_action"].lower())
    @patch("mcp_server.api_request", return_value={"login": "author-user"})
    @patch("mcp_server.get_auth_header", return_value="token author-pass")
    def test_close_pr_known_and_lookalike_tasks_still_fail_closed(self, _auth, _api):
        # close_pr resolves (no Unknown-task error); near-miss spellings keep
        # failing closed so no untracked close fallback can be rationalized.
        with patch.dict(os.environ, self._env("author-profile")):
            res = mcp_server.gitea_resolve_task_capability(task="close_pr", remote="prgs")
            self.assertEqual(res["required_operation_permission"], "gitea.pr.close")
            for unknown in ("close_pull_request", "close", "pr_close"):
                with self.assertRaises(ValueError):
                    mcp_server.gitea_resolve_task_capability(task=unknown, remote="prgs")
 if __name__ == "__main__":
    unittest.main()
@@ -2,8 +2,6 @@
 Mocks api_request and credentials.
 """
 import io
 import os
 import sys
 import unittest
 from unittest.mock import patch
@@ -29,11 +27,6 @@ FAKE_PR_DATA = {
 class TestArgParsing(unittest.TestCase):
    def setUp(self):
        self.exists_patcher = patch("os.path.exists", return_value=False)
        self.exists_patcher.start()
        self.addCleanup(self.exists_patcher.stop)
    @patch("review_pr.get_auth_header", return_value=FAKE_CREDS)
    def test_missing_pr_number_exits(self, _auth):
        with self.assertRaises(SystemExit):
@@ -42,11 +35,6 @@ class TestArgParsing(unittest.TestCase):
 class TestAPIPayload(unittest.TestCase):
    def setUp(self):
        self.exists_patcher = patch("os.path.exists", return_value=False)
        self.exists_patcher.start()
        self.addCleanup(self.exists_patcher.stop)
    @patch("review_pr.api_request")
    @patch("review_pr.get_auth_header", return_value=FAKE_CREDS)
    def test_payload_fields_and_workflow(self, _auth, mock_api):
@@ -111,55 +99,5 @@ class TestAPIPayload(unittest.TestCase):
        self.assertIn("gitea_merge_pr", msg)
 class TestMutationAuthorityLock(unittest.TestCase):
    """#199 (refs #194): the CLI refuses to run under a profile that differs
    from the session profile lock exported by the launching MCP session."""
    @patch("review_pr.get_profile")
    def test_cli_blocked_on_session_lock_mismatch(self, mock_get_profile):
        # An author-bound session exported the lock; the CLI resolves a
        # reviewer profile (GITEA_MCP_PROFILE side-channel override) — reject
        # before any API call.
        mock_get_profile.return_value = {"profile_name": "prgs-reviewer"}
        import io
        buf = io.StringIO()
        with patch.dict(os.environ,
                        {"GITEA_SESSION_PROFILE_LOCK": "prgs-author"}), \
             patch.object(sys, "stderr", buf):
            rc = review_pr.main([
                "--pr-number", "81", "--event", "APPROVE",
            ])
        self.assertEqual(rc, 3)
        msg = buf.getvalue().lower()
        self.assertIn("cli override rejected", msg)
    @patch("review_pr.get_profile")
    @patch("review_pr.api_request")
    @patch("review_pr.get_auth_header", return_value=FAKE_CREDS)
    def test_cli_allowed_on_profile_match(self, _auth, mock_api, mock_get_profile):
        mock_get_profile.return_value = {"profile_name": "prgs-reviewer"}
        mock_api.side_effect = [FAKE_PR_DATA, {}]
        with patch.dict(os.environ,
                        {"GITEA_SESSION_PROFILE_LOCK": "prgs-reviewer"}):
            rc = review_pr.main([
                "--pr-number", "81", "--event", "APPROVE",
            ])
        self.assertEqual(rc, 0)
    @patch("review_pr.api_request")
    @patch("review_pr.get_auth_header", return_value=FAKE_CREDS)
    def test_cli_allowed_without_session_lock(self, _auth, mock_api):
        # No lock in the environment = direct operator CLI use; the wall
        # does not apply and the normal flow proceeds.
        mock_api.side_effect = [FAKE_PR_DATA, {}]
        env = {k: v for k, v in os.environ.items()
               if k != "GITEA_SESSION_PROFILE_LOCK"}
        with patch.dict(os.environ, env, clear=True):
            rc = review_pr.main([
                "--pr-number", "81", "--event", "APPROVE",
            ])
        self.assertEqual(rc, 0)
 if __name__ == "__main__":
    unittest.main()
@@ -1,189 +0,0 @@
 """Tests for pre-task role/session router (#206)."""
 import json
 import os
 import sys
 import tempfile
 import unittest
 from unittest.mock import patch
 sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
 import gitea_config
 import mcp_server
 import role_session_router
 from review_proofs import assess_role_route_handoff
 CONFIG = {
    "version": 2,
    "contexts": {
        "ctx": {
            "enabled": True,
            "gitea": {"enabled": True, "base_url": "https://gitea.example.com"},
        }
    },
    "profiles": {
        "prgs-author": {
            "enabled": True,
            "context": "ctx",
            "role": "author",
            "username": "jcwalker3",
            "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", "gitea.pr.review",
            ],
            "execution_profile": "prgs-author",
        },
        "prgs-reviewer": {
            "enabled": True,
            "context": "ctx",
            "role": "reviewer",
            "username": "sysadmin",
            "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", "gitea.issue.create",
            ],
            "execution_profile": "prgs-reviewer",
        },
    },
    "rules": {"allow_runtime_switching": False},
 }
 class TestRoleSessionRouter(unittest.TestCase):
    def setUp(self):
        role_session_router.clear_route_state()
        self._remotes = patch.dict(mcp_server.REMOTES, {
            "prgs": {
                "host": "gitea.example.com",
                "org": "Scaled-Tech-Consulting",
                "repo": "Gitea-Tools",
            },
        })
        self._remotes.start()
        mcp_server._IDENTITY_CACHE.clear()
        gitea_config._active_profile_override = None
        self._dir = tempfile.TemporaryDirectory()
        self.config_path = os.path.join(self._dir.name, "profiles.json")
        with open(self.config_path, "w", encoding="utf-8") as fh:
            fh.write(json.dumps(CONFIG))
    def tearDown(self):
        self._remotes.stop()
        role_session_router.clear_route_state()
        mcp_server._IDENTITY_CACHE.clear()
        gitea_config._active_profile_override = None
        self._dir.cleanup()
    def _env(self, profile):
        return {
            "GITEA_MCP_CONFIG": self.config_path,
            "GITEA_MCP_PROFILE": profile,
            "GITEA_TOKEN_AUTHOR": "author-pass",
            "GITEA_TOKEN_REVIEWER": "reviewer-pass",
        }
    def test_reviewer_task_under_author_profile_wrong_role_stop(self):
        with patch.dict(os.environ, self._env("prgs-author")):
            route = mcp_server.gitea_route_task_session(
                task_type="review_pr", remote="prgs"
            )
        self.assertEqual(route["route_result"], role_session_router.ROUTE_WRONG_ROLE)
        self.assertFalse(route["downstream_allowed"])
        self.assertIn("Wrong role/session for reviewer task", route["message"])
    @patch("mcp_server.api_request")
    @patch("mcp_server.get_auth_header", return_value="token author-pass")
    def test_issue_creation_blocked_after_reviewer_wrong_role_stop(
        self, _auth, mock_api
    ):
        with patch.dict(os.environ, self._env("prgs-author")):
            mcp_server.gitea_route_task_session(task_type="review_pr", remote="prgs")
            result = mcp_server.gitea_create_issue(
                title="process issue fallback",
                body="should not post",
                remote="prgs",
            )
        self.assertFalse(result.get("success", True))
        self.assertFalse(result.get("performed", True))
        issue_posts = [
            c for c in mock_api.call_args_list
            if c.args[0] == "POST" and str(c.args[1]).endswith("/issues")
        ]
        self.assertEqual(issue_posts, [])
    @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")
    def test_author_task_allowed_after_explicit_author_route(
        self, _auth, _api, _get_all
    ):
        with patch.dict(os.environ, self._env("prgs-author")):
            route = mcp_server.gitea_route_task_session(
                task_type="create_issue", remote="prgs"
            )
            self.assertEqual(
                route["route_result"], role_session_router.ROUTE_ALLOWED
            )
            result = mcp_server.gitea_create_issue(
                title="legit author task",
                remote="prgs",
            )
        self.assertEqual(result.get("number"), 999)
    @patch("mcp_server.api_request", return_value={"login": "sysadmin"})
    @patch("mcp_server.get_auth_header", return_value="token reviewer-pass")
    def test_reviewer_task_under_reviewer_profile_allowed(self, _auth, _api):
        with patch.dict(os.environ, self._env("prgs-reviewer")):
            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"])
    @patch("mcp_server.api_request", return_value={"login": "sysadmin"})
    @patch("mcp_server.get_auth_header", return_value="token reviewer-pass")
    def test_author_task_under_reviewer_profile_routes_to_author(self, _auth, _api):
        with patch.dict(os.environ, self._env("prgs-reviewer")):
            route = mcp_server.gitea_route_task_session(
                task_type="create_issue", remote="prgs"
            )
        self.assertEqual(
            route["route_result"], role_session_router.ROUTE_TO_AUTHOR
        )
        self.assertFalse(route["downstream_allowed"])
    def test_activate_profile_blocked_in_static_mode(self):
        with patch.dict(os.environ, self._env("prgs-author")):
            result = mcp_server.gitea_activate_profile(
                profile_name="prgs-reviewer", remote="prgs"
            )
        self.assertFalse(result["success"])
        self.assertIn("switching is disabled", result["message"])
    def test_handoff_requires_route_fields(self):
        incomplete = assess_role_route_handoff("Task done.")
        self.assertFalse(incomplete["complete"])
        self.assertIn("Task type", incomplete["missing_fields"])
        complete = assess_role_route_handoff(
            "\n".join([
                "Task type: review_pr",
                "Required role: reviewer",
                "Active role: author",
                "Route result: wrong_role_stop",
            ]),
            route_result={"route_result": "wrong_role_stop"},
        )
        self.assertTrue(complete["complete"])
 if __name__ == "__main__":
    unittest.main()