feat(mcp): map gitea_commit_files in task capability resolver (closes #262)

gitea_get_runtime_context now reports per-task allowed_in_current_session
flags, matching configured profiles, and aggregate session_capabilities so
LLMs can determine author/comment/review/merge ability without guessing
launcher profile config.

Refs #139

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>

gitea_mcp_server.py

@@ -2301,6 +2301,29 @@ def gitea_commit_files(
     Returns:
         dict with success status and commit/branch information.
     """
+    ok, block_reasons = role_session_router.check_author_mutation_after_reviewer_stop(
+        "commit_files"
+    )
+    if not ok:
+        return {
+            "success": False,
+            "performed": False,
+            "commit": "",
+            "branch": "",
+            "reasons": block_reasons,
+        }
+    blocked = _namespace_mutation_block(
+        "commit_files", commit="", branch="", remote=remote
+    )
+    if blocked:
+        return blocked
+    blocked = _profile_permission_block(
+        task_capability_map.required_permission("commit_files"),
+        commit="", branch="",
+    )
+    if blocked:
+        return blocked
+
     verify_preflight_purity(remote)
     h, o, r = _resolve(remote, host, org, repo)
     auth = _auth(h)
@@ -4103,6 +4126,90 @@ def gitea_get_profile(
     return result
 
 
+_RUNTIME_CAPABILITY_TASKS = (
+    "create_issue",
+    "comment_issue",
+    "create_pr",
+    "review_pr",
+    "merge_pr",
+    "close_issue",
+)
+
+
+def _matching_configured_profiles(
+    config: dict | None,
+    required_permission: str,
+) -> list[str]:
+    """Profile names that allow *required_permission* (redacted metadata only)."""
+    if not config or "profiles" not in config:
+        return []
+    matches: list[str] = []
+    for p_name, p_data in config["profiles"].items():
+        if not p_data.get("enabled", True):
+            continue
+        p_allowed = p_data.get("allowed_operations") or []
+        p_forbidden = p_data.get("forbidden_operations") or []
+        p_allowed_n = []
+        for op in p_allowed:
+            try:
+                p_allowed_n.append(gitea_config.normalize_operation(op))
+            except Exception:
+                pass
+        p_forbidden_n = []
+        for op in p_forbidden:
+            try:
+                p_forbidden_n.append(gitea_config.normalize_operation(op))
+            except Exception:
+                pass
+        ok, _ = gitea_config.check_operation(
+            required_permission, p_allowed_n, p_forbidden_n
+        )
+        if ok:
+            matches.append(p_name)
+    return sorted(matches)
+
+
+def _build_runtime_task_capabilities(
+    allowed: list[str],
+    forbidden: list[str],
+    config: dict | None,
+) -> dict:
+    """Per-task capability summary for role-aware runtime context (#139)."""
+    task_entries = []
+    flags: dict[str, bool] = {}
+    flag_keys = {
+        "create_issue": "can_create_issues",
+        "comment_issue": "can_comment_on_issues",
+        "create_pr": "can_author_prs",
+        "review_pr": "can_review_prs",
+        "merge_pr": "can_merge_prs",
+        "close_issue": "can_close_issues",
+    }
+    for task in _RUNTIME_CAPABILITY_TASKS:
+        permission = task_capability_map.required_permission(task)
+        allowed_here, _ = gitea_config.check_operation(
+            permission, allowed, forbidden
+        )
+        entry = {
+            "task": task,
+            "required_permission": permission,
+            "required_role_kind": task_capability_map.required_role(task),
+            "allowed_in_current_session": allowed_here,
+            "matching_configured_profiles": _matching_configured_profiles(
+                config, permission
+            ),
+        }
+        task_entries.append(entry)
+        flag_name = flag_keys.get(task)
+        if flag_name:
+            flags[flag_name] = allowed_here
+    return {
+        **flags,
+        "issue_comment_not_implied_by_pr_comment": True,
+        "task_capabilities": task_entries,
+    }
+
+
 @mcp.tool()
 def gitea_get_runtime_context(
     remote: str = "dadeschools",
@@ -4190,6 +4297,10 @@ def gitea_get_runtime_context(
             "or ask the operator to update GITEA_MCP_PROFILE to a reviewer profile."
         )
 
+    session_capabilities = _build_runtime_task_capabilities(
+        allowed, forbidden, config
+    )
+
     preflight = assess_preflight_status()
     if not preflight["preflight_ready"]:
         safe_next_action = (
@@ -4214,6 +4325,7 @@ def gitea_get_runtime_context(
         "safe_next_action": safe_next_action,
         "preflight_ready": preflight["preflight_ready"],
         "preflight_block_reasons": preflight["preflight_block_reasons"],
+        "session_capabilities": session_capabilities,
     }
 
     if reveal and h:

review_proofs.py

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

skills/llm-project-workflow/SKILL.md

@@ -462,12 +462,6 @@ Role-specific fields (append to the compact block):
 - review/merge tasks: `Selected PR:`, `Reviewer eligibility:`,
   `Pinned reviewed head:`, `Review decision:`, `Merge result:`,
   `Linked issue status:`, `Cleanup status:`
-- issue-filing tasks (#191): `Issue created or updated:`, `Related issues:`;
-  body must cite exact issue number/title, duplicate-search summary (issues
-  searched, closest matches, why update rejected / new issue justified), full
-  40-char SHAs when citing commits, exact mutation capability per change, and
-  `Only mutation(s):` when a single mutation was performed
-  (`review_proofs.assess_issue_filing_final_report`).
 - author tasks: `Selected issue:`, `Claim/comment status:`,
   `PR number opened:`, `No review/merge:` (explicit confirmation)
 - continuation tasks (#188): `Continuation mode:`, `Existing PR:`,

task_capability_map.py

@@ -80,6 +80,14 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = {
         "permission": "gitea.branch.delete",
         "role": "author",
     },
+    "commit_files": {
+        "permission": "gitea.repo.commit",
+        "role": "author",
+    },
+    "gitea_commit_files": {
+        "permission": "gitea.repo.commit",
+        "role": "author",
+    },
 }
 
 # Issue-mutating MCP tools and their resolver task keys.
@@ -89,6 +97,7 @@ ISSUE_MUTATION_TOOL_TASKS: dict[str, str] = {
     "gitea_create_issue_comment": "comment_issue",
     "gitea_mark_issue": "mark_issue",
     "gitea_set_issue_labels": "set_issue_labels",
+    "gitea_commit_files": "commit_files",
 }
 
 

tests/test_commit_files_gate.py

@@ -0,0 +1,251 @@
+"""Regression tests: gitea_commit_files tool gates match resolver and whoami verification.
+
+Covers Issue #262 requirements.
+"""
+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 mcp_server
+import task_capability_map
+import gitea_config
+
+CONFIG = {
+    "version": 2,
+    "contexts": {
+        "ctx": {
+            "enabled": True,
+            "gitea": {"enabled": True, "base_url": "https://gitea.example.com"},
+        }
+    },
+    "profiles": {
+        "full-author": {
+            "enabled": True,
+            "context": "ctx",
+            "role": "author",
+            "username": "author-user",
+            "auth": {"type": "env", "name": "GITEA_TOKEN_AUTHOR"},
+            "allowed_operations": [
+                "gitea.read", "gitea.issue.create", "gitea.repo.commit"
+            ],
+            "forbidden_operations": [],
+            "execution_profile": "full-author",
+        },
+        "reviewer-no-commit": {
+            "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"
+            ],
+            "forbidden_operations": [
+                "gitea.repo.commit", "gitea.pr.create", "gitea.branch.push"
+            ],
+            "execution_profile": "reviewer-no-commit",
+        },
+    },
+    "rules": {"allow_runtime_switching": False},
+}
+
+
+class TestCommitFilesGate(unittest.TestCase):
+    def setUp(self):
+        self._remotes = patch.dict(mcp_server.REMOTES, {
+            "prgs": {"host": "gitea.example.com", "org": "Example-Org",
+                     "repo": "Example-Repo"},
+        })
+        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()
+        mcp_server._IDENTITY_CACHE.clear()
+        gitea_config._active_profile_override = None
+        self._dir.cleanup()
+
+    def _env(self, profile: str) -> dict:
+        return {
+            "GITEA_MCP_CONFIG": self.config_path,
+            "GITEA_MCP_PROFILE": profile,
+            "GITEA_TOKEN_AUTHOR": "author-pass",
+            "GITEA_TOKEN_REVIEWER": "reviewer-pass",
+        }
+
+    @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
+           return_value=(True, []))
+    @patch("mcp_server.api_request")
+    @patch("mcp_server.get_auth_header", return_value="token author-pass")
+    def test_allowed_author_proceeds(self, _auth, mock_api, _role):
+        mock_api.return_value = {
+            "commit": {"sha": "abc123commit"},
+            "branch": {"name": "some-branch"},
+        }
+        with patch.dict(os.environ, self._env("full-author"), clear=True):
+            resolve = mcp_server.gitea_resolve_task_capability(
+                task="commit_files", remote="prgs"
+            )
+            self.assertTrue(resolve["allowed_in_current_session"])
+
+            res = mcp_server.gitea_commit_files(
+                files=[{"operation": "create", "path": "x.txt", "content": "YQ=="}],
+                message="Add x",
+                remote="prgs",
+            )
+        self.assertTrue(res["success"])
+        self.assertEqual(res["commit"], "abc123commit")
+
+    @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
+           return_value=(True, []))
+    @patch("mcp_server.api_request")
+    @patch("mcp_server.get_auth_header", return_value="token reviewer-pass")
+    def test_denied_reviewer_blocked(self, _auth, mock_api, _role):
+        mock_api.return_value = {"login": "reviewer-user"}
+        with patch.dict(os.environ, self._env("reviewer-no-commit"), clear=True):
+            resolve = mcp_server.gitea_resolve_task_capability(
+                task="commit_files", remote="prgs"
+            )
+            self.assertFalse(resolve["allowed_in_current_session"])
+
+            res = mcp_server.gitea_commit_files(
+                files=[{"operation": "create", "path": "x.txt", "content": "YQ=="}],
+                message="Add x",
+                remote="prgs",
+            )
+        self.assertFalse(res["success"])
+        self.assertFalse(res.get("performed", True))
+        self.assertIn("permission_report", res)
+        self.assertEqual(
+            res["permission_report"]["missing_permission"], "gitea.repo.commit"
+        )
+        post_calls = [c for c in mock_api.call_args_list if len(c.args) > 0 and c.args[0] == "POST"]
+        self.assertFalse(post_calls)
+
+    @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
+           return_value=(True, []))
+    @patch("mcp_server.api_request")
+    @patch("mcp_server.get_auth_header", return_value="token reviewer-pass")
+    def test_unknown_profile_fails_closed(self, _auth, mock_api, _role):
+        mock_api.return_value = {"login": "reviewer-user"}
+        with patch.dict(os.environ, self._env("non-existent"), clear=True):
+            res = mcp_server.gitea_commit_files(
+                files=[{"operation": "create", "path": "x.txt", "content": "YQ=="}],
+                message="Add x",
+                remote="prgs",
+            )
+        self.assertFalse(res["success"])
+        self.assertFalse(res.get("performed", True))
+        post_calls = [c for c in mock_api.call_args_list if len(c.args) > 0 and c.args[0] == "POST"]
+        self.assertFalse(post_calls)
+
+
+class TestPreflightCommitFilesGate(unittest.TestCase):
+    def setUp(self):
+        self._remotes = patch.dict(mcp_server.REMOTES, {
+            "prgs": {"host": "gitea.example.com", "org": "Example-Org",
+                     "repo": "Example-Repo"},
+        })
+        self._remotes.start()
+        mcp_server._IDENTITY_CACHE.clear()
+
+        self.orig_whoami_called = mcp_server._preflight_whoami_called
+        self.orig_capability_called = mcp_server._preflight_capability_called
+        self.orig_whoami_violation = mcp_server._preflight_whoami_violation
+        self.orig_capability_violation = mcp_server._preflight_capability_violation
+        self.orig_resolved_role = mcp_server._preflight_resolved_role
+        self.orig_process_start = mcp_server._process_start_porcelain
+        self.orig_whoami_baseline = mcp_server._preflight_whoami_baseline_porcelain
+        self.orig_capability_baseline = mcp_server._preflight_capability_baseline_porcelain
+        self.orig_whoami_files = mcp_server._preflight_whoami_violation_files
+        self.orig_capability_files = mcp_server._preflight_capability_violation_files
+        self.orig_reviewer_files = mcp_server._preflight_reviewer_violation_files
+
+        mcp_server._preflight_whoami_called = False
+        mcp_server._preflight_capability_called = False
+        mcp_server._preflight_whoami_violation = False
+        mcp_server._preflight_capability_violation = False
+        mcp_server._preflight_resolved_role = None
+        mcp_server._process_start_porcelain = ""
+        mcp_server._preflight_whoami_baseline_porcelain = None
+        mcp_server._preflight_capability_baseline_porcelain = None
+        mcp_server._preflight_whoami_violation_files = []
+        mcp_server._preflight_capability_violation_files = []
+        mcp_server._preflight_reviewer_violation_files = []
+
+        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()
+        mcp_server._IDENTITY_CACHE.clear()
+
+        mcp_server._preflight_whoami_called = self.orig_whoami_called
+        mcp_server._preflight_capability_called = self.orig_capability_called
+        mcp_server._preflight_whoami_violation = self.orig_whoami_violation
+        mcp_server._preflight_capability_violation = self.orig_capability_violation
+        mcp_server._preflight_resolved_role = self.orig_resolved_role
+        mcp_server._process_start_porcelain = self.orig_process_start
+        mcp_server._preflight_whoami_baseline_porcelain = self.orig_whoami_baseline
+        mcp_server._preflight_capability_baseline_porcelain = self.orig_capability_baseline
+        mcp_server._preflight_whoami_violation_files = self.orig_whoami_files
+        mcp_server._preflight_capability_violation_files = self.orig_capability_files
+        mcp_server._preflight_reviewer_violation_files = self.orig_reviewer_files
+
+        self._dir.cleanup()
+        os.environ.pop("GITEA_TEST_FORCE_DIRTY", None)
+        os.environ.pop("GITEA_TEST_PORCELAIN", None)
+
+    def _env(self, profile: str) -> dict:
+        return {
+            "GITEA_MCP_CONFIG": self.config_path,
+            "GITEA_MCP_PROFILE": profile,
+            "GITEA_TOKEN_AUTHOR": "author-pass",
+        }
+
+    @patch("mcp_server.api_request")
+    @patch("mcp_server.get_auth_header", return_value="token author-pass")
+    def test_preflight_not_called_blocks_commit(self, _auth, mock_api):
+        mock_api.return_value = {"login": "author-user"}
+        with patch.dict(os.environ, self._env("full-author"), clear=True):
+            os.environ["GITEA_TEST_PORCELAIN"] = ""
+            with self.assertRaises(RuntimeError) as ctx:
+                mcp_server.gitea_commit_files(
+                    files=[{"operation": "create", "path": "x.txt", "content": "YQ=="}],
+                    message="Add x",
+                    remote="prgs",
+                )
+            self.assertIn("Identity (gitea_whoami) has not been verified", str(ctx.exception))
+
+    @patch("mcp_server.api_request")
+    @patch("mcp_server.get_auth_header", return_value="token author-pass")
+    def test_dirty_workspace_before_whoami_blocks_commit(self, _auth, mock_api):
+        mock_api.return_value = {"login": "author-user"}
+        with patch.dict(os.environ, self._env("full-author"), clear=True):
+            os.environ["GITEA_TEST_FORCE_DIRTY"] = "1"
+            mcp_server.record_preflight_check("whoami")
+            mcp_server.record_preflight_check("capability", resolved_role="author")
+
+            with self.assertRaises(RuntimeError) as ctx:
+                mcp_server.gitea_commit_files(
+                    files=[{"operation": "create", "path": "x.txt", "content": "YQ=="}],
+                    message="Add x",
+                    remote="prgs",
+                )
+            self.assertIn("Workspace file edits occurred before gitea_whoami verification", str(ctx.exception))
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_mcp_server.py

@@ -1071,9 +1071,11 @@ class TestGetFile(unittest.TestCase):
 # ---------------------------------------------------------------------------
 class TestCommitFiles(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.get_auth_header", return_value=FAKE_AUTH)
-    def test_commit_files_success(self, _auth, mock_api):
+    def test_commit_files_success(self, _auth, mock_api, _role):
         mock_api.return_value = {
             "commit": {"sha": "commit-sha-123"},
             "branch": {"name": "test-branch"}
@@ -1081,11 +1083,15 @@ class TestCommitFiles(unittest.TestCase):
         files = [
             {"operation": "create", "path": "test.txt", "content": "SGVsbG8="}
         ]
-        result = gitea_commit_files(
-            files=files,
-            message="Initial commit",
-            new_branch="test-branch"
-        )
+        env = {
+            "GITEA_ALLOWED_OPERATIONS": "gitea.repo.commit",
+        }
+        with patch.dict(os.environ, env, clear=True):
+            result = gitea_commit_files(
+                files=files,
+                message="Initial commit",
+                new_branch="test-branch"
+            )
         self.assertTrue(result["success"])
         self.assertEqual(result["commit"], "commit-sha-123")
         self.assertEqual(result["branch"], "test-branch")

tests/test_review_proofs.py

@@ -33,9 +33,6 @@ from review_proofs import (  # noqa: E402
     assess_empty_queue_report,
     assess_fresh_issue_selection,
     assess_inventory_completeness,
-    assess_issue_filing_final_report,
-    assess_issue_filing_mutation_capability,
-    assess_issue_filing_sha_evidence,
     assess_issue_selection_final_report,
     assess_queue_target_final_report,
     assess_reviewer_queue_inventory,
@@ -1932,115 +1929,5 @@ class TestIssueSelectionContinuation(unittest.TestCase):
         self.assertEqual(result["grade"], "A")
 
 
-class TestIssueFilingFinalReport(unittest.TestCase):
-    """Issue #191: issue-filing runs need A-bar final report proofs."""
-
-    ISSUE_TITLE = (
-        "Implement fail-closed continuation mode for issues already "
-        "represented by open PRs"
-    )
-    FULL_SHA = "a4060c5de00f2b1c9e88f4f6f0f3f9a7b2c1d0e9"
-    CAPABILITIES = {
-        "create_issue": {
-            "requested_task": "create_issue",
-            "required_operation_permission": "gitea.issue.create",
-            "allowed_in_current_session": True,
-        },
-    }
-    CLOSEST = [{"number": 183, "title": "Harden author-run reporting", "state": "open"}]
-
-    def _good_report(self, *, sha_line=None):
-        lines = [
-            "Created Gitea-Tools #189 — "
-            f"`{self.ISSUE_TITLE}`",
-            "Duplicate check: searched 12 open issues.",
-            "Closest existing: #183 — Harden author-run reporting "
-            "(update rejected — different scope).",
-            "Why new issue justified: continuation mode is distinct from #183.",
-            "Only mutation: issue creation (gitea.issue.create).",
-            "Confirm no labels, comments, PRs, reviews, merges, or closes.",
-            "## Controller Handoff",
-            "- Task: file issue",
-            "- Repo: Scaled-Tech-Consulting/Gitea-Tools",
-            "- Role: author",
-            "- Identity: prgs-author",
-            "- Issue/PR: #189",
-            "- Branch/SHA: n/a",
-            "- Files changed: none",
-            "- Validation: duplicate gate + create_issue capability resolved",
-            "- Mutations: create_issue via gitea.issue.create",
-            "- Workspace mutations: none",
-            "- Current status: issue created",
-            "- Blockers: none",
-            "- Next: implementation",
-            "- Safety: no review/merge",
-            "- Issue created or updated: created #189",
-            "- Related issues: #183, #188",
-        ]
-        if sha_line:
-            lines.insert(4, sha_line)
-        return "\n".join(lines)
-
-    def test_complete_issue_filing_report_earns_a(self):
-        result = assess_issue_filing_final_report(
-            self._good_report(),
-            issue_number=189,
-            issue_title=self.ISSUE_TITLE,
-            action="created",
-            mutations=["create_issue"],
-            resolved_capabilities=self.CAPABILITIES,
-            issues_searched=12,
-            closest_matches=self.CLOSEST,
-            performed_mutations=["create_issue"],
-        )
-        self.assertEqual(result["grade"], "A")
-        self.assertFalse(result["downgraded"])
-
-    def test_missing_controller_handoff_downgrades(self):
-        report = self._good_report().replace("## Controller Handoff", "")
-        result = assess_issue_filing_final_report(
-            report,
-            issue_number=189,
-            issue_title=self.ISSUE_TITLE,
-            mutations=["create_issue"],
-            resolved_capabilities=self.CAPABILITIES,
-            issues_searched=12,
-            performed_mutations=["create_issue"],
-        )
-        self.assertTrue(result["downgraded"])
-
-    def test_abbreviated_sha_downgrades(self):
-        result = assess_issue_filing_sha_evidence(
-            f"old head: {self.FULL_SHA[:7]}"
-        )
-        self.assertTrue(result["downgraded"])
-
-    def test_mutation_without_capability_proof_downgrades(self):
-        report = self._good_report().replace("gitea.issue.create", "allowed")
-        result = assess_issue_filing_mutation_capability(
-            report,
-            ["create_issue"],
-            self.CAPABILITIES,
-        )
-        self.assertTrue(result["downgraded"])
-
-    def test_label_mutation_without_label_proof_downgrades(self):
-        caps = {
-            "set_issue_labels": {
-                "requested_task": "set_issue_labels",
-                "required_operation_permission": "gitea.issue.comment",
-                "allowed_in_current_session": True,
-            },
-        }
-        report = "\n".join([
-            "Updated labels with gitea.issue.comment capability resolved.",
-            "Only mutations: set_issue_labels",
-        ])
-        result = assess_issue_filing_mutation_capability(
-            report, ["set_issue_labels"], caps
-        )
-        self.assertTrue(result["downgraded"])
-
-
 if __name__ == "__main__":
     unittest.main()

tests/test_runtime_clarity.py

@@ -95,9 +95,13 @@ class TestRuntimeClarity(unittest.TestCase):
     # -------------------------------------------------------------------------
     # gitea_get_runtime_context
     # -------------------------------------------------------------------------
+    @patch(
+        "mcp_server.assess_preflight_status",
+        return_value={"preflight_ready": True, "preflight_block_reasons": []},
+    )
     @patch("mcp_server.api_request", return_value={"login": "author-user"})
     @patch("mcp_server.get_auth_header", return_value="token author-pass")
-    def test_get_runtime_context_author(self, _auth, _api):
+    def test_get_runtime_context_author(self, _auth, _api, _preflight):
         with patch.dict(os.environ, self._env("author-profile"), clear=True):
             ctx = mcp_server.gitea_get_runtime_context(remote="dadeschools")
         self.assertEqual(ctx["active_profile"], "author-profile")
@@ -110,10 +114,26 @@ class TestRuntimeClarity(unittest.TestCase):
         self.assertEqual(ctx["suggested_fix"], "reviewer namespace")
         self.assertIn("does not permit review or merge", ctx["review_merge_blocked_reasons"][0])
         self.assertIn("Switch to the reviewer MCP session", ctx["safe_next_action"])
+        caps = ctx["session_capabilities"]
+        self.assertTrue(caps["can_author_prs"])
+        self.assertFalse(caps["can_create_issues"])
+        self.assertFalse(caps["can_comment_on_issues"])
+        self.assertFalse(caps["can_review_prs"])
+        self.assertFalse(caps["can_merge_prs"])
+        self.assertTrue(caps["issue_comment_not_implied_by_pr_comment"])
+        merge_entry = next(
+            t for t in caps["task_capabilities"] if t["task"] == "merge_pr"
+        )
+        self.assertFalse(merge_entry["allowed_in_current_session"])
+        self.assertIn("reviewer-profile", merge_entry["matching_configured_profiles"])
 
+    @patch(
+        "mcp_server.assess_preflight_status",
+        return_value={"preflight_ready": True, "preflight_block_reasons": []},
+    )
     @patch("mcp_server.api_request", return_value={"login": "reviewer-user"})
     @patch("mcp_server.get_auth_header", return_value="token reviewer-pass")
-    def test_get_runtime_context_reviewer(self, _auth, _api):
+    def test_get_runtime_context_reviewer(self, _auth, _api, _preflight):
         with patch.dict(os.environ, self._env("reviewer-profile"), clear=True):
             ctx = mcp_server.gitea_get_runtime_context(remote="dadeschools")
         self.assertEqual(ctx["active_profile"], "reviewer-profile")
@@ -121,6 +141,15 @@ class TestRuntimeClarity(unittest.TestCase):
         self.assertTrue(ctx["review_merge_allowed"])
         self.assertEqual(ctx["suggested_fix"], "none")
         self.assertEqual(ctx["safe_next_action"], "None; ready for operations.")
+        caps = ctx["session_capabilities"]
+        self.assertFalse(caps["can_author_prs"])
+        self.assertFalse(caps["can_create_issues"])
+        self.assertFalse(caps["can_review_prs"])
+        self.assertTrue(caps["can_merge_prs"])
+        merge_entry = next(
+            t for t in caps["task_capabilities"] if t["task"] == "merge_pr"
+        )
+        self.assertTrue(merge_entry["allowed_in_current_session"])
 
     # -------------------------------------------------------------------------
     # gitea_list_profiles