Merge branch 'master' into feat/issue-266-subagent-gate-inheritance

.gitignore

@@ -10,3 +10,7 @@ gitea-mcp*.json
 .vscode/
 graphify-out/
 branches/
+# Throwaway agent commit-encoding helpers (#261) — never commit.
+/_encode_*.py
+/_emit_*.py
+/_inline_*.py

agent_temp_artifacts.py

@@ -0,0 +1,26 @@
+"""Detect throwaway agent helper scripts left in the repo root (#261)."""
+
+from __future__ import annotations
+
+import fnmatch
+
+AGENT_TEMP_BASENAME_PATTERNS = (
+    "_encode_*.py",
+    "_emit_*.py",
+    "_inline_*.py",
+)
+
+
+def find_agent_temp_artifacts_from_porcelain(porcelain: str) -> list[str]:
+    """Return untracked repo-root helper paths matching agent temp patterns."""
+    found: list[str] = []
+    for line in (porcelain or "").splitlines():
+        if not line.startswith("??"):
+            continue
+        path = line[3:].strip()
+        if not path or "/" in path or "\\" in path:
+            continue
+        basename = path.split("/")[-1]
+        if any(fnmatch.fnmatch(basename, pat) for pat in AGENT_TEMP_BASENAME_PATTERNS):
+            found.append(path)
+    return sorted(found)

docs/llm-workflow-runbooks.md

@@ -316,6 +316,30 @@ may edit another issue's branch folder unless explicitly assigned to that issue.
 No LLM may clean another issue's branch folder unless the PR is merged or closed
 and cleanup is explicitly part of the task.
 
+## Agent temp artifact cleanup (#261)
+
+Failed or aborted MCP commit attempts sometimes leave throwaway helper scripts in
+the **repository root**. These are not part of any issue scope and pollute
+`git status`, which can break `gitea_lock_issue` and preflight checks.
+
+**Patterns (repo root only, untracked):**
+
+- `_encode_*.py` — base64 payload encoders
+- `_emit_*.py` — commit payload emitters
+- `_inline_*.py` — inline encoding helpers
+
+**Required cleanup (after MCP commit completes or aborts):**
+
+1. Delete any matching files at the repo root (`rm ./_encode_*.py` etc.).
+2. Confirm `git status` is clean on the orchestration checkout before
+   `gitea_lock_issue`.
+3. Prefer native `gitea_commit_files` / gated commit paths — do not leave shell
+   encoding fallbacks behind.
+
+Root-level matches are listed in `.gitignore` so they never get committed.
+`gitea_get_runtime_context` and `gitea_lock_issue` surface **warnings** (not
+hard blocks) when these artifacts are still present.
+
 Implementation work and review work must use separate branch folders. For
 example, an implementation branch might live under
 `branches/fix-issue-123-example`, while a review branch for the resulting PR

gitea_mcp_server.py

@@ -260,9 +260,19 @@ def assess_preflight_status() -> dict:
             "Reviewer profile modified tracked workspace files after capability resolution "
             f"(offending files: {_format_preflight_files(_preflight_reviewer_violation_files)})"
         )
+    warnings: list[str] = []
+    agent_artifacts = agent_temp_artifacts.find_agent_temp_artifacts_from_porcelain(
+        _get_workspace_porcelain()
+    )
+    if agent_artifacts:
+        warnings.append(
+            "Agent temp artifacts detected at repo root (delete before mutations): "
+            f"{_format_preflight_files(agent_artifacts)}"
+        )
     return {
         "preflight_ready": not reasons,
         "preflight_block_reasons": reasons,
+        "preflight_warnings": warnings,
         "preflight_whoami_verified": _preflight_whoami_called,
         "preflight_capability_resolved": _preflight_capability_called,
         "preflight_whoami_violation_files": list(_preflight_whoami_violation_files),
@@ -373,6 +383,7 @@ import role_session_router  # noqa: E402
 import role_namespace_gate  # noqa: E402
 import task_capability_map  # noqa: E402
 import review_proofs  # noqa: E402
+import agent_temp_artifacts
 import issue_lock_worktree  # noqa: E402
 
 
@@ -877,7 +888,10 @@ def gitea_lock_issue(
     except Exception as e:
         raise RuntimeError(f"Could not write issue lock file: {e}")
 
-    return {
+    agent_artifacts = agent_temp_artifacts.find_agent_temp_artifacts_from_porcelain(
+        git_state.get("porcelain_status") or ""
+    )
+    result = {
         "success": True,
         "message": (
             f"Successfully locked issue #{issue_number} to branch '{branch_name}' "
@@ -887,6 +901,12 @@ def gitea_lock_issue(
         "branch_name": branch_name,
         "worktree_path": resolved_worktree,
     }
+    if agent_artifacts:
+        result["warnings"] = [
+            "Agent temp artifacts at repo root (delete before implementation): "
+            + ", ".join(agent_artifacts)
+        ]
+    return result
 
 
 @mcp.tool()
@@ -2275,6 +2295,106 @@ def gitea_get_file(
     }
 
 
+def _prepare_commit_payload_files(files: list[dict]) -> tuple[list[dict], list[dict]]:
+    import base64
+    import json
+    import os
+
+    processed_files = []
+    source_proofs = []
+
+    lock_data = {}
+    if os.path.exists(ISSUE_LOCK_FILE):
+        try:
+            with open(ISSUE_LOCK_FILE, "r", encoding="utf-8") as f:
+                lock_data = json.load(f)
+        except Exception:
+            pass
+
+    locked_worktree = lock_data.get("worktree_path")
+    if locked_worktree:
+        locked_worktree = os.path.realpath(locked_worktree)
+
+    for f in files:
+        f_copy = dict(f)
+        path = f_copy.get("path", "")
+
+        content_keys = [k for k in ["content", "content_plain", "workspace_path", "local_path"] if k in f_copy]
+        if len(content_keys) > 1:
+            raise ValueError(
+                f"Multiple content sources specified for file '{path}'; provide exactly one of "
+                f"'content', 'content_plain', 'workspace_path', or 'local_path' (fail closed)"
+            )
+
+        source = "none"
+        if "content_plain" in f_copy:
+            content_plain = f_copy.pop("content_plain")
+            if content_plain is not None:
+                content_bytes = content_plain.encode("utf-8")
+                f_copy["content"] = base64.b64encode(content_bytes).decode("utf-8")
+                source = "inline_plain"
+        elif "workspace_path" in f_copy:
+            workspace_path = f_copy.pop("workspace_path")
+            if not locked_worktree:
+                raise RuntimeError(
+                    f"Issue lock is missing or does not define a worktree path. "
+                    f"Cannot resolve workspace_path '{workspace_path}' (fail closed)."
+                )
+            target_path = os.path.realpath(os.path.abspath(os.path.join(locked_worktree, workspace_path)))
+
+            is_inside = target_path == locked_worktree or target_path.startswith(os.path.join(locked_worktree, ""))
+            is_tmp = target_path.startswith(os.path.realpath("/tmp") + os.sep) or target_path.startswith("/tmp" + os.sep)
+
+            if not (is_inside or is_tmp):
+                raise ValueError(
+                    f"workspace_path '{workspace_path}' resolves to '{target_path}' which falls outside "
+                    f"of locked worktree '{locked_worktree}' (fail closed)"
+                )
+
+            if not os.path.exists(target_path):
+                raise FileNotFoundError(f"File not found: {target_path} (fail closed)")
+
+            with open(target_path, "rb") as fh:
+                file_bytes = fh.read()
+            f_copy["content"] = base64.b64encode(file_bytes).decode("utf-8")
+            source = "workspace_path"
+        elif "local_path" in f_copy:
+            local_path = f_copy.pop("local_path")
+            if not locked_worktree:
+                raise RuntimeError(
+                    f"Issue lock is missing or does not define a worktree path. "
+                    f"Cannot resolve local_path '{local_path}' (fail closed)."
+                )
+            target_path = os.path.realpath(os.path.abspath(local_path))
+
+            is_inside = target_path == locked_worktree or target_path.startswith(os.path.join(locked_worktree, ""))
+            is_tmp = target_path.startswith(os.path.realpath("/tmp") + os.sep) or target_path.startswith("/tmp" + os.sep)
+
+            if not (is_inside or is_tmp):
+                raise ValueError(
+                    f"local_path '{local_path}' resolves to '{target_path}' which falls outside "
+                    f"of locked worktree '{locked_worktree}' (fail closed)"
+                )
+
+            if not os.path.exists(target_path):
+                raise FileNotFoundError(f"File not found: {target_path} (fail closed)")
+
+            with open(target_path, "rb") as fh:
+                file_bytes = fh.read()
+            f_copy["content"] = base64.b64encode(file_bytes).decode("utf-8")
+            source = "local_path"
+        elif "content" in f_copy:
+            source = "inline_base64"
+
+        processed_files.append(f_copy)
+        source_proofs.append({
+            "path": path,
+            "source": source
+        })
+
+    return processed_files, source_proofs
+
+
 @mcp.tool()
 def gitea_commit_files(
     files: list[dict],
@@ -2289,7 +2409,7 @@ def gitea_commit_files(
     """Commit changes to multiple files in a Gitea repository in a single atomic commit.
 
     Args:
-        files: List of file operations. Each file dict must contain 'operation' ('create', 'update', 'delete', 'rename'), 'path', and 'content' (base64 encoded for create/update), and optionally 'sha' (required for update/delete) or 'from_path' (for rename).
+        files: List of file operations. Each file dict must contain 'operation' ('create', 'update', 'delete', 'rename'), 'path', and one of content payload sources: 'content' (base64), 'content_plain', 'workspace_path', or 'local_path'.
         message: The commit message.
         branch: Optional existing branch to start/commit from.
         new_branch: Optional new branch name to create for this commit.
@@ -2301,13 +2421,38 @@ 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)
+    processed_files, source_proofs = _prepare_commit_payload_files(files)
+
     h, o, r = _resolve(remote, host, org, repo)
     auth = _auth(h)
     url = f"{repo_api_url(h, o, r)}/contents"
 
     payload = {
-        "files": files,
+        "files": processed_files,
         "message": message,
     }
     if branch is not None:
@@ -2318,13 +2463,14 @@ def gitea_commit_files(
     with _audited("commit_files", host=h, remote=remote, org=o, repo=r,
                   target_branch=(new_branch or branch),
                   request_metadata={"message": message,
-                                    "paths": [f.get("path") for f in files],
-                                    "operations": [f.get("operation") for f in files]}):
+                                    "paths": [f.get("path") for f in processed_files],
+                                    "operations": [f.get("operation") for f in processed_files]}):
         data = api_request("POST", url, auth, payload)
     return {
         "success": True,
         "commit": data.get("commit", {}).get("sha", ""),
         "branch": data.get("branch", {}).get("name", ""),
+        "content_source_proof": source_proofs,
     }
 
 

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_agent_temp_artifacts.py

@@ -0,0 +1,87 @@
+"""Tests for agent temp artifact detection and preflight warnings (#261)."""
+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 agent_temp_artifacts
+import mcp_server
+
+
+class TestAgentTempArtifactPatterns(unittest.TestCase):
+    def test_detects_root_level_encode_emit_inline(self):
+        porcelain = (
+            "?? _encode_commit_payload.py\n"
+            "?? _emit_payload.py\n"
+            "?? _inline_b64.py\n"
+        )
+        self.assertEqual(
+            agent_temp_artifacts.find_agent_temp_artifacts_from_porcelain(porcelain),
+            ["_emit_payload.py", "_encode_commit_payload.py", "_inline_b64.py"],
+        )
+
+    def test_ignores_nested_and_unrelated_untracked(self):
+        porcelain = (
+            "?? scripts/_encode_x.py\n"
+            "?? README-draft.md\n"
+            " M task_capability_map.py\n"
+        )
+        self.assertEqual(
+            agent_temp_artifacts.find_agent_temp_artifacts_from_porcelain(porcelain),
+            [],
+        )
+
+
+class TestPreflightWarnings(unittest.TestCase):
+    def setUp(self):
+        self._snapshot = (
+            mcp_server._preflight_whoami_called,
+            mcp_server._preflight_capability_called,
+        )
+        mcp_server._preflight_whoami_called = True
+        mcp_server._preflight_capability_called = True
+
+    def tearDown(self):
+        mcp_server._preflight_whoami_called, mcp_server._preflight_capability_called = (
+            self._snapshot
+        )
+
+    @patch(
+        "mcp_server._get_workspace_porcelain",
+        return_value="?? _encode_commit_payload.py\n",
+    )
+    def test_assess_preflight_surfaces_warning_not_block(self, _porcelain):
+        status = mcp_server.assess_preflight_status()
+        self.assertTrue(status["preflight_ready"])
+        self.assertEqual(status["preflight_block_reasons"], [])
+        self.assertEqual(len(status["preflight_warnings"]), 1)
+        self.assertIn("_encode_commit_payload.py", status["preflight_warnings"][0])
+
+
+class TestIssueLockArtifactWarning(unittest.TestCase):
+    @patch("mcp_server.api_get_all", return_value=[])
+    @patch("mcp_server._auth", return_value="token x")
+    @patch("mcp_server._resolve", return_value=("h", "o", "r"))
+    @patch("mcp_server.ISSUE_LOCK_FILE", new_callable=lambda: tempfile.mktemp())
+    @patch("issue_lock_worktree.read_worktree_git_state")
+    def test_lock_success_includes_artifact_warning(self, mock_state, _lock_file, *_mocks):
+        mock_state.return_value = {
+            "current_branch": "master",
+            "porcelain_status": "?? _emit_payload.py\n",
+        }
+        result = mcp_server.gitea_lock_issue(
+            issue_number=261,
+            branch_name="docs/issue-261-agent-artifact-cleanup",
+            remote="prgs",
+        )
+        self.assertTrue(result["success"])
+        self.assertIn("warnings", result)
+        self.assertIn("_emit_payload.py", result["warnings"][0])
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_commit_files_capability.py

@@ -0,0 +1,199 @@
+"""Tests for commit_files task capability resolver mapping (#262)."""
+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
+
+CONFIG = {
+    "version": 2,
+    "contexts": {
+        "ctx": {
+            "enabled": True,
+            "gitea": {"enabled": True, "base_url": "https://gitea.example.com"},
+        }
+    },
+    "profiles": {
+        "commit-author": {
+            "enabled": True,
+            "context": "ctx",
+            "role": "author",
+            "username": "author-user",
+            "auth": {"type": "env", "name": "GITEA_TOKEN_AUTHOR"},
+            "allowed_operations": ["gitea.read", "gitea.repo.commit"],
+            "forbidden_operations": ["gitea.pr.create"],
+            "execution_profile": "commit-author",
+        },
+        "pr-only-author": {
+            "enabled": True,
+            "context": "ctx",
+            "role": "author",
+            "username": "pr-author",
+            "auth": {"type": "env", "name": "GITEA_TOKEN_AUTHOR"},
+            "allowed_operations": ["gitea.read", "gitea.pr.create"],
+            "forbidden_operations": ["gitea.repo.commit"],
+            "execution_profile": "pr-only-author",
+        },
+        "reviewer-profile": {
+            "enabled": True,
+            "context": "ctx",
+            "role": "reviewer",
+            "username": "reviewer-user",
+            "auth": {"type": "env", "name": "GITEA_TOKEN_REVIEWER"},
+            "allowed_operations": ["gitea.read", "gitea.pr.review", "gitea.pr.merge"],
+            "forbidden_operations": [
+                "gitea.repo.commit", "gitea.pr.create", "gitea.branch.push",
+            ],
+            "execution_profile": "reviewer-profile",
+        },
+    },
+    "rules": {"allow_runtime_switching": False},
+}
+
+
+class CommitFilesCapabilityBase(unittest.TestCase):
+    def setUp(self):
+        self._preflight_snapshot = (
+            mcp_server._preflight_whoami_called,
+            mcp_server._preflight_capability_called,
+        )
+        mcp_server._preflight_whoami_called = False
+        mcp_server._preflight_capability_called = False
+        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._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, mcp_server._preflight_capability_called = (
+            self._preflight_snapshot
+        )
+        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",
+        }
+
+
+class TestCommitFilesResolver(CommitFilesCapabilityBase):
+    @patch("mcp_server.api_request", return_value={"login": "author-user"})
+    @patch("mcp_server.get_auth_header", return_value="token author-pass")
+    def test_commit_files_resolves_to_repo_commit(self, _auth, _api):
+        with patch.dict(os.environ, self._env("commit-author"), clear=True):
+            for task in ("commit_files", "gitea_commit_files"):
+                with self.subTest(task=task):
+                    res = mcp_server.gitea_resolve_task_capability(
+                        task=task, remote="prgs")
+                    self.assertEqual(res["required_operation_permission"],
+                                     "gitea.repo.commit")
+                    self.assertEqual(res["required_role_kind"], "author")
+                    self.assertTrue(res["allowed_in_current_session"])
+
+    @patch("mcp_server.api_request", return_value={"login": "pr-author"})
+    @patch("mcp_server.get_auth_header", return_value="token author-pass")
+    def test_create_pr_does_not_satisfy_commit_files(self, _auth, _api):
+        with patch.dict(os.environ, self._env("pr-only-author"), clear=True):
+            commit_res = mcp_server.gitea_resolve_task_capability(
+                task="commit_files", remote="prgs")
+            pr_res = mcp_server.gitea_resolve_task_capability(
+                task="create_pr", remote="prgs")
+            self.assertFalse(commit_res["allowed_in_current_session"])
+            self.assertTrue(pr_res["allowed_in_current_session"])
+            self.assertEqual(commit_res["required_operation_permission"],
+                             "gitea.repo.commit")
+            self.assertEqual(pr_res["required_operation_permission"],
+                             "gitea.pr.create")
+
+    @patch("mcp_server.api_request", return_value={"login": "reviewer-user"})
+    @patch("mcp_server.get_auth_header", return_value="token reviewer-pass")
+    def test_reviewer_denied_commit_files(self, _auth, _api):
+        with patch.dict(os.environ, self._env("reviewer-profile"), clear=True):
+            res = mcp_server.gitea_resolve_task_capability(
+                task="commit_files", remote="prgs")
+            self.assertFalse(res["allowed_in_current_session"])
+            self.assertEqual(res["required_operation_permission"],
+                             "gitea.repo.commit")
+            self.assertTrue(res["different_mcp_namespace_required"])
+
+
+class TestCommitFilesToolGate(CommitFilesCapabilityBase):
+    @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": "abc"},
+            "branch": {"name": "feat/x"},
+        }
+        mcp_server.record_preflight_check("whoami")
+        mcp_server.record_preflight_check("capability", resolved_role="author")
+        with patch.dict(os.environ, self._env("commit-author"), clear=True):
+            res = mcp_server.gitea_commit_files(
+                files=[{"operation": "create", "path": "a.txt", "content": "YQ=="}],
+                message="test",
+                new_branch="feat/x",
+                remote="prgs",
+            )
+        self.assertTrue(res["success"])
+
+    @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
+           return_value=(True, []))
+    @patch("mcp_server.get_auth_header", return_value="token author-pass")
+    def test_reviewer_blocked_with_permission_report(self, _auth, _role):
+        mcp_server.record_preflight_check("whoami")
+        mcp_server.record_preflight_check("capability", resolved_role="author")
+        with patch.dict(os.environ, self._env("reviewer-profile"), clear=True):
+            res = mcp_server.gitea_commit_files(
+                files=[{"operation": "create", "path": "a.txt", "content": "YQ=="}],
+                message="test",
+                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")
+
+    @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
+           return_value=(True, []))
+    def test_preflight_blocks_before_api(self, _role):
+        env = {**self._env("commit-author"), "GITEA_TEST_FORCE_DIRTY": "1"}
+        with patch.dict(os.environ, env, clear=True):
+            with self.assertRaises(RuntimeError) as ctx:
+                mcp_server.gitea_commit_files(
+                    files=[{"operation": "create", "path": "a.txt", "content": "YQ=="}],
+                    message="test",
+                    remote="prgs",
+                )
+        self.assertIn("gitea_whoami", str(ctx.exception))
+
+
+class TestCommitFilesMapAlignment(unittest.TestCase):
+    def test_tool_gate_matches_resolver(self):
+        self.assertEqual(
+            task_capability_map.tool_required_permission("gitea_commit_files"),
+            task_capability_map.required_permission("commit_files"),
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()

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_commit_payloads.py

@@ -0,0 +1,251 @@
+import json
+import os
+import tempfile
+import unittest
+from unittest.mock import MagicMock, patch
+
+import gitea_config
+import mcp_server
+import task_capability_map
+
+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.repo.commit",
+                "gitea.pr.create",
+                "gitea.branch.push",
+            ],
+            "forbidden_operations": [
+                "gitea.pr.approve",
+                "gitea.pr.merge",
+                "gitea.pr.review",
+            ],
+            "execution_profile": "full-author",
+        },
+    },
+}
+
+
+class TestCommitPayloads(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()
+
+        # Setup temp directories and lock 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))
+
+        self.locked_worktree_dir = tempfile.TemporaryDirectory()
+        self.locked_worktree_path = os.path.realpath(self.locked_worktree_dir.name)
+
+        self.lock_file_path = "/tmp/gitea_issue_lock.json"
+        self.lock_data = {
+            "issue_number": 263,
+            "branch_name": "feat/issue-263-native-commit-payloads",
+            "remote": "prgs",
+            "org": "Example-Org",
+            "repo": "Example-Repo",
+            "worktree_path": self.locked_worktree_path,
+        }
+        with open(self.lock_file_path, "w", encoding="utf-8") as fh:
+            fh.write(json.dumps(self.lock_data))
+
+        # Reset preflight status to bypass/pass verification in tests
+        self.orig_whoami_called = mcp_server._preflight_whoami_called
+        self.orig_capability_called = mcp_server._preflight_capability_called
+        mcp_server._preflight_whoami_called = True
+        mcp_server._preflight_capability_called = True
+
+    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
+
+        self._dir.cleanup()
+        self.locked_worktree_dir.cleanup()
+        if os.path.exists(self.lock_file_path):
+            os.remove(self.lock_file_path)
+
+    def _env(self, profile: str) -> dict:
+        return {
+            "GITEA_MCP_CONFIG": self.config_path,
+            "GITEA_MCP_PROFILE": profile,
+            "GITEA_TOKEN_AUTHOR": "author-pass",
+            "GITEA_TEST_PORCELAIN": "",
+        }
+
+    @patch("mcp_server.api_request")
+    @patch("mcp_server.get_auth_header", return_value="token author-pass")
+    def test_commit_files_content_plain(self, _auth, mock_api):
+        mock_api.return_value = {
+            "commit": {"sha": "sha-123"},
+            "branch": {"name": "feat/issue-263-native-commit-payloads"}
+        }
+        with patch.dict(os.environ, self._env("full-author"), clear=True):
+            res = mcp_server.gitea_commit_files(
+                files=[
+                    {
+                        "operation": "create",
+                        "path": "hello.txt",
+                        "content_plain": "Hello World!",
+                    }
+                ],
+                message="Add hello.txt",
+                remote="prgs",
+            )
+        print("DEBUG RES IS:", res)
+        self.assertTrue(res["success"])
+        self.assertEqual(res["commit"], "sha-123")
+        self.assertEqual(res["content_source_proof"][0]["source"], "inline_plain")
+
+        # Verify posted base64 content
+        post_args = mock_api.call_args[0]
+        payload = post_args[3]
+        self.assertEqual(payload["files"][0]["content"], "SGVsbG8gV29ybGQh")
+
+    @patch("mcp_server.api_request")
+    @patch("mcp_server.get_auth_header", return_value="token author-pass")
+    def test_commit_files_workspace_path(self, _auth, mock_api):
+        mock_api.return_value = {
+            "commit": {"sha": "sha-456"},
+            "branch": {"name": "feat/issue-263-native-commit-payloads"}
+        }
+
+        # Create a workspace file
+        file_relative = "src/code.py"
+        file_abs = os.path.join(self.locked_worktree_path, file_relative)
+        os.makedirs(os.path.dirname(file_abs), exist_ok=True)
+        with open(file_abs, "wb") as fh:
+            fh.write(b"print('hello')")
+
+        with patch.dict(os.environ, self._env("full-author"), clear=True):
+            res = mcp_server.gitea_commit_files(
+                files=[
+                    {
+                        "operation": "create",
+                        "path": "src/code.py",
+                        "workspace_path": file_relative,
+                    }
+                ],
+                message="Add code.py",
+                remote="prgs",
+            )
+        self.assertTrue(res["success"])
+        self.assertEqual(res["content_source_proof"][0]["source"], "workspace_path")
+
+        post_args = mock_api.call_args[0]
+        payload = post_args[3]
+        # "print('hello')" in base64 is cHJpbnQoJ2hlbGxvJyk=
+        self.assertEqual(payload["files"][0]["content"], "cHJpbnQoJ2hlbGxvJyk=")
+
+    @patch("mcp_server.api_request")
+    @patch("mcp_server.get_auth_header", return_value="token author-pass")
+    def test_commit_files_local_path(self, _auth, mock_api):
+        mock_api.return_value = {
+            "commit": {"sha": "sha-789"},
+            "branch": {"name": "feat/issue-263-native-commit-payloads"}
+        }
+
+        file_abs = os.path.join(self.locked_worktree_path, "local.bin")
+        with open(file_abs, "wb") as fh:
+            fh.write(b"\x00\x01\x02\x03\xff")
+
+        with patch.dict(os.environ, self._env("full-author"), clear=True):
+            res = mcp_server.gitea_commit_files(
+                files=[
+                    {
+                        "operation": "create",
+                        "path": "local.bin",
+                        "local_path": file_abs,
+                    }
+                ],
+                message="Add local.bin",
+                remote="prgs",
+            )
+        self.assertTrue(res["success"])
+        self.assertEqual(res["content_source_proof"][0]["source"], "local_path")
+
+        post_args = mock_api.call_args[0]
+        payload = post_args[3]
+        # b"\x00\x01\x02\x03\xff" in base64 is AAECA/8=
+        self.assertEqual(payload["files"][0]["content"], "AAECA/8=")
+
+    @patch("mcp_server.api_request")
+    @patch("mcp_server.get_auth_header", return_value="token author-pass")
+    def test_commit_files_traversal_blocked(self, _auth, mock_api):
+        # Remove active lock file to ensure it fails on traversal/invalid locks
+        os.remove(self.lock_file_path)
+
+        with patch.dict(os.environ, self._env("full-author"), clear=True):
+            with self.assertRaises(RuntimeError) as ctx:
+                mcp_server.gitea_commit_files(
+                    files=[
+                        {
+                            "operation": "create",
+                            "path": "hack.txt",
+                            "workspace_path": "any.txt",
+                        }
+                    ],
+                    message="Add hack",
+                    remote="prgs",
+                )
+            self.assertIn("Issue lock is missing", str(ctx.exception))
+
+    @patch("mcp_server.api_request")
+    @patch("mcp_server.get_auth_header", return_value="token author-pass")
+    def test_commit_files_outside_scope_blocked(self, _auth, mock_api):
+        with patch.dict(os.environ, self._env("full-author"), clear=True):
+            with self.assertRaises(ValueError) as ctx:
+                mcp_server.gitea_commit_files(
+                    files=[
+                        {
+                            "operation": "create",
+                            "path": "hack.txt",
+                            "workspace_path": "../outside.txt",
+                        }
+                    ],
+                    message="Add hack",
+                    remote="prgs",
+                )
+            self.assertIn("falls outside of locked worktree", str(ctx.exception))
+
+    @patch("mcp_server.api_request")
+    @patch("mcp_server.get_auth_header", return_value="token author-pass")
+    def test_commit_files_multiple_sources_blocked(self, _auth, mock_api):
+        with patch.dict(os.environ, self._env("full-author"), clear=True):
+            with self.assertRaises(ValueError) as ctx:
+                mcp_server.gitea_commit_files(
+                    files=[
+                        {
+                            "operation": "create",
+                            "path": "hello.txt",
+                            "content_plain": "Hello!",
+                            "content": "SGVsbG8=",
+                        }
+                    ],
+                    message="Add hello",
+                    remote="prgs",
+                )
+            self.assertIn("Multiple content sources specified", str(ctx.exception))

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")