Scaled-Tech-Consulting/Gitea-Tools
2
0
Fork 0
Code Issues 74 Pull Requests 9 Actions Packages Projects Releases Wiki Activity

merge: resolve master conflicts for #259 subagent tool-budget docs

Browse Source 
Integrate split-workflow router SKILL.md (#333) with subagent tool-budget
guardrails (#259). Keep shell spawn hard-stop (#258) and subagent
delegation (#266) operator-guide rules alongside subagent_tool_budget.
This commit is contained in:
sysadmin
2026-07-07 04:35:46 -04:00
parent 82e6d3a3f4 fd396df334
commit edf26410a0
39 changed files with 6471 additions and 690 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_agent_temp_artifacts.py
+87
View File
@@ -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
+199
View File
@@ -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
+251
View File
@@ -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
+251
View File
@@ -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_identity_disclosure.py
+102
View File
@@ -0,0 +1,102 @@
"""Identity-disclosure checks for workflow reports (#305).
Workflow reports sometimes included the authenticated user's personal
email even though username/profile identity is sufficient (observed:
``Identity: jcwalker3 / [email protected]`` in a reconciliation
handoff). These tests pin the no-email identity summary helper and the
final-report validator that flags unnecessary email disclosure.
"""
import sys
import unittest
sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
import review_proofs
class TestIdentitySummary(unittest.TestCase):
def test_author_summary_uses_username_and_profile_only(self):
summary = review_proofs.format_identity_summary(
"jcwalker3", "prgs-author")
self.assertEqual(summary, "jcwalker3 / prgs-author")
self.assertNotIn("@", summary)
def test_reviewer_summary_uses_username_and_profile_only(self):
summary = review_proofs.format_identity_summary(
"sysadmin", "prgs-reviewer")
self.assertEqual(summary, "sysadmin / prgs-reviewer")
def test_summary_appends_role_and_remote_without_email(self):
summary = review_proofs.format_identity_summary(
"jcwalker3", "prgs-author", role="author", remote="prgs")
self.assertIn("jcwalker3 / prgs-author", summary)
self.assertIn("author", summary)
self.assertIn("prgs", summary)
self.assertNotIn("@", summary)
def test_summary_never_leaks_email_passed_as_username(self):
"""Defense in depth: an email in the username slot is reduced."""
summary = review_proofs.format_identity_summary(
"[email protected]", "prgs-author")
self.assertNotIn("@", summary)
self.assertIn("jcwalker3", summary)
class TestEmailDisclosureAssessment(unittest.TestCase):
def test_report_without_email_passes(self):
report = "\n".join([
"Controller Handoff",
"Identity: jcwalker3 / prgs-author",
"Role: author",
])
res = review_proofs.assess_email_disclosure(report)
self.assertTrue(res["proven"])
self.assertFalse(res["flagged"])
self.assertEqual(res["emails"], [])
self.assertEqual(res["reasons"], [])
def test_unnecessary_email_is_flagged(self):
report = "\n".join([
"Controller Handoff",
"Identity: jcwalker3 / [email protected]",
])
res = review_proofs.assess_email_disclosure(report)
self.assertFalse(res["proven"])
self.assertTrue(res["flagged"])
self.assertIn("[email protected]", res["emails"])
self.assertTrue(res["reasons"])
self.assertFalse(res["justified"])
def test_multiple_emails_all_reported(self):
report = (
"Identity: [email protected] author\n"
"Reviewer: [email protected]\n"
)
res = review_proofs.assess_email_disclosure(report)
self.assertTrue(res["flagged"])
self.assertEqual(
sorted(res["emails"]),
["[email protected]", "[email protected]"],
)
def test_justified_email_with_explanation_is_not_flagged(self):
report = "\n".join([
"Identity: jcwalker3 / prgs-author",
"Contact email: [email protected]",
"Email required because two accounts share the username and the",
"address is necessary to disambiguate identity.",
])
res = review_proofs.assess_email_disclosure(report)
self.assertFalse(res["flagged"])
self.assertTrue(res["justified"])
self.assertIn("[email protected]", res["emails"])
def test_email_without_justification_language_is_flagged(self):
report = "Contact email: [email protected] just in case.\n"
res = review_proofs.assess_email_disclosure(report)
self.assertTrue(res["flagged"])
self.assertFalse(res["justified"])
if __name__ == "__main__":
unittest.main()
tests/test_issue_lock_worktree.py
+27 -2
View File
@@ -36,7 +36,32 @@ class TestIssueLockWorktreeAssessment(unittest.TestCase):
porcelain_status="",
)
self.assertFalse(result["proven"])
self.assertIn("issue lock must be taken from base branch", result["reasons"][0])
self.assertIn("base-equivalence could not be proven", result["reasons"][0])
def test_base_equivalent_feature_branch_passes(self):
result = issue_lock_worktree.assess_issue_lock_worktree(
worktree_path="/repo/branches/issue-275",
current_branch="feat/issue-275-claim-lock-branches-worktree",
porcelain_status="",
base_equivalent=True,
inspected_git_root="/repo/branches/issue-275",
base_branch="origin/master",
)
self.assertTrue(result["proven"])
self.assertFalse(result["block"])
self.assertEqual(result["base_branch"], "origin/master")
def test_non_base_equivalent_branch_fails(self):
result = issue_lock_worktree.assess_issue_lock_worktree(
worktree_path="/repo/branches/issue-275",
current_branch="feat/issue-275-claim-lock-branches-worktree",
porcelain_status="",
base_equivalent=False,
inspected_git_root="/repo/branches/issue-275",
base_branch=None,
)
self.assertFalse(result["proven"])
self.assertIn("must be base-equivalent", result["reasons"][0])
def test_untracked_files_do_not_block(self):
result = issue_lock_worktree.assess_issue_lock_worktree(
@@ -96,4 +121,4 @@ class TestPrWorktreeMatch(unittest.TestCase):
if __name__ == "__main__":
unittest.main()
unittest.main()
tests/test_llm_workflow_split.py
+100
View File
@@ -0,0 +1,100 @@
"""Doc-contract checks for task-specific LLM workflow split (#333)."""
from pathlib import Path
REPO_ROOT = Path(__file__).resolve().parent.parent
SKILL_DIR = REPO_ROOT / "skills" / "llm-project-workflow"
SKILL = (SKILL_DIR / "SKILL.md").read_text(encoding="utf-8")
def test_skill_md_exists():
assert SKILL_DIR.joinpath("SKILL.md").is_file()
def test_all_workflow_files_exist():
for name in (
"review-merge-pr.md",
"reconcile-landed-pr.md",
"create-issue.md",
"work-issue.md",
):
assert (SKILL_DIR / "workflows" / name).is_file(), name
def test_skill_references_all_workflow_files():
for name in (
"workflows/review-merge-pr.md",
"workflows/reconcile-landed-pr.md",
"workflows/create-issue.md",
"workflows/work-issue.md",
):
assert name in SKILL
def test_skill_contains_mode_isolation_language():
assert "## Mode isolation" in SKILL
assert "review-merge-pr" in SKILL
assert "reconcile-landed-pr" in SKILL
assert "create-issue" in SKILL
assert "work-issue" in SKILL
assert "Do not mix modes" in SKILL or "do not mix modes" in SKILL.lower()
def test_skill_is_router_not_monolithic_review_body():
assert "This skill is a **router**" in SKILL or "router" in SKILL.lower()
assert "## F. Review workflow" not in SKILL
assert "gitea_mark_final_review_decision" not in SKILL
assert "gitea_submit_pr_review" not in SKILL
def test_review_merge_workflow_contract():
text = (SKILL_DIR / "workflows" / "review-merge-pr.md").read_text(encoding="utf-8")
assert "canonical: true" in text
assert "## 26A. Terminal review mutation hard-stop" in text
assert "## 11A. Skipped PRs are read-only" in text
assert "## 35. Duplicate request-changes prevention" in text
assert "ALREADY_LANDED_RECONCILE_REQUIRED" in text
def test_reconcile_landed_workflow_contract():
text = (SKILL_DIR / "workflows" / "reconcile-landed-pr.md").read_text(encoding="utf-8")
assert "canonical: true" in text
assert "Do not review or merge normal PRs" in text
assert "Already-landed proof" in text
def test_create_issue_workflow_contract():
text = (SKILL_DIR / "workflows" / "create-issue.md").read_text(encoding="utf-8")
assert "canonical: true" in text
assert "## 9. Duplicate search before mutation" in text
def test_work_issue_workflow_contract():
text = (SKILL_DIR / "workflows" / "work-issue.md").read_text(encoding="utf-8")
assert "canonical: true" in text
assert "Do not merge your own PR" in text
assert "## 21. PR creation rules" in text
def test_final_report_schemas_exist():
for name in (
"review-merge-final-report.md",
"reconcile-landed-final-report.md",
"create-issue-final-report.md",
"work-issue-final-report.md",
):
assert (SKILL_DIR / "schemas" / name).is_file(), name
def test_review_pr_template_references_workflow():
text = (SKILL_DIR / "templates" / "review-pr.md").read_text(encoding="utf-8")
assert "workflows/review-merge-pr.md" in text
def test_merge_pr_template_references_workflow():
text = (SKILL_DIR / "templates" / "merge-pr.md").read_text(encoding="utf-8")
assert "workflows/review-merge-pr.md" in text
def test_start_issue_template_references_work_issue_workflow():
text = (SKILL_DIR / "templates" / "start-issue.md").read_text(encoding="utf-8")
assert "workflows/work-issue.md" in text
tests/test_mcp_server.py
+71 -15
View File
@@ -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")
@@ -2931,20 +2937,31 @@ class TestVerifyMutationAuthority(unittest.TestCase):
mcp_server.verify_mutation_authority("prgs")
def _clean_master_git_state_for_lock():
return {
"current_branch": "master",
"porcelain_status": "",
"base_equivalent": True,
"inspected_git_root": "/scratch/wt",
"base_branch": "origin/master",
}
class TestIssueLocking(unittest.TestCase):
"""Test issue locking and PR gating constraints."""
@staticmethod
def _clean_master_git_state():
return {"current_branch": "master", "porcelain_status": ""}
def setUp(self):
self._env_patcher = patch.dict(os.environ, ISSUE_WRITE_ENV, clear=True)
self._env_patcher.start()
def tearDown(self):
self._env_patcher.stop()
if os.path.exists(ISSUE_LOCK_FILE):
os.remove(ISSUE_LOCK_FILE)
@patch(
"mcp_server.issue_lock_worktree.read_worktree_git_state",
return_value={"current_branch": "master", "porcelain_status": ""},
return_value=_clean_master_git_state_for_lock(),
)
@patch("mcp_server.api_get_all")
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
@@ -2964,7 +2981,7 @@ class TestIssueLocking(unittest.TestCase):
@patch(
"mcp_server.issue_lock_worktree.read_worktree_git_state",
return_value={"current_branch": "master", "porcelain_status": ""},
return_value=_clean_master_git_state_for_lock(),
)
@patch("mcp_server.api_get_all")
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
@@ -2981,7 +2998,7 @@ class TestIssueLocking(unittest.TestCase):
@patch(
"mcp_server.issue_lock_worktree.read_worktree_git_state",
return_value={"current_branch": "master", "porcelain_status": ""},
return_value=_clean_master_git_state_for_lock(),
)
@patch("mcp_server.api_get_all")
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
@@ -3002,7 +3019,7 @@ class TestIssueLocking(unittest.TestCase):
scratch = "/tmp/gitea-tools-author-scratch/issue-249-clean"
with patch(
"mcp_server.issue_lock_worktree.read_worktree_git_state",
return_value={"current_branch": "master", "porcelain_status": ""},
return_value=_clean_master_git_state_for_lock(),
) as mock_git:
res = gitea_lock_issue(
issue_number=249,
@@ -3022,6 +3039,7 @@ class TestIssueLocking(unittest.TestCase):
return_value={
"current_branch": "master",
"porcelain_status": " M gitea_mcp_server.py\n",
"base_equivalent": True,
},
):
with self.assertRaises(RuntimeError) as ctx:
@@ -3041,6 +3059,7 @@ class TestIssueLocking(unittest.TestCase):
return_value={
"current_branch": "feat/issue-243-forbidden-git-gaps",
"porcelain_status": "",
"base_equivalent": False,
},
):
with self.assertRaises(RuntimeError) as ctx:
@@ -3050,7 +3069,7 @@ class TestIssueLocking(unittest.TestCase):
remote="prgs",
worktree_path="/tmp/scratch/wt",
)
self.assertIn("issue lock must be taken from base branch", str(ctx.exception))
self.assertIn("base-equivalent", str(ctx.exception))
@patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
return_value=(True, []))
@@ -3193,7 +3212,12 @@ class TestPreflightVerification(unittest.TestCase):
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
for key in ("GITEA_TEST_FORCE_DIRTY", "GITEA_TEST_PORCELAIN"):
for key in (
"GITEA_TEST_FORCE_DIRTY",
"GITEA_TEST_PORCELAIN",
"GITEA_ACTIVE_WORKTREE",
"GITEA_AUTHOR_WORKTREE",
):
if key in os.environ:
del os.environ[key]
@@ -3283,3 +3307,35 @@ class TestPreflightVerification(unittest.TestCase):
status = mcp_server.assess_preflight_status()
self.assertFalse(status["preflight_ready"])
self.assertIn("gitea_whoami", status["preflight_block_reasons"][0])
def test_declared_clean_task_worktree_ignores_control_checkout_violation(self):
"""#275: active branches/ worktree is the inspected mutation workspace."""
import mcp_server
mcp_server._preflight_whoami_called = True
mcp_server._preflight_capability_called = True
mcp_server._preflight_whoami_violation = True
mcp_server._preflight_whoami_violation_files = ["mcp_server.py"]
os.environ["GITEA_TEST_PORCELAIN"] = ""
worktree = "/repo/branches/issue-275-clean"
status = mcp_server.assess_preflight_status(worktree_path=worktree)
self.assertTrue(status["preflight_ready"])
self.assertEqual(status["preflight_block_reasons"], [])
self.assertIn("preflight_workspace", status)
mcp_server.verify_preflight_purity(worktree_path=worktree)
def test_declared_dirty_task_worktree_reports_workspace_diagnostics(self):
"""#275: dirty failures name the inspected task workspace, not just files."""
import mcp_server
mcp_server._preflight_whoami_called = True
mcp_server._preflight_capability_called = True
os.environ["GITEA_TEST_PORCELAIN"] = " M task_file.py\n"
worktree = "/repo/branches/issue-275-dirty"
with self.assertRaises(RuntimeError) as ctx:
mcp_server.verify_preflight_purity(worktree_path=worktree)
msg = str(ctx.exception)
self.assertIn("active task workspace root", msg)
self.assertIn("inspected git root", msg)
self.assertIn("dirty files: task_file.py", msg)
self.assertIn("dirty scope:", msg)
tests/test_mutation_ledger_report.py
+102
View File
@@ -0,0 +1,102 @@
"""Tests for final-report mutation ledger verifier (#331)."""
import sys
import unittest
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from review_proofs import assess_mutation_ledger_report # noqa: E402
class TestMutationLedgerReport(unittest.TestCase):
def test_none_claim_with_performed_edit_blocks(self):
report = (
"Review decision: request_changes.\n"
"File edits by reviewer: none\n"
)
action_log = [
{"action": "Edited", "path": "walkthrough.md", "tracked": False},
]
result = assess_mutation_ledger_report(report, action_log=action_log)
self.assertFalse(result["proven"])
self.assertTrue(result["file_edits_claimed_none"])
def test_reported_edit_passes(self):
report = (
"File edits by reviewer: Edited walkthrough.md (untracked)\n"
"Mutation ledger: Edited walkthrough.md untracked in review worktree\n"
)
action_log = [
{
"action": "Edited",
"path": "walkthrough.md",
"tracked": False,
"in_repo": True,
},
]
result = assess_mutation_ledger_report(
report,
action_log=action_log,
walkthrough_explicitly_requested=True,
)
self.assertTrue(result["proven"])
def test_unreported_mutation_blocks(self):
report = "File edits by reviewer: none\n"
action_log = [{"action": "Wrote", "path": "/tmp/review-notes.txt"}]
result = assess_mutation_ledger_report(report, action_log=action_log)
self.assertFalse(result["proven"])
self.assertIn("/tmp/review-notes.txt", result["unreported_paths"])
def test_outside_repo_requires_label(self):
report = (
"File edits by reviewer: Wrote /tmp/review-notes.txt\n"
)
action_log = [
{
"action": "Wrote",
"path": "/tmp/review-notes.txt",
"outside_repo": True,
},
]
result = assess_mutation_ledger_report(report, action_log=action_log)
self.assertFalse(result["proven"])
self.assertIn("outside repo", " ".join(result["reasons"]).lower())
def test_gated_rejection_excluded_from_performed(self):
report = "File edits by reviewer: none\nRejected gated calls: submit_pr_review blocked\n"
action_log = [
{
"action": "Edited",
"path": "walkthrough.md",
"performed": False,
"gated_rejected": True,
},
]
result = assess_mutation_ledger_report(report, action_log=action_log)
self.assertTrue(result["proven"])
self.assertEqual(result["performed_mutations"], [])
def test_post_status_artifact_requires_final_git_status(self):
report = (
"File edits by reviewer: Created scratch/notes.md (untracked)\n"
)
action_log = [
{
"action": "Created",
"path": "scratch/notes.md",
"tracked": False,
"after_git_status": True,
},
]
result = assess_mutation_ledger_report(
report,
action_log=action_log,
final_git_status_reported=False,
)
self.assertFalse(result["proven"])
self.assertIn("final git status", " ".join(result["reasons"]).lower())
if __name__ == "__main__":
unittest.main()
tests/test_operator_guide.py
+2 -1
View File
@@ -134,7 +134,8 @@ class TestControlPlaneGuide(GuideTestBase):
"merge_confirmation", "redaction", "separation",
"profile_switching", "identity_verification",
"work_selection", "global_worktree",
"subagent_tool_budget"):
"shell_spawn_hard_stop", "subagent_tool_budget",
"subagent_delegation"):
self.assertIn(key, rules)
self.assertIn("MERGE PR", json.dumps(rules["merge_confirmation"]))
self.assertTrue(rules["hard_stops"])
tests/test_request_changes_approval_proof.py
+104
View File
@@ -0,0 +1,104 @@
"""Tests for REQUEST_CHANGES override proof before approval (#326)."""
import sys
import unittest
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from review_proofs import assess_request_changes_approval_proof # noqa: E402
HEAD_A = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
HEAD_B = "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"
def _feedback(**overrides):
base = {
"success": True,
"current_head_sha": HEAD_A,
"has_blocking_change_requests": False,
"author_pushed_after_request_changes": False,
"reviews": [],
}
base.update(overrides)
return base
def _blocking_review(**overrides):
entry = {
"reviewer": "reviewer-a",
"verdict": "REQUEST_CHANGES",
"body": "Tests fail on pinned head; fix test_commit_files_gate.",
"submitted_at": "2026-07-07T01:00:00-05:00",
"reviewed_head_sha": HEAD_A,
"dismissed": False,
"stale": False,
}
entry.update(overrides)
return entry
class TestRequestChangesApprovalProof(unittest.TestCase):
def test_missing_feedback_blocks_approval(self):
result = assess_request_changes_approval_proof(None)
self.assertFalse(result["approve_allowed"])
self.assertTrue(result["block"])
def test_no_prior_request_changes_allows_approval(self):
feedback = _feedback(
reviews=[{
"reviewer": "sysadmin",
"verdict": "APPROVED",
"body": "LGTM",
"submitted_at": "2026-07-07T02:00:00-05:00",
"reviewed_head_sha": HEAD_A,
"dismissed": False,
}]
)
result = assess_request_changes_approval_proof(feedback)
self.assertTrue(result["approve_allowed"])
self.assertIsNone(result["blocking_review"])
def test_changed_head_since_blocker_allows_approval(self):
feedback = _feedback(
current_head_sha=HEAD_B,
author_pushed_after_request_changes=True,
has_blocking_change_requests=True,
reviews=[_blocking_review(reviewed_head_sha=HEAD_A)],
)
result = assess_request_changes_approval_proof(feedback)
self.assertTrue(result["approve_allowed"])
self.assertTrue(result["head_changed_since_blocker"])
def test_unchanged_head_without_override_blocks_approval(self):
feedback = _feedback(
has_blocking_change_requests=True,
reviews=[_blocking_review()],
)
result = assess_request_changes_approval_proof(feedback)
self.assertFalse(result["approve_allowed"])
self.assertIn("override_reason", " ".join(result["reasons"]))
def test_unchanged_head_with_valid_override_allows_approval(self):
blocker = _blocking_review()
feedback = _feedback(
has_blocking_change_requests=True,
reviews=[blocker],
)
report = (
"Blocker text: Tests fail on pinned head; fix test_commit_files_gate.\n"
"Override: wrong_validation_environment — CI used stale worktree."
)
result = assess_request_changes_approval_proof(
feedback,
override_reason="wrong_validation_environment",
override_explanation="CI used stale worktree; local rerun passed.",
report_text=report,
)
self.assertTrue(result["approve_allowed"])
self.assertEqual(
result["blocking_review"]["blocking_reviewer"], "reviewer-a"
)
if __name__ == "__main__":
unittest.main()
tests/test_resolve_task_capability.py
+11
View File
@@ -141,6 +141,17 @@ class TestResolveTaskCapability(unittest.TestCase):
self.assertTrue(res["allowed_in_current_session"])
self.assertIn("gitea.issue.comment", res["active_profile_allowed_operations"])
@patch("mcp_server.api_request", return_value={"login": "author-user"})
@patch("mcp_server.get_auth_header", return_value="token author-pass")
def test_resolve_lock_issue_author_profile_allowed(self, _auth, _api):
# #275: lock_issue must be an exact resolver task for claim/lock gates.
with patch.dict(os.environ, self._env("author-profile")):
res = mcp_server.gitea_resolve_task_capability(task="lock_issue", remote="prgs")
self.assertEqual(res["requested_task"], "lock_issue")
self.assertEqual(res["required_operation_permission"], "gitea.issue.comment")
self.assertTrue(res["allowed_in_current_session"])
self.assertIn("gitea.issue.comment", res["active_profile_allowed_operations"])
def test_resolve_unknown_task_fails_closed(self):
with patch.dict(os.environ, self._env("author-profile")):
with self.assertRaises(ValueError):
tests/test_role_session_router.py
+25 -18
View File
@@ -206,28 +206,35 @@ class TestRoleSessionRouter(unittest.TestCase):
class TestCheckMidMerge(unittest.TestCase):
def test_skip_scan_walk_root_skips_sibling_worktrees_only(self):
worktree_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
main_root = os.path.dirname(os.path.dirname(worktree_root))
self.assertFalse(
role_session_router.skip_python_scan_walk_root(
main_root, main_root
# Hermetic <main>/branches/<worktree> layout (#283): deriving these
# paths from __file__ made the test pass only when the suite ran from
# a branches/ worktree, because skip_python_scan_walk_root skips a
# branches/ walk root only when <project_root>/branches exists.
with tempfile.TemporaryDirectory() as tmp:
main_root = os.path.join(tmp, "main")
worktree_root = os.path.join(main_root, "branches", "fix-issue-1")
os.makedirs(os.path.join(worktree_root, "tests"))
self.assertFalse(
role_session_router.skip_python_scan_walk_root(
main_root, main_root
)
)
)
self.assertTrue(
role_session_router.skip_python_scan_walk_root(
main_root, os.path.join(main_root, "branches", "fix-issue-1")
self.assertTrue(
role_session_router.skip_python_scan_walk_root(
main_root, os.path.join(main_root, "branches", "fix-issue-1")
)
)
)
self.assertFalse(
role_session_router.skip_python_scan_walk_root(
worktree_root, worktree_root
self.assertFalse(
role_session_router.skip_python_scan_walk_root(
worktree_root, worktree_root
)
)
)
self.assertFalse(
role_session_router.skip_python_scan_walk_root(
worktree_root, os.path.join(worktree_root, "tests")
self.assertFalse(
role_session_router.skip_python_scan_walk_root(
worktree_root, os.path.join(worktree_root, "tests")
)
)
)
def test_decorative_equals_banner_is_not_mid_merge(self):
self.assertFalse(role_session_router.check_mid_merge())
tests/test_shell_spawn_hard_stop_docs.py
+82
View File
@@ -0,0 +1,82 @@
"""Doc-contract checks for the shell-spawn hard-stop rule (#258).
When the shell executor fails to spawn (exit_code: -1 with empty
stdout/stderr), agents must probe once, mark shell unavailable, hard-stop
after two consecutive spawn failures, and emit a recovery report instead of
retry-spiralling. These tests pin that guidance in the runbooks, the portable
skill doc, and the operator guide so it cannot silently regress.
"""
from pathlib import Path
REPO_ROOT = Path(__file__).resolve().parent.parent
RUNBOOK = REPO_ROOT / "docs" / "llm-workflow-runbooks.md"
SKILL = REPO_ROOT / "skills" / "llm-project-workflow" / "SKILL.md"
def _normalize(text: str) -> str:
"""Collapse whitespace so phrase checks survive markdown line wrapping."""
return " ".join(text.split())
def _runbook_text() -> str:
return _normalize(RUNBOOK.read_text(encoding="utf-8"))
def _skill_text() -> str:
return _normalize(SKILL.read_text(encoding="utf-8"))
def test_runbook_has_shell_spawn_hard_stop_section():
text = _runbook_text()
assert "## Shell Spawn Hard-Stop Rule" in text
for phrase in (
"exit_code: -1",
"empty stdout/stderr",
"trivial probe",
"two consecutive spawn failures",
"mark shell unavailable",
"recovery report",
):
assert phrase in text, f"runbook missing phrase: {phrase!r}"
def test_runbook_recovery_report_names_required_steps():
text = _runbook_text()
for phrase in (
"restart the session",
"kill hung background terminals",
"MCP-native",
):
assert phrase in text, f"runbook recovery guidance missing: {phrase!r}"
def test_runbook_forbids_retry_spirals():
text = _runbook_text()
assert "never retry the same failing spawn" in text
def test_skill_doc_declares_shell_spawn_hard_stop_rule():
text = _skill_text()
assert "## Shell Spawn Hard-Stop Rule" in text
for phrase in (
"exit_code: -1",
"two consecutive spawn failures",
"recovery report",
):
assert phrase in text, f"SKILL.md missing phrase: {phrase!r}"
def test_operator_guide_rules_include_shell_spawn_hard_stop():
import sys
sys.path.insert(0, str(REPO_ROOT))
import gitea_mcp_server
rule = gitea_mcp_server._GUIDE_RULES["shell_spawn_hard_stop"]
for phrase in (
"exit_code: -1",
"two consecutive",
"recovery report",
):
assert phrase in rule, f"operator guide rule missing: {phrase!r}"
tests/test_subagent_gate.py
+152
View File
@@ -0,0 +1,152 @@
"""Tests for fail-closed subagent delegation gates (#266).
Covers the acceptance criteria: blocked write delegation, allowed read-only
delegation, missing inherited context, and invalid subagent final reports.
"""
import unittest
import subagent_gate
def _full_context():
return {
"issue_lock": "issue #266 locked to feat/issue-266-subagent-gate-inheritance",
"branch": "feat/issue-266-subagent-gate-inheritance",
"worktree_path": "branches/feat-issue-266-subagent-gate-inheritance",
"identity_profile": "jcwalker3/prgs-author",
"allowed_tool_class": "read_write_files",
"command_deny_list": "git push --force; rm -rf",
"validation_ledger_requirement": "record command/exit/output for every validation claim",
"final_report_schema": "controller-handoff-v1",
}
def _full_report():
return {
"identity_profile": "jcwalker3/prgs-author",
"worktree_path": "branches/feat-issue-266-subagent-gate-inheritance",
"branch": "feat/issue-266-subagent-gate-inheritance",
"changed_files": "subagent_gate.py, tests/test_subagent_gate.py",
"validation_results": "pytest tests/test_subagent_gate.py -q: exit 0, 14 passed",
"workspace_mutations": "edited subagent_gate.py",
}
class TestAssessSubagentDelegation(unittest.TestCase):
# AC1: deterministic write tasks are blocked by default
def test_write_delegation_blocked_by_default(self):
for task in ("claim_issue", "create_branch", "edit_code", "commit",
"push_branch", "create_pr", "review_pr", "merge_pr",
"cleanup_branch"):
res = subagent_gate.assess_subagent_delegation(task)
self.assertTrue(res["block"], task)
self.assertFalse(res["allowed"], task)
self.assertEqual(res["task_class"], "deterministic_write", task)
self.assertTrue(
any("explicitly allowed" in r for r in res["reasons"]), task)
# AC2: explicit allowance without a recorded justification still blocks
def test_write_delegation_requires_recorded_justification(self):
res = subagent_gate.assess_subagent_delegation(
"create_pr", explicitly_allowed=True,
inherited_context=_full_context())
self.assertTrue(res["block"])
self.assertTrue(
any("justification" in r for r in res["reasons"]))
# AC3: explicit allowance + justification but missing inherited context
def test_write_delegation_blocks_on_missing_inherited_context(self):
context = _full_context()
del context["issue_lock"]
del context["command_deny_list"]
res = subagent_gate.assess_subagent_delegation(
"commit", explicitly_allowed=True,
justification="parent session proved batch commit needs isolation",
inherited_context=context)
self.assertTrue(res["block"])
self.assertIn("issue_lock", res["missing_context"])
self.assertIn("command_deny_list", res["missing_context"])
def test_write_delegation_blocks_on_empty_context_values(self):
context = _full_context()
context["worktree_path"] = " "
res = subagent_gate.assess_subagent_delegation(
"commit", explicitly_allowed=True,
justification="isolation required",
inherited_context=context)
self.assertTrue(res["block"])
self.assertIn("worktree_path", res["missing_context"])
def test_write_delegation_allowed_with_authorization_and_full_context(self):
res = subagent_gate.assess_subagent_delegation(
"edit_code", explicitly_allowed=True,
justification="parallel mechanical rename across many files",
inherited_context=_full_context())
self.assertFalse(res["block"])
self.assertTrue(res["allowed"])
self.assertEqual(res["missing_context"], [])
# Allowed read-only delegation needs no explicit authorization
def test_read_only_delegation_allowed(self):
for task in ("read_files", "code_search", "inventory_prs",
"summarize_issue"):
res = subagent_gate.assess_subagent_delegation(task)
self.assertFalse(res["block"], task)
self.assertTrue(res["allowed"], task)
self.assertEqual(res["task_class"], "read_only", task)
# Unknown task types fail closed
def test_unknown_task_fails_closed(self):
res = subagent_gate.assess_subagent_delegation("launch_missiles")
self.assertTrue(res["block"])
self.assertEqual(res["task_class"], "unknown")
def test_blank_task_fails_closed(self):
res = subagent_gate.assess_subagent_delegation(" ")
self.assertTrue(res["block"])
self.assertEqual(res["task_class"], "unknown")
class TestValidateSubagentReport(unittest.TestCase):
# AC4: subagent output must carry the same proof fields as the parent
def test_full_report_valid(self):
res = subagent_gate.validate_subagent_report(_full_report())
self.assertTrue(res["valid"])
self.assertFalse(res["block"])
self.assertEqual(res["missing_fields"], [])
def test_missing_proof_fields_invalid(self):
report = _full_report()
del report["validation_results"]
del report["worktree_path"]
res = subagent_gate.validate_subagent_report(report)
self.assertFalse(res["valid"])
self.assertTrue(res["block"])
self.assertIn("validation_results", res["missing_fields"])
self.assertIn("worktree_path", res["missing_fields"])
def test_empty_field_values_invalid(self):
report = _full_report()
report["changed_files"] = ""
res = subagent_gate.validate_subagent_report(report)
self.assertFalse(res["valid"])
self.assertIn("changed_files", res["missing_fields"])
def test_none_report_invalid(self):
res = subagent_gate.validate_subagent_report(None)
self.assertFalse(res["valid"])
self.assertTrue(res["block"])
class TestOperatorGuideRule(unittest.TestCase):
def test_operator_guide_declares_subagent_rule(self):
import gitea_mcp_server
rule = gitea_mcp_server._GUIDE_RULES["subagent_delegation"].lower()
for phrase in ("deterministic write", "inherit", "read-only",
"fail closed"):
self.assertIn(phrase, rule)
if __name__ == "__main__":
unittest.main()
tests/test_validation_ledger.py
+155
View File
@@ -0,0 +1,155 @@
"""Tests for validation ledger and final-report verifier (#271)."""
import sys
import unittest
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from validation_ledger import ( # noqa: E402
assess_full_suite_claim,
assess_ledger_entry,
new_ledger_entry,
verify_final_report,
)
class TestLedgerEntry(unittest.TestCase):
def test_valid_entry_is_claimable(self):
entry = new_ledger_entry(
command="pytest tests/ -q",
cwd="/repo/branches/feat-issue-271",
exit_code=0,
output_summary="994 passed, 6 skipped",
)
result = assess_ledger_entry(entry)
self.assertTrue(result["claimable"])
self.assertEqual(result["verdict"], "strong")
def test_missing_command_output_is_invalid(self):
entry = new_ledger_entry(
command="pytest tests/ -q",
cwd="/repo/branches/task",
exit_code=0,
output_summary="",
)
result = assess_ledger_entry(entry)
self.assertFalse(result["claimable"])
self.assertEqual(result["verdict"], "invalid")
class TestFullSuiteClaim(unittest.TestCase):
def _entry(self, **kwargs):
base = new_ledger_entry(
command="pytest tests/ -q",
cwd="/repo/branches/task",
exit_code=0,
output_summary="10 passed",
)
base.update(kwargs)
return base
def test_full_suite_overclaim_with_ignored_paths_fails(self):
entry = self._entry(
ignored_paths=[{"path": "tests/integration/", "justification": ""}],
)
result = assess_full_suite_claim(
[entry],
report_text="Validation: full suite passed.",
)
self.assertFalse(result["claimable"])
self.assertTrue(
any("full suite" in r.lower() for r in result["reasons"])
)
def test_full_suite_except_note_allows_ignored_paths(self):
entry = self._entry(
ignored_paths=[{
"path": "tests/integration/",
"justification": "network tests require GITEA_INTEGRATION=1",
}],
)
result = assess_full_suite_claim(
[entry],
report_text="Validation: full suite except integration tests.",
)
self.assertTrue(result["claimable"])
class TestFinalReportVerifier(unittest.TestCase):
def _base_report(self, **overrides):
report = {
"identity": "jcwalker3",
"profile": "prgs-author",
"capability_resolved": True,
"worktree_path": "/repo/branches/feat-issue-271",
"changed_files": ["validation_ledger.py"],
"validation_ledger": [
new_ledger_entry(
command="pytest tests/test_validation_ledger.py -q",
cwd="/repo/branches/feat-issue-271",
exit_code=0,
output_summary="12 passed",
)
],
"validation_claim": "passed",
"review_submitted": False,
"review_visible_approved": False,
"review_pending": False,
"merge_performed": False,
"merge_blocker": "not a reviewer task",
"workspace_clean": True,
}
report.update(overrides)
return report
def test_complete_report_is_proven(self):
result = verify_final_report(self._base_report())
self.assertTrue(result["proven"])
self.assertFalse(result["block"])
def test_validation_overclaim_fails(self):
report = self._base_report(
validation_claim="passed",
validation_ledger=[
new_ledger_entry(
command="pytest tests/ -q",
cwd="/repo/branches/task",
exit_code=1,
output_summary="3 failed",
)
],
)
result = verify_final_report(report)
self.assertFalse(result["proven"])
self.assertTrue(result["contradictions"])
def test_pending_approval_misreported_as_approved_fails(self):
report = self._base_report(
claims_approved_review=True,
review_visible_approved=False,
review_pending=True,
)
result = verify_final_report(report)
self.assertFalse(result["proven"])
self.assertTrue(
any("approved" in c.lower() for c in result["contradictions"])
)
def test_missing_command_output_in_ledger_fails(self):
report = self._base_report(
validation_ledger=[
new_ledger_entry(
command="pytest tests/ -q",
cwd="/repo/branches/task",
exit_code=0,
output_summary="",
)
],
)
result = verify_final_report(report)
self.assertFalse(result["proven"])
self.assertTrue(result["reasons"])
if __name__ == "__main__":
unittest.main()
tests/test_workflow_commit_path_docs.py
+33
View File
@@ -0,0 +1,33 @@
"""Documentation checks for MCP-native commit path rules (#260)."""
from pathlib import Path
REPO_ROOT = Path(__file__).resolve().parent.parent
RUNBOOK = REPO_ROOT / "docs" / "llm-workflow-runbooks.md"
SAFETY = REPO_ROOT / "docs" / "safety-model.md"
def test_runbook_requires_mcp_native_commit_path():
text = RUNBOOK.read_text(encoding="utf-8")
assert "MCP-native commit path" in text
assert "gitea_commit_files" in text
assert "gitea.repo.commit" in text
assert "only" in text.lower() and "approved" in text.lower()
lower = text.lower()
for forbidden in ("webfetch", "playwright", "manual llm-generated base64"):
assert forbidden in lower
def test_runbook_forbids_fallback_loops():
lower = RUNBOOK.read_text(encoding="utf-8").lower()
assert "retry shell encoding" in lower
assert "loop" in lower
assert "recovery report" in lower
def test_safety_model_documents_commit_fallback_ban():
text = SAFETY.read_text(encoding="utf-8")
assert "Agent Commit Path" in text
assert "gitea_commit_files" in text
assert "WebFetch" in text
assert "Playwright" in text
assert "llm-workflow-runbooks.md" in text