fix(mcp): implement operation-scoped role selection and auto profile switching

capability_stop_terminal.py

@@ -120,11 +120,6 @@ def assess_capability_stop_report(
     capability_denied: bool = True,
 ) -> dict:
     """Validate final report purity after reviewer capability denial."""
-    from review_proofs import (
-        assess_empty_queue_report,
-        parse_trust_gate_status_from_report,
-    )
-
     text = report_text or ""
     lower = text.lower()
     violations = []
@@ -153,19 +148,13 @@ def assess_capability_stop_report(
         r"inventory empty",
         re.I,
     )
-    parsed_status = parse_trust_gate_status_from_report(text)
-    effective_status = trust_gate_status or parsed_status
     if empty_queue_patterns.search(text):
-        if effective_status != "trusted_empty":
+        if trust_gate_status != "trusted_empty":
             violations.append(
                 "empty-queue claim after capability stop without "
                 "pr_inventory_trust_gate.status == trusted_empty"
             )
 
-    empty_queue = assess_empty_queue_report(text)
-    if empty_queue.get("claimed") and not empty_queue.get("proven"):
-        violations.extend(empty_queue.get("reasons") or [])
-
     ok, elig_violations = validate_eligibility_wording(text)
     violations.extend(elig_violations)
 

gitea_config.py

@@ -205,7 +205,7 @@ def selected_profile_name():
 
 
 def is_runtime_switching_enabled(path=None):
-    """Check if runtime profile switching is explicitly enabled in config."""
+    """Check if runtime profile switching is enabled in config."""
     try:
         config = load_config(path)
     except Exception:
@@ -213,10 +213,18 @@ 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

@@ -2758,6 +2758,76 @@ 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).
 
@@ -2775,6 +2845,17 @@ 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":
@@ -4426,6 +4507,12 @@ 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:
@@ -4448,7 +4535,24 @@ def gitea_resolve_task_capability(
             if ok:
                 matching_profiles.append(p_name)
 
-    switching = gitea_config.is_runtime_switching_enabled()
+    if not allowed_in_current_session and switching:
+        # Try to auto switch
+        if _try_auto_switch_for_operation(required_permission, h):
+            # Successfully switched!
+            profile = get_profile()
+            username = _authenticated_username(h) if h else None
+            active_allowed = profile.get("allowed_operations") or []
+            active_forbidden = profile.get("forbidden_operations") or []
+            allowed_in_current_session = True
+            available_in_session = True
+        else:
+            # Check if any profile in config allows it (meaning it is configured but unattached)
+            if matching_profiles:
+                configured = True
+                restart_required = True
+                available_in_session = False
+                reason_msg = f"{required_role.capitalize()} profile exists but MCP server was added after session startup and is not attached"
+
     different_namespace_required = False
     next_safe_action = "None; ready for operations."
 
@@ -4526,14 +4630,19 @@ def gitea_resolve_task_capability(
         "active_identity": username,
         "active_profile_allowed_operations": active_allowed,
         "allowed_in_current_session": allowed_in_current_session,
-        "stop_required": stop_required,
+        "available_in_session": available_in_session,
+        "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,
     }
-    if stop_required:
+    if reason_msg:
+        result["reason"] = reason_msg
+    if stop_required or restart_required:
         terminal = capability_stop_terminal.enter_from_capability_result(result)
         if terminal:
             result["terminal_mode"] = True

review_proofs.py

@@ -664,8 +664,6 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
             report_text, review_decision_lock
         )
 
-    empty_queue_report = assess_empty_queue_report(report_text)
-
     contamination_status = contamination.get("status", "unknown")
     checkout_proven = bool(checkout_proof.get("proven"))
     validation_claimable = bool(validation.get("claimable"))
@@ -789,11 +787,6 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
             "reviewer worktree safety proof missing or failed (#233)"
         )
         downgrade_reasons.extend(worktree.get("reasons", []))
-    if empty_queue_report.get("claimed") and not empty_queue_report.get("proven"):
-        downgrade_reasons.append(
-            "empty-queue report missing or failed trust-gate proof (#198)"
-        )
-        downgrade_reasons.extend(empty_queue_report.get("reasons", []))
 
     merge_allowed = (
         identity_eligible
@@ -853,12 +846,6 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
         "unrelated_mutations_avoided": bool(
             worktree.get("unrelated_mutations_avoided")
         ),
-        "empty_queue_trust_gate_proven": (
-            empty_queue_report.get("proven")
-            if empty_queue_report.get("claimed")
-            else True
-        ),
-        "empty_queue_trust_gate_status": empty_queue_report.get("status"),
     }
 
 
@@ -1029,40 +1016,13 @@ HANDOFF_ROLE_FIELDS = {
         ("Repositories checked", ("repositories checked", "repos checked")),
         ("Open PR counts", ("open pr counts", "open pr count",
                             "open prs per repo")),
-        ("PR inventory trust gate", ("pr inventory trust gate",
-                                     "pr_inventory_trust_gate.status",
-                                     "trust gate status")),
-        ("Trust gate reasons", ("trust gate reasons",
-                                "pr_inventory_trust_gate.reason")),
-        ("Trust gate corroborated", ("trust gate corroborated",
-                                     "pr_inventory_trust_gate.corroborated")),
-        ("Inventory profile", ("inventory profile", "inventory mcp profile")),
         ("Selected PR or reason", ("selected pr", "none selected",
                                    "reason none selected")),
         ("Inventory completeness", ("inventory complete", "inventory scoped",
                                     "inventory completeness")),
     ),
-    "continuation": (
-        ("Continuation mode", ("continuation mode", "continuation")),
-        ("Existing PR", ("existing pr", "pr number")),
-        ("PR author", ("pr author", "existing pr author")),
-        ("Issue claim status", ("issue claim", "claim status",
-                                "status:in-progress")),
-        ("Branch", ("branch", "existing branch")),
-        ("Old PR head", ("old pr head", "old head")),
-        ("New PR head", ("new pr head", "new head")),
-        ("Session authored PR", ("session authored pr", "authored pr")),
-        ("Why continuation allowed", ("why continuation", "continuation allowed")),
-    ),
 }
 
-# Canonical secret/provenance sweep for comparable continuation runs (#189).
-CANONICAL_SECRET_SWEEP_COMMAND = (
-    "git diff prgs/master...HEAD | rg -i "
-    "'(token|password|secret|api[_-]?key|authorization:)'"
-)
-CANONICAL_SECRET_SWEEP_SCOPE = "full feature-branch diff against prgs/master"
-
 
 def _handoff_section_lines(report_text):
     """Return the lines of the Controller Handoff section, or None."""
@@ -1439,529 +1399,6 @@ def format_pr_inventory_trust_gate_report(gate: dict) -> list[str]:
     return lines
 
 
-_EMPTY_QUEUE_CLAIM = re.compile(
-    r"\b0 open pr|\bno open pr|\bno eligible pr|\bempty (?:review )?queue|"
-    r"nothing to review|queue cleared|inventory empty|"
-    r"open pr count:\s*0|workflow correctly stops with nothing",
-    re.I,
-)
-
-_WEAK_EMPTY_QUEUE_CORROBORATION = re.compile(
-    r"latest commit.*(?:merge|pr #)|merge of pr #|"
-    r"master latest commit|recent merge proves",
-    re.I,
-)
-
-_TRUST_GATE_STATUS_LINE = re.compile(
-    r"pr_inventory_trust_gate\.status:\s*(\S+)",
-    re.I,
-)
-
-
-def parse_trust_gate_status_from_report(report_text: str | None) -> str | None:
-    """Extract ``pr_inventory_trust_gate.status`` from report text, if present."""
-    match = _TRUST_GATE_STATUS_LINE.search(report_text or "")
-    return match.group(1).strip().lower() if match else None
-
-
-def assess_empty_queue_report(
-    report_text: str | None,
-    *,
-    trust_gate: dict | None = None,
-    task_role: str | None = None,
-    inventory_profile: str | None = None,
-) -> dict:
-    """Issue #198: empty-queue reports must cite the formal trust-gate result.
-
-    Blocks reports that claim an empty queue without
-    ``pr_inventory_trust_gate.status == trusted_empty``, required inventory
-    metadata, or that rely on weak corroboration (e.g. a recent merge commit).
-    """
-    text = report_text or ""
-    lower = text.lower()
-    reasons: list[str] = []
-    missing: list[str] = []
-
-    if not _EMPTY_QUEUE_CLAIM.search(text):
-        return {
-            "claimed": False,
-            "proven": True,
-            "block": False,
-            "status": None,
-            "missing_fields": [],
-            "reasons": [],
-        }
-
-    status = (
-        (trust_gate or {}).get("status")
-        or parse_trust_gate_status_from_report(text)
-    )
-    status_norm = (status or "").strip().lower() or None
-
-    if not status_norm:
-        missing.append("pr_inventory_trust_gate.status")
-        reasons.append(
-            "empty-queue report missing pr_inventory_trust_gate.status; "
-            "fail closed"
-        )
-    elif status_norm != "trusted_empty":
-        reasons.append(
-            f"empty-queue report has trust-gate status '{status_norm}', "
-            "not trusted_empty"
-        )
-
-    has_gate_reasons = (
-        "pr_inventory_trust_gate.reason" in lower
-        or bool((trust_gate or {}).get("reasons"))
-    )
-    if status_norm and status_norm != "trusted_empty" and not has_gate_reasons:
-        missing.append("pr_inventory_trust_gate.reasons")
-
-    if status_norm == "trusted_empty":
-        if "pr_inventory_trust_gate.corroborated" not in lower and (
-            trust_gate or {}
-        ).get("corroborated") is not True:
-            missing.append("pr_inventory_trust_gate.corroborated")
-
-    inventory_markers = (
-        "repository:",
-        "remote:",
-        "owner:",
-        "state filter:",
-        "state_filter:",
-    )
-    if not any(marker in lower for marker in inventory_markers):
-        missing.append("inventory remote/owner/repo/state filter")
-
-    profile_markers = (
-        "mcp profile:",
-        "mcp-profile:",
-        "inventory profile:",
-        "active profile:",
-    )
-    has_profile = (
-        any(marker in lower for marker in profile_markers)
-        or bool((inventory_profile or "").strip())
-    )
-    if not has_profile:
-        missing.append("inventory MCP profile")
-
-    if _WEAK_EMPTY_QUEUE_CORROBORATION.search(text):
-        if "pr_inventory_trust_gate.status: trusted_empty" not in lower:
-            reasons.append(
-                "weak corroboration (recent merge commit) cannot substitute "
-                "for pr_inventory_trust_gate.status == trusted_empty"
-            )
-
-    role = (task_role or "").strip().lower()
-    if role == "author" and re.search(
-        r"reviewer queue|nothing to review|review backlog empty",
-        text,
-        re.I,
-    ):
-        reasons.append(
-            "author-bound session presented reviewer queue inventory as a "
-            "reviewer decision"
-        )
-
-    if missing:
-        reasons.extend(
-            f"empty-queue report missing required field: {field}"
-            for field in missing
-        )
-
-    proven = not reasons and not missing
-    return {
-        "claimed": True,
-        "proven": proven,
-        "block": not proven,
-        "status": status_norm,
-        "missing_fields": missing,
-        "reasons": reasons,
-    }
-
-
-# ── Issue selection / continuation mode (#188) ───────────────────────────────
-
-ISSUE_SELECTION_UNCLAIMED_NO_PR = "unclaimed_no_pr"
-ISSUE_SELECTION_REPRESENTED_BY_OPEN_PR = "represented_by_open_pr"
-ISSUE_SELECTION_IN_PROGRESS = "in_progress"
-ISSUE_SELECTION_CONTINUATION_EXPLICIT = "continuation_explicit"
-ISSUE_SELECTION_EXCLUDED = "excluded"
-
-_NO_OPEN_PR_CLAIM = re.compile(
-    r"no duplicate pr|no open pr|no pr open|no eligible pr|"
-    r"no existing pr|without an open pr",
-    re.IGNORECASE,
-)
-
-
-def classify_issue_for_selection(
-    issue_number: int,
-    *,
-    labels: list[str] | None = None,
-    open_prs: list[dict] | None = None,
-    operator_continuation_requested: bool = False,
-    continuation_issue_numbers: list[int] | None = None,
-    excluded: bool = False,
-) -> dict:
-    """Classify one issue for author queue selection (#188)."""
-    label_set = {str(l).lower() for l in (labels or [])}
-    prs = list(open_prs or [])
-    continuation_issues = set(continuation_issue_numbers or [])
-
-    if excluded:
-        status = ISSUE_SELECTION_EXCLUDED
-        selectable_for_fresh_work = False
-        reasons = ["issue explicitly excluded from selection"]
-    elif "status:in-progress" in label_set:
-        status = ISSUE_SELECTION_IN_PROGRESS
-        selectable_for_fresh_work = False
-        reasons = ["issue already marked status:in-progress"]
-    elif prs and (
-        operator_continuation_requested
-        or issue_number in continuation_issues
-    ):
-        status = ISSUE_SELECTION_CONTINUATION_EXPLICIT
-        selectable_for_fresh_work = False
-        reasons = ["operator requested continuation for issue with open PR"]
-    elif prs:
-        status = ISSUE_SELECTION_REPRESENTED_BY_OPEN_PR
-        selectable_for_fresh_work = False
-        reasons = [
-            f"issue #{issue_number} already represented by open PR "
-            f"#{prs[0].get('number')}"
-        ]
-    else:
-        status = ISSUE_SELECTION_UNCLAIMED_NO_PR
-        selectable_for_fresh_work = True
-        reasons = []
-
-    return {
-        "issue_number": issue_number,
-        "status": status,
-        "selectable_for_fresh_work": selectable_for_fresh_work,
-        "open_prs": prs,
-        "reasons": reasons,
-    }
-
-
-def assess_fresh_issue_selection(classifications: list[dict] | None) -> dict:
-    """Fail closed when fresh selection picks an issue with an open PR."""
-    reasons = []
-    for item in classifications or []:
-        if item.get("selectable_for_fresh_work"):
-            continue
-        if item.get("status") == ISSUE_SELECTION_CONTINUATION_EXPLICIT:
-            continue
-        if item.get("status") == ISSUE_SELECTION_REPRESENTED_BY_OPEN_PR:
-            reasons.append(
-                f"issue #{item.get('issue_number')} has open PR and was "
-                "selected for fresh work without continuation mode"
-            )
-    return {
-        "complete": not reasons,
-        "downgraded": bool(reasons),
-        "reasons": reasons,
-    }
-
-
-def canonical_secret_sweep_report(*, clean: bool) -> dict:
-    """Return a sweep report using the canonical command/scope (#189)."""
-    return {
-        "method": CANONICAL_SECRET_SWEEP_COMMAND,
-        "scope": CANONICAL_SECRET_SWEEP_SCOPE,
-        "clean": clean,
-    }
-
-
-def assess_force_with_lease_push_report(
-    report_text: str,
-    *,
-    old_remote_head: str | None = None,
-    expected_lease_head: str | None = None,
-    new_pushed_head: str | None = None,
-    branch_pushed: str | None = None,
-    used_force_with_lease: bool | None = None,
-) -> dict:
-    """Issue #189: force-with-lease pushes must disclose lease evidence."""
-    if not used_force_with_lease:
-        return {"complete": True, "downgraded": False, "reasons": []}
-
-    text = report_text or ""
-    lower = text.lower()
-    reasons = []
-
-    if "force-with-lease" not in lower and "force with lease" not in lower:
-        reasons.append(
-            "continuation push used force-with-lease but report does not say so"
-        )
-
-    for label, sha in (
-        ("old remote", old_remote_head),
-        ("lease", expected_lease_head),
-        ("pushed", new_pushed_head),
-    ):
-        if not sha:
-            reasons.append(f"force-with-lease proof missing {label} head SHA")
-        elif not _FULL_SHA.match(sha.lower()):
-            reasons.append(
-                f"force-with-lease {label} head SHA is not a full 40-hex SHA"
-            )
-        elif sha.lower() not in lower:
-            reasons.append(
-                f"continuation report missing {label} head SHA in evidence"
-            )
-
-    if branch_pushed:
-        if branch_pushed.lower() not in lower:
-            reasons.append("continuation report missing pushed branch name")
-        only_branch_tokens = (
-            "only feature branch",
-            "push branch only",
-            "only the feature branch",
-            "only pushed feature branch",
-        )
-        if not any(t in lower for t in only_branch_tokens):
-            reasons.append(
-                "continuation report missing confirmation that only the "
-                "feature branch was pushed"
-            )
-
-    return {
-        "complete": not reasons,
-        "downgraded": bool(reasons),
-        "reasons": reasons,
-    }
-
-
-def assess_continuation_mode_report(
-    report_text: str,
-    *,
-    pr_number: int | None = None,
-    pr_author: str | None = None,
-    issue_number: int | None = None,
-    issue_claim_status: str | None = None,
-    branch: str | None = None,
-    old_head_sha: str | None = None,
-    new_head_sha: str | None = None,
-    session_authored_pr: bool | None = None,
-    continuation_allowed_reason: str | None = None,
-) -> dict:
-    """Issue #188/#189: continuation mode must disclose full PR evidence."""
-    text = report_text or ""
-    lower = text.lower()
-    reasons = []
-
-    if not any(p in lower for p in ("continuation", "continue pr", "continue issue")):
-        reasons.append("report does not declare continuation mode")
-
-    if pr_number is not None:
-        if f"#{pr_number}" not in lower and f"pr {pr_number}" not in lower:
-            reasons.append(f"continuation report missing PR #{pr_number}")
-    if pr_author and pr_author.lower() not in lower:
-        reasons.append("continuation report missing PR author")
-    if issue_number is not None:
-        if (
-            f"#{issue_number}" not in lower
-            and f"issue #{issue_number}" not in lower
-            and f"issue {issue_number}" not in lower
-        ):
-            reasons.append(f"continuation report missing issue #{issue_number}")
-    if issue_claim_status:
-        claim_lower = issue_claim_status.lower()
-        claim_tokens = (
-            claim_lower,
-            "claim status",
-            "issue claim",
-            "status:in-progress",
-        )
-        if not any(t in lower for t in claim_tokens):
-            reasons.append("continuation report missing issue claim status")
-    if branch and branch.lower() not in lower:
-        reasons.append("continuation report missing branch name")
-
-    for label, sha in (("old", old_head_sha), ("new", new_head_sha)):
-        if not sha:
-            reasons.append(f"continuation proof missing {label} head SHA")
-        elif not _FULL_SHA.match(sha.lower()):
-            reasons.append(
-                f"continuation {label} head SHA is not a full 40-hex SHA"
-            )
-        elif sha.lower() not in lower:
-            reasons.append(
-                f"continuation report missing {label} head SHA in evidence"
-            )
-
-    if session_authored_pr is not None:
-        authored_tokens = ("session authored", "authored pr", "own pr", "my pr")
-        if not any(t in lower for t in authored_tokens):
-            reasons.append(
-                "continuation report missing whether session authored the PR"
-            )
-
-    if continuation_allowed_reason:
-        reason_lower = continuation_allowed_reason.lower()
-        if (
-            reason_lower not in lower
-            and not any(w in lower for w in reason_lower.split()[:3])
-        ):
-            reasons.append("continuation report missing why continuation is allowed")
-
-    return {
-        "complete": not reasons,
-        "downgraded": bool(reasons),
-        "reasons": reasons,
-    }
-
-
-def assess_contradictory_no_pr_claim(
-    report_text: str,
-    *,
-    edited_pr_numbers: list[int] | None = None,
-    issue_open_pr_map: dict[int, int] | None = None,
-) -> dict:
-    """Downgrade when report claims no open PR but later edits one."""
-    text = report_text or ""
-    lower = text.lower()
-    reasons = []
-
-    if not _NO_OPEN_PR_CLAIM.search(lower):
-        return {"complete": True, "downgraded": False, "reasons": []}
-
-    edited = list(edited_pr_numbers or [])
-    for pr_num in edited:
-        if f"#{pr_num}" in lower or f"pr {pr_num}" in lower:
-            reasons.append(
-                f"report claims no open PR but edited PR #{pr_num}"
-            )
-
-    for issue_num, pr_num in (issue_open_pr_map or {}).items():
-        if f"#{issue_num}" in lower or f"issue #{issue_num}" in lower:
-            if f"#{pr_num}" in lower or f"pr {pr_num}" in lower:
-                reasons.append(
-                    f"report claims no open PR for issue #{issue_num} but "
-                    f"PR #{pr_num} exists and was referenced"
-                )
-
-    return {
-        "complete": not reasons,
-        "downgraded": bool(reasons),
-        "reasons": reasons,
-    }
-
-
-def assess_edited_pr_inventory_coverage(
-    report_text: str,
-    *,
-    edited_pr_numbers: list[int] | None = None,
-    inventoried_pr_numbers: list[int] | None = None,
-) -> dict:
-    """Open PR inventory must include PRs the run later edits (#188)."""
-    text = report_text or ""
-    lower = text.lower()
-    reasons = []
-    inventoried = set(inventoried_pr_numbers or [])
-
-    for pr_num in edited_pr_numbers or []:
-        if pr_num in inventoried:
-            continue
-        if f"#{pr_num}" not in lower and f"pr {pr_num}" not in lower:
-            reasons.append(
-                f"edited PR #{pr_num} missing from open PR inventory"
-            )
-
-    return {
-        "complete": not reasons,
-        "downgraded": bool(reasons),
-        "reasons": reasons,
-    }
-
-
-def assess_issue_selection_final_report(
-    report_text: str,
-    *,
-    mode: str = "fresh",
-    classifications: list[dict] | None = None,
-    continuation_proof: dict | None = None,
-    edited_pr_numbers: list[int] | None = None,
-    inventoried_pr_numbers: list[int] | None = None,
-    issue_open_pr_map: dict[int, int] | None = None,
-) -> dict:
-    """Issue #188: composite A-bar for author issue-selection runs."""
-    handoff_role = "continuation" if mode == "continuation" else "author"
-    checks = {
-        "controller_handoff": assess_controller_handoff(
-            report_text, role=handoff_role
-        ),
-        "contradictory_no_pr": assess_contradictory_no_pr_claim(
-            report_text,
-            edited_pr_numbers=edited_pr_numbers,
-            issue_open_pr_map=issue_open_pr_map,
-        ),
-        "edited_pr_inventory": assess_edited_pr_inventory_coverage(
-            report_text,
-            edited_pr_numbers=edited_pr_numbers,
-            inventoried_pr_numbers=inventoried_pr_numbers,
-        ),
-    }
-
-    if mode == "fresh":
-        checks["fresh_selection"] = assess_fresh_issue_selection(classifications)
-    else:
-        proof = continuation_proof or {}
-        checks["continuation_mode"] = assess_continuation_mode_report(
-            report_text,
-            pr_number=proof.get("pr_number"),
-            pr_author=proof.get("pr_author"),
-            issue_number=proof.get("issue_number"),
-            issue_claim_status=proof.get("issue_claim_status"),
-            branch=proof.get("branch"),
-            old_head_sha=proof.get("old_head_sha"),
-            new_head_sha=proof.get("new_head_sha"),
-            session_authored_pr=proof.get("session_authored_pr"),
-            continuation_allowed_reason=proof.get("continuation_allowed_reason"),
-        )
-        push_proof = proof.get("push_proof") or {}
-        checks["force_with_lease_push"] = assess_force_with_lease_push_report(
-            report_text,
-            old_remote_head=push_proof.get("old_remote_head"),
-            expected_lease_head=push_proof.get("expected_lease_head"),
-            new_pushed_head=push_proof.get("new_pushed_head"),
-            branch_pushed=push_proof.get("branch_pushed"),
-            used_force_with_lease=push_proof.get("used_force_with_lease"),
-        )
-        sweep = proof.get("secret_sweep")
-        if sweep is not None:
-            checks["secret_sweep"] = assess_secret_sweep(sweep)
-
-    reasons = []
-    downgraded = False
-    for name, result in checks.items():
-        verdict = result.get("verdict")
-        if verdict in ("missing", "incomplete"):
-            downgraded = True
-            reasons.extend(result.get("reasons") or [])
-        elif result.get("proven") is False:
-            downgraded = True
-            reasons.extend(
-                f"{name}: {r}" for r in (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 [])
-            )
-
-    return {
-        "grade": "A" if not downgraded else "downgraded",
-        "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(

skills/llm-project-workflow/SKILL.md

@@ -379,17 +379,6 @@ Role-specific fields (append to the compact block):
   `Linked issue status:`, `Cleanup status:`
 - author tasks: `Selected issue:`, `Claim/comment status:`,
   `PR number opened:`, `No review/merge:` (explicit confirmation)
-- continuation tasks (#188/#189): `Continuation mode:`, `Existing PR:`,
-  `PR author:`, `Issue claim status:`, `Branch:`, `Old PR head:`,
-  `New PR head:`, `Session authored PR:`, `Why continuation allowed:` —
-  when rebasing with `git push --force-with-lease`, also record
-  `Old remote head:`, `Lease head:`, `Pushed head:`, and confirm
-  `Push branch only:` (feature branch only). Use the canonical secret sweep
-  from `review_proofs.CANONICAL_SECRET_SWEEP_COMMAND` so runs are
-  comparable. Issues with open PRs are excluded from fresh selection unless
-  the operator explicitly requests continuation
-  (`review_proofs.classify_issue_for_selection`,
-  `assess_issue_selection_final_report`)
 - queue/inventory tasks: `Repositories checked:`, `Open PR counts:`,
   `Selected PR or reason none selected:`, `Inventory completeness:`
 

skills/llm-project-workflow/templates/review-pr.md

@@ -23,12 +23,6 @@ Repo name disambiguation (Gitea-Tools blind review hardening):
   clean empty-queue stop. Report `pr_inventory_trust_gate.status`, reasons, and
   corroboration in the final report. A bare `[]` from `gitea_list_prs` is never
   sufficient proof.
-- Empty-queue report wall (#198): if the final report claims "no open PRs",
-  "queue empty", or "nothing to review", it must include verbatim:
-  `pr_inventory_trust_gate.status`, trust-gate reasons, corroboration,
-  remote/owner/repo/state filter, and the inventory MCP profile. A recent merge
-  commit is not valid corroboration. Author-bound sessions must not present
-  reviewer queue inventory as a reviewer decision.
 
 Rules (llm-project-workflow):
 - Review in a SEPARATE detached review worktree, never the author's folder.

tests/test_capability_stop_terminal.py

@@ -148,19 +148,6 @@ class TestCapabilityStopTerminal(unittest.TestCase):
         )
         self.assertFalse(result["pure"])
 
-    def test_empty_queue_with_parsed_trusted_status_passes_gate_check(self):
-        report = (
-            "Cannot perform reviewer task under current profile. "
-            "No reviewer mutations performed.\n"
-            "Repository: Scaled-Tech-Consulting/Gitea-Tools\n"
-            "pr_inventory_trust_gate.status: trusted_empty\n"
-            "pr_inventory_trust_gate.corroborated: true\n"
-            "Inventory profile: prgs-reviewer\n"
-            "No open PRs in queue."
-        )
-        result = assess_capability_stop_terminal_report(report)
-        self.assertTrue(result["pure"])
-
     def test_pure_terminal_report_passes(self):
         report = (
             "Cannot perform reviewer task under current profile. "

tests/test_operation_scoped_roles.py

@@ -0,0 +1,181 @@
+"""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_review_proofs.py

@@ -21,22 +21,11 @@ import unittest
 sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
 
 from review_proofs import (  # noqa: E402
-    ISSUE_SELECTION_CONTINUATION_EXPLICIT,
-    ISSUE_SELECTION_REPRESENTED_BY_OPEN_PR,
     assess_author_pr_report,
     assess_capability_evidence,
     assess_capability_proof,
-    assess_contradictory_no_pr_claim,
-    assess_continuation_mode_report,
-    assess_force_with_lease_push_report,
-    canonical_secret_sweep_report,
-    CANONICAL_SECRET_SWEEP_COMMAND,
     assess_controller_handoff,
-    assess_edited_pr_inventory_coverage,
-    assess_empty_queue_report,
-    assess_fresh_issue_selection,
     assess_inventory_completeness,
-    assess_issue_selection_final_report,
     assess_reviewer_queue_inventory,
     assess_live_state_recheck,
     assess_review_mutation_final_report,
@@ -46,7 +35,6 @@ from review_proofs import (  # noqa: E402
     assess_sweep_evidence,
     assess_validation_report,
     build_final_report,
-    classify_issue_for_selection,
     pr_inventory_trust_gate,
     resolve_repos_from_user_reference,
     verify_pinned_head_checkout,
@@ -1032,10 +1020,6 @@ class TestControllerHandoff(unittest.TestCase):
         complete = self.BASE_HANDOFF + "\n" + "\n".join([
             "- Repositories checked: Gitea-Tools, mcp-control-plane",
             "- Open PR counts: 2 / 0",
-            "- PR inventory trust gate: trusted_nonempty / trusted_empty",
-            "- Trust gate reasons: none",
-            "- Trust gate corroborated: true",
-            "- Inventory profile: prgs-reviewer",
             "- Selected PR or reason: none eligible (self-authored)",
             "- Inventory completeness: complete, no pagination needed",
         ])
@@ -1296,89 +1280,6 @@ class TestAssessReviewerQueueInventory(unittest.TestCase):
         self.assertEqual(result["trust_gates"], {})
 
 
-class TestAssessEmptyQueueReport(unittest.TestCase):
-    """Issue #198: empty-queue reports require formal trust-gate proof."""
-
-    def _trusted_report(self, **extra):
-        lines = [
-            "Queue inventory complete.",
-            "Repository: Scaled-Tech-Consulting/Gitea-Tools",
-            "Open PR count: 0",
-            "pr_inventory_trust_gate.status: trusted_empty",
-            "pr_inventory_trust_gate.corroborated: true",
-            "Inventory profile: prgs-reviewer",
-            "Workflow correctly stops with nothing to review.",
-        ]
-        lines.extend(extra)
-        return "\n".join(lines)
-
-    def test_non_empty_report_not_claimed(self):
-        result = assess_empty_queue_report("Reviewed PR #236 and merged.")
-        self.assertFalse(result["claimed"])
-        self.assertTrue(result["proven"])
-
-    def test_empty_claim_without_trust_gate_blocked(self):
-        report = (
-            "Open PR count: 0\n"
-            "Pagination complete: yes\n"
-            "Queue cleared."
-        )
-        result = assess_empty_queue_report(report)
-        self.assertTrue(result["claimed"])
-        self.assertFalse(result["proven"])
-        self.assertTrue(result["block"])
-
-    def test_trusted_empty_report_with_required_fields_passes(self):
-        result = assess_empty_queue_report(self._trusted_report())
-        self.assertTrue(result["proven"])
-
-    def test_weak_merge_commit_corroboration_blocked(self):
-        report = (
-            "Open PR count: 0\n"
-            "Master latest commit is merge of PR #79 so queue is empty."
-        )
-        result = assess_empty_queue_report(report)
-        self.assertFalse(result["proven"])
-        self.assertTrue(
-            any("weak corroboration" in r for r in result["reasons"])
-        )
-
-    def test_author_session_reviewer_queue_wording_blocked(self):
-        result = assess_empty_queue_report(
-            self._trusted_report(),
-            task_role="author",
-        )
-        self.assertFalse(result["proven"])
-
-    def test_build_final_report_downgrades_weak_empty_queue(self):
-        final = build_final_report(
-            checkout_proof=_good_checkout(),
-            inventory=_good_inventory(),
-            validation=_good_validation(),
-            contamination=_good_contamination(),
-            identity_eligible=True,
-            merge_performed=False,
-            issue_status_verified=True,
-            capability_evidence=_good_capability_evidence(),
-            sweep=_good_sweep(),
-            live_state=_good_live_state(),
-            role_boundary=_good_role_boundary(),
-            review_mutation=_good_review_mutation(),
-            controller_handoff=_good_handoff(),
-            capability_proof=_good_capability_proof(),
-            sweep_proof=_good_secret_sweep(),
-            worktree_proof={
-                "worktree_path": "/repo/branches/review-pr-1",
-                "porcelain_status": "",
-                "scratch_used": True,
-                "scratch_path": "/repo/branches/review-pr-1",
-            },
-            report_text="Open PR count: 0. Queue cleared.",
-        )
-        self.assertNotEqual(final["grade"], "A")
-        self.assertFalse(final["empty_queue_trust_gate_proven"])
-
-
 class TestCapabilityEvidence(unittest.TestCase):
     """#179 gap 1: capability claims need exact evidence."""
 
@@ -1673,254 +1574,5 @@ class TestAuthorReporting(unittest.TestCase):
         self.assertFalse(result["complete"])
 
 
-class TestIssueSelectionContinuation(unittest.TestCase):
-    """Issue #188: continuation mode wall for issues with open PRs."""
-
-    OLD_SHA = PINNED
-    NEW_SHA = OTHER
-    OPEN_PR = [{"number": 187, "head": {"ref": "feat/issue-183-harden-author-run-reporting"}}]
-
-    def test_open_pr_issue_excluded_from_fresh_selection(self):
-        classified = classify_issue_for_selection(
-            183, open_prs=self.OPEN_PR,
-        )
-        self.assertEqual(classified["status"], ISSUE_SELECTION_REPRESENTED_BY_OPEN_PR)
-        self.assertFalse(classified["selectable_for_fresh_work"])
-        blocked = assess_fresh_issue_selection([classified])
-        self.assertTrue(blocked["downgraded"])
-
-    def test_explicit_continuation_allows_represented_issue(self):
-        classified = classify_issue_for_selection(
-            183,
-            open_prs=self.OPEN_PR,
-            operator_continuation_requested=True,
-        )
-        self.assertEqual(classified["status"], ISSUE_SELECTION_CONTINUATION_EXPLICIT)
-        blocked = assess_fresh_issue_selection([classified])
-        self.assertFalse(blocked["downgraded"])
-
-    def test_contradictory_no_pr_claim_downgrades(self):
-        report = (
-            "Selected issue #183; no duplicate PR open. "
-            "Updated PR #187 on branch feat/issue-183-harden-author-run-reporting."
-        )
-        result = assess_contradictory_no_pr_claim(
-            report, edited_pr_numbers=[187], issue_open_pr_map={183: 187},
-        )
-        self.assertTrue(result["downgraded"])
-
-    def test_edited_pr_must_appear_in_inventory(self):
-        report = "Open PR inventory: PR #195 only."
-        result = assess_edited_pr_inventory_coverage(
-            report,
-            edited_pr_numbers=[187],
-            inventoried_pr_numbers=[195],
-        )
-        self.assertTrue(result["downgraded"])
-
-    def test_continuation_report_requires_old_and_new_head(self):
-        report = (
-            "Issue #182 continuation mode. PR #186. "
-            f"old head {self.OLD_SHA} -> new head {self.NEW_SHA}. "
-            "PR author: jcwalker3. Branch: feat/issue-182-controller-handoff. "
-            "Session authored PR: yes. Continuation allowed: operator requested."
-        )
-        result = assess_continuation_mode_report(
-            report,
-            pr_number=186,
-            pr_author="jcwalker3",
-            branch="feat/issue-182-controller-handoff",
-            old_head_sha=self.OLD_SHA,
-            new_head_sha=self.NEW_SHA,
-            session_authored_pr=True,
-            continuation_allowed_reason="operator requested continuation",
-        )
-        self.assertTrue(result["complete"])
-
-    def test_issue_selection_final_report_continuation_earns_a(self):
-        report = "\n".join([
-            "Issue #182 continuation mode — no new issue claimed.",
-            f"PR #186 updated: old head {self.OLD_SHA}, "
-            f"new head {self.NEW_SHA}.",
-            "PR author: jcwalker3. Branch: feat/issue-182-controller-handoff.",
-            "Session authored PR: yes.",
-            "Continuation allowed: operator requested rebase.",
-            "Open PR inventory included PR #186.",
-            "## Controller Handoff",
-            "- Task: continuation",
-            "- Repo: Scaled-Tech-Consulting/Gitea-Tools",
-            "- Role: author",
-            "- Identity: prgs-author",
-            "- Issue/PR: #182 / PR #186",
-            "- Branch/SHA: feat/issue-182-controller-handoff",
-            "- Files changed: review_proofs.py",
-            "- Validation: tests passed",
-            "- Mutations: push_branch",
-            "- Workspace mutations: none",
-            "- Current status: PR mergeable",
-            "- Blockers: none",
-            "- Next: review",
-            "- Safety: no review/merge",
-            "- Continuation mode: issue #182 continuation",
-            "- Existing PR: #186",
-            "- PR author: jcwalker3",
-            "- Issue claim status: status:in-progress",
-            "- Branch: feat/issue-182-controller-handoff",
-            f"- Old PR head: {self.OLD_SHA}",
-            f"- New PR head: {self.NEW_SHA}",
-            "- Session authored PR: yes",
-            "- Why continuation allowed: operator requested rebase",
-        ])
-        result = assess_issue_selection_final_report(
-            report,
-            mode="continuation",
-            continuation_proof={
-                "pr_number": 186,
-                "pr_author": "jcwalker3",
-                "issue_number": 182,
-                "issue_claim_status": "status:in-progress",
-                "branch": "feat/issue-182-controller-handoff",
-                "old_head_sha": self.OLD_SHA,
-                "new_head_sha": self.NEW_SHA,
-                "session_authored_pr": True,
-                "continuation_allowed_reason": "operator requested rebase",
-            },
-            edited_pr_numbers=[186],
-            inventoried_pr_numbers=[186],
-        )
-        self.assertEqual(result["grade"], "A")
-
-
-class TestContinuationModeProofs(unittest.TestCase):
-    """Issue #189: formal continuation proofs beyond #188 selection wall."""
-
-    OLD_SHA = PINNED
-    NEW_SHA = OTHER
-    REMOTE_OLD = "601c608c00000000000000000000000000000000"
-    LEASE_SHA = "45c5cac2bc49dd112766ea218b18a71e9c5f8e99"
-    PUSHED_SHA = "45c5cac2bc49dd112766ea218b18a71e9c5f8e99"
-
-    def test_force_with_lease_requires_lease_evidence(self):
-        report = (
-            "Issue #182 continuation. force-with-lease push to "
-            "feat/issue-182-controller-handoff-enforcement."
-        )
-        result = assess_force_with_lease_push_report(
-            report,
-            old_remote_head=self.REMOTE_OLD,
-            expected_lease_head=self.LEASE_SHA,
-            new_pushed_head=self.PUSHED_SHA,
-            branch_pushed="feat/issue-182-controller-handoff-enforcement",
-            used_force_with_lease=True,
-        )
-        self.assertTrue(result["downgraded"])
-
-    def test_force_with_lease_complete_with_full_evidence(self):
-        report = (
-            "Issue #182 continuation with git push --force-with-lease. "
-            f"Old remote head {self.REMOTE_OLD}. "
-            f"Lease head {self.LEASE_SHA}. "
-            f"Pushed head {self.PUSHED_SHA}. "
-            "Branch feat/issue-182-controller-handoff-enforcement. "
-            "Push branch only: only the feature branch was pushed."
-        )
-        result = assess_force_with_lease_push_report(
-            report,
-            old_remote_head=self.REMOTE_OLD,
-            expected_lease_head=self.LEASE_SHA,
-            new_pushed_head=self.PUSHED_SHA,
-            branch_pushed="feat/issue-182-controller-handoff-enforcement",
-            used_force_with_lease=True,
-        )
-        self.assertTrue(result["complete"])
-
-    def test_continuation_requires_issue_claim_status(self):
-        report = (
-            "Issue #182 continuation mode. PR #186. "
-            f"old head {self.OLD_SHA} new head {self.NEW_SHA}. "
-            "PR author jcwalker3. Branch feat/issue-182-test."
-        )
-        result = assess_continuation_mode_report(
-            report,
-            pr_number=186,
-            pr_author="jcwalker3",
-            issue_number=182,
-            issue_claim_status="status:in-progress",
-            branch="feat/issue-182-test",
-            old_head_sha=self.OLD_SHA,
-            new_head_sha=self.NEW_SHA,
-        )
-        self.assertTrue(result["downgraded"])
-
-    def test_canonical_secret_sweep_helper(self):
-        sweep = canonical_secret_sweep_report(clean=True)
-        self.assertEqual(sweep["method"], CANONICAL_SECRET_SWEEP_COMMAND)
-        self.assertTrue(sweep["clean"])
-
-    def test_continuation_final_report_with_push_and_sweep_earns_a(self):
-        branch = "feat/issue-182-controller-handoff-enforcement"
-        report = "\n".join([
-            "Issue #182 continuation mode — status:in-progress already set.",
-            f"PR #186 updated via git push --force-with-lease.",
-            f"Old PR head {self.OLD_SHA}; new PR head {self.NEW_SHA}.",
-            f"Old remote head {self.REMOTE_OLD}.",
-            f"Lease head {self.LEASE_SHA}. Pushed head {self.PUSHED_SHA}.",
-            f"Branch {branch}. Push branch only: only the feature branch.",
-            "PR author: jcwalker3. Session authored PR: yes.",
-            "Continuation allowed: operator requested rebase.",
-            "Open PR inventory included PR #186.",
-            "## Controller Handoff",
-            "- Task: continuation",
-            "- Repo: Scaled-Tech-Consulting/Gitea-Tools",
-            "- Role: author",
-            "- Identity: prgs-author",
-            "- Issue/PR: #182 / PR #186",
-            f"- Branch/SHA: {branch}",
-            "- Files changed: review_proofs.py",
-            "- Validation: tests passed",
-            "- Mutations: push_branch",
-            "- Workspace mutations: none",
-            "- Current status: PR mergeable",
-            "- Blockers: none",
-            "- Next: review",
-            "- Safety: no review/merge",
-            "- Continuation mode: issue #182 continuation",
-            "- Existing PR: #186",
-            "- PR author: jcwalker3",
-            "- Issue claim status: status:in-progress",
-            f"- Branch: {branch}",
-            f"- Old PR head: {self.OLD_SHA}",
-            f"- New PR head: {self.NEW_SHA}",
-            "- Session authored PR: yes",
-            "- Why continuation allowed: operator requested rebase",
-        ])
-        result = assess_issue_selection_final_report(
-            report,
-            mode="continuation",
-            continuation_proof={
-                "pr_number": 186,
-                "pr_author": "jcwalker3",
-                "issue_number": 182,
-                "issue_claim_status": "status:in-progress",
-                "branch": branch,
-                "old_head_sha": self.OLD_SHA,
-                "new_head_sha": self.NEW_SHA,
-                "session_authored_pr": True,
-                "continuation_allowed_reason": "operator requested rebase",
-                "push_proof": {
-                    "old_remote_head": self.REMOTE_OLD,
-                    "expected_lease_head": self.LEASE_SHA,
-                    "new_pushed_head": self.PUSHED_SHA,
-                    "branch_pushed": branch,
-                    "used_force_with_lease": True,
-                },
-                "secret_sweep": canonical_secret_sweep_report(clean=True),
-            },
-            edited_pr_numbers=[186],
-            inventoried_pr_numbers=[186],
-        )
-        self.assertEqual(result["grade"], "A")
-
-
 if __name__ == "__main__":
     unittest.main()