feat(issue-filing): harden final report proofs for handoff and mutations (#191)

Add assess_issue_filing_final_report and supporting checks for Controller
Handoff (issue_filing role), exact issue number/title, full 40-char SHAs,
per-mutation capability proof, single-mutation scope statements, and
duplicate-search summaries. Workflow skill documents the issue-filing
handoff fields.

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

gitea_mcp_server.py

@@ -2301,29 +2301,6 @@ 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)
@@ -4126,90 +4103,6 @@ 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",
@@ -4297,10 +4190,6 @@ 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 = (
@@ -4325,7 +4214,6 @@ 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,6 +1297,16 @@ 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",)),
+    ),
 }
 
 
@@ -2135,6 +2145,304 @@ 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,6 +462,12 @@ 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,14 +80,6 @@ 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.
@@ -97,7 +89,6 @@ 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

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

tests/test_review_proofs.py

@@ -33,6 +33,9 @@ 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,
@@ -1929,5 +1932,115 @@ 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,13 +95,9 @@ 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, _preflight):
+    def test_get_runtime_context_author(self, _auth, _api):
         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")
@@ -114,26 +110,10 @@ 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, _preflight):
+    def test_get_runtime_context_reviewer(self, _auth, _api):
         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")
@@ -141,15 +121,6 @@ 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