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

2026-07-06 14:38:07 -04:00
10 changed files with 295 additions and 546 deletions
@@ -10,8 +10,3 @@ gitea-mcp*.json
 .vscode/
 graphify-out/
 branches/
 # Throwaway agent payload/encoding helpers (#261): delete after the MCP
 # commit completes or aborts; ignored here so strays cannot block issue locks.
 _encode_*.py
 _emit_*.py
 _inline_*.py
@@ -459,33 +459,6 @@ touching anything.
  files, detected secret, or any production/deploy behavior — **stop, report the
  blocker, and take no mutating action.** Fail closed; never work around a gate.
 ## Agent temp artifact cleanup (#261)
 Failed or aborted subagent runs have left throwaway helper scripts in the
 shared working tree — for example `_encode_commit_payload.py` and
 `_emit_payload.py` from the #152 commit-recovery attempt. These artifacts are
 never part of any issue scope. They pollute `git status`, which trips
 fail-closed gates: `gitea_lock_issue` refuses to lock over tracked edits, and
 the pre-flight purity checks attribute the dirt to the current session.
 Rules:
 - **Delete throwaway helpers as soon as the MCP commit completes or aborts.**
  Any file matching `_encode_*`, `_emit_*`, or `_inline_*` created as a
  scratch payload/encoding helper must be removed in the same session that
  created it — success and failure paths alike.
 - **Prefer the scratchpad.** Payload preparation belongs in the session
  scratchpad or a temp directory, never inside the repository worktree
  (see the shell-free commit payload work in #263).
 - **`.gitignore` backstop.** The patterns `_encode_*.py`, `_emit_*.py`, and
  `_inline_*.py` are ignored so a leftover helper cannot show up as an
  untracked file and block issue locks for later sessions. Ignoring them is a
  safety net, not permission to leave them behind.
 - **Cleanup checklist before ending any author session:** run
  `git status --porcelain` in the worktree you mutated; remove any
  `_encode_*` / `_emit_*` / `_inline_*` files you created; report the removal
  in the final run report like any other local mutation.
 ## Task/role alignment (#167)
 The **requested task** decides what a session may do — not the credential it
@@ -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)
@@ -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,309 +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 not in text:
                continue
            denial_phrases = (
                f"no {absent}",
                f"no {absent}s",
                f"without {absent}",
                "none performed",
                "none.",
                "confirm no",
            )
            if any(phrase in text for phrase in denial_phrases):
                continue
            reasons.append(
                f"single-mutation run mentions '{absent}' without "
                f"confirming it was not performed"
            )
        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():
        verdict = result.get("verdict")
        if verdict in ("missing", "incomplete"):
            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(
@@ -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:`,
@@ -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",
 }
@@ -1,76 +0,0 @@
 """Docs checks for agent temp-artifact cleanup guidance (#261).
 Failed subagent runs left throwaway helper scripts (e.g.
 ``_encode_commit_payload.py``, ``_emit_payload.py``) in the shared working
 tree, polluting ``git status`` and tripping issue-lock / preflight gates.
 These tests keep the cleanup runbook guidance present and the ignore
 patterns in place so stray helpers can never dirty a worktree again.
 """
 from pathlib import Path
 REPO_ROOT = Path(__file__).resolve().parent.parent
 RUNBOOK = REPO_ROOT / "docs" / "llm-workflow-runbooks.md"
 GITIGNORE = REPO_ROOT / ".gitignore"
 ARTIFACT_PATTERNS = ("_encode_*.py", "_emit_*.py", "_inline_*.py")
 def _runbook_text():
    assert RUNBOOK.is_file(), "missing docs/llm-workflow-runbooks.md"
    return RUNBOOK.read_text(encoding="utf-8")
 def _gitignore_text():
    assert GITIGNORE.is_file(), "missing .gitignore"
    return GITIGNORE.read_text(encoding="utf-8")
 def test_runbook_has_agent_temp_artifact_cleanup_section():
    text = _runbook_text()
    assert "## Agent temp artifact cleanup" in text, (
        "runbook lacks an 'Agent temp artifact cleanup' section")
    assert "#261" in text, "cleanup guidance does not reference issue #261"
 def test_runbook_names_known_stray_artifacts():
    text = _runbook_text()
    for name in ("_encode_commit_payload.py", "_emit_payload.py"):
        assert name in text, f"runbook does not name stray artifact {name!r}"
 def test_runbook_covers_all_artifact_patterns():
    text = _runbook_text()
    for pattern in ("_encode_*", "_emit_*", "_inline_*"):
        assert pattern in text, f"runbook does not cover pattern {pattern!r}"
 def test_runbook_requires_cleanup_after_commit_or_abort():
    text = _runbook_text().lower()
    assert "delete" in text or "remove" in text
    for phase in ("completes", "aborts"):
        assert phase in text, (
            f"runbook does not require cleanup when the MCP commit {phase}")
 def test_runbook_explains_lock_and_preflight_impact():
    text = _runbook_text().lower()
    assert "gitea_lock_issue" in text
    assert "preflight" in text or "pre-flight" in text
 def test_gitignore_blocks_agent_helper_patterns():
    lines = [
        ln.strip() for ln in _gitignore_text().splitlines()
        if ln.strip() and not ln.strip().startswith("#")
    ]
    for pattern in ARTIFACT_PATTERNS:
        assert pattern in lines, f".gitignore lacks pattern {pattern!r}"
 def test_no_stray_agent_helpers_in_repo_root():
    """Acceptance: artifacts stay removed from default worktrees."""
    stray = [
        p.name for pattern in ARTIFACT_PATTERNS
        for p in REPO_ROOT.glob(pattern)
    ]
    assert not stray, f"stray agent helper artifacts present: {stray}"
@@ -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()
@@ -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(
+        env = {
-            files=files,
+            "GITEA_ALLOWED_OPERATIONS": "gitea.repo.commit",
-            message="Initial commit",
+        }
-            new_branch="test-branch"
+        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")
@@ -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()