Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
 sysadmin
|
0f11be9569 |
@@ -10,8 +10,3 @@ gitea-mcp*.json
|
||||
.vscode/
|
||||
graphify-out/
|
||||
branches/
|
||||
# Throwaway agent payload/encoding helpers (#261): delete after the MCP
|
||||
# commit completes or aborts; ignored here so strays cannot block issue locks.
|
||||
_encode_*.py
|
||||
_emit_*.py
|
||||
_inline_*.py
|
||||
|
||||
@@ -459,33 +459,6 @@ touching anything.
|
||||
files, detected secret, or any production/deploy behavior — **stop, report the
|
||||
blocker, and take no mutating action.** Fail closed; never work around a gate.
|
||||
|
||||
## Agent temp artifact cleanup (#261)
|
||||
|
||||
Failed or aborted subagent runs have left throwaway helper scripts in the
|
||||
shared working tree — for example `_encode_commit_payload.py` and
|
||||
`_emit_payload.py` from the #152 commit-recovery attempt. These artifacts are
|
||||
never part of any issue scope. They pollute `git status`, which trips
|
||||
fail-closed gates: `gitea_lock_issue` refuses to lock over tracked edits, and
|
||||
the pre-flight purity checks attribute the dirt to the current session.
|
||||
|
||||
Rules:
|
||||
|
||||
- **Delete throwaway helpers as soon as the MCP commit completes or aborts.**
|
||||
Any file matching `_encode_*`, `_emit_*`, or `_inline_*` created as a
|
||||
scratch payload/encoding helper must be removed in the same session that
|
||||
created it — success and failure paths alike.
|
||||
- **Prefer the scratchpad.** Payload preparation belongs in the session
|
||||
scratchpad or a temp directory, never inside the repository worktree
|
||||
(see the shell-free commit payload work in #263).
|
||||
- **`.gitignore` backstop.** The patterns `_encode_*.py`, `_emit_*.py`, and
|
||||
`_inline_*.py` are ignored so a leftover helper cannot show up as an
|
||||
untracked file and block issue locks for later sessions. Ignoring them is a
|
||||
safety net, not permission to leave them behind.
|
||||
- **Cleanup checklist before ending any author session:** run
|
||||
`git status --porcelain` in the worktree you mutated; remove any
|
||||
`_encode_*` / `_emit_*` / `_inline_*` files you created; report the removal
|
||||
in the final run report like any other local mutation.
|
||||
|
||||
## Task/role alignment (#167)
|
||||
|
||||
The **requested task** decides what a session may do — not the credential it
|
||||
|
||||
@@ -2301,6 +2301,29 @@ def gitea_commit_files(
|
||||
Returns:
|
||||
dict with success status and commit/branch information.
|
||||
"""
|
||||
ok, block_reasons = role_session_router.check_author_mutation_after_reviewer_stop(
|
||||
"commit_files"
|
||||
)
|
||||
if not ok:
|
||||
return {
|
||||
"success": False,
|
||||
"performed": False,
|
||||
"commit": "",
|
||||
"branch": "",
|
||||
"reasons": block_reasons,
|
||||
}
|
||||
blocked = _namespace_mutation_block(
|
||||
"commit_files", commit="", branch="", remote=remote
|
||||
)
|
||||
if blocked:
|
||||
return blocked
|
||||
blocked = _profile_permission_block(
|
||||
task_capability_map.required_permission("commit_files"),
|
||||
commit="", branch="",
|
||||
)
|
||||
if blocked:
|
||||
return blocked
|
||||
|
||||
verify_preflight_purity(remote)
|
||||
h, o, r = _resolve(remote, host, org, repo)
|
||||
auth = _auth(h)
|
||||
|
||||
@@ -1297,16 +1297,6 @@ HANDOFF_ROLE_FIELDS = {
|
||||
("Session authored PR", ("session authored pr", "authored pr")),
|
||||
("Why continuation allowed", ("why continuation", "continuation allowed")),
|
||||
),
|
||||
"issue_filing": (
|
||||
("Issue created or updated", (
|
||||
"issue created or updated",
|
||||
"issue created",
|
||||
"issue updated",
|
||||
"created issue",
|
||||
"updated issue",
|
||||
)),
|
||||
("Related issues", ("related issues",)),
|
||||
),
|
||||
}
|
||||
|
||||
|
||||
@@ -2145,309 +2135,6 @@ def assess_issue_selection_final_report(
|
||||
}
|
||||
|
||||
|
||||
_SHA_EVIDENCE_LABEL = re.compile(
|
||||
r"(?:\b(?:old|new)\s+head\b|\bhead\s+sha\b|\bcommit\s+sha\b|\bsha\b)"
|
||||
r"\s*[:=]?\s*([0-9a-f]+)",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
|
||||
|
||||
def assess_issue_filing_sha_evidence(report_text: str) -> dict:
|
||||
"""Issue #191: commit evidence must use full 40-hex SHAs."""
|
||||
text = report_text or ""
|
||||
reasons = []
|
||||
for match in _SHA_EVIDENCE_LABEL.finditer(text):
|
||||
sha = match.group(1).strip().lower()
|
||||
if sha and not _FULL_SHA.match(sha):
|
||||
reasons.append(
|
||||
f"abbreviated SHA in evidence ({len(sha)} hex chars); "
|
||||
"full 40-character SHA required"
|
||||
)
|
||||
return {
|
||||
"complete": not reasons,
|
||||
"downgraded": bool(reasons),
|
||||
"reasons": reasons,
|
||||
}
|
||||
|
||||
|
||||
def assess_issue_filing_issue_reference(
|
||||
report_text: str,
|
||||
*,
|
||||
issue_number: int | None = None,
|
||||
issue_title: str | None = None,
|
||||
action: str = "created",
|
||||
) -> dict:
|
||||
"""Issue #191: reports must cite exact issue number and title."""
|
||||
text = (report_text or "").lower()
|
||||
reasons = []
|
||||
if issue_number is None:
|
||||
reasons.append("issue number proof missing from assessment context")
|
||||
else:
|
||||
num_patterns = (f"#{issue_number}", f"issue #{issue_number}",
|
||||
f"issue {issue_number}")
|
||||
if not any(p in text for p in num_patterns):
|
||||
reasons.append(
|
||||
f"report missing exact issue number #{issue_number}"
|
||||
)
|
||||
if issue_title:
|
||||
title_fragment = issue_title.strip().lower()[:40]
|
||||
if title_fragment and title_fragment not in text:
|
||||
reasons.append("report missing exact issue title")
|
||||
action = (action or "created").strip().lower()
|
||||
if action == "created" and "creat" not in text:
|
||||
reasons.append(
|
||||
"report must distinguish new issue creation from update"
|
||||
)
|
||||
if action == "updated" and "updat" not in text:
|
||||
reasons.append(
|
||||
"report must distinguish issue update from new creation"
|
||||
)
|
||||
return {
|
||||
"complete": not reasons,
|
||||
"downgraded": bool(reasons),
|
||||
"reasons": reasons,
|
||||
}
|
||||
|
||||
|
||||
def assess_issue_filing_mutation_capability(
|
||||
report_text: str,
|
||||
mutations: list[str] | None,
|
||||
resolved_capabilities: dict | None = None,
|
||||
) -> dict:
|
||||
"""Issue #191: every mutation needs exact capability proof in the report."""
|
||||
import task_capability_map
|
||||
|
||||
text = (report_text or "").lower()
|
||||
reasons = []
|
||||
mutation_list = list(mutations or [])
|
||||
if not mutation_list:
|
||||
reasons.append("no mutations listed for capability proof")
|
||||
cap_proof = assess_capability_proof(resolved_capabilities or {})
|
||||
if not cap_proof.get("proven"):
|
||||
reasons.extend(cap_proof.get("reasons") or [])
|
||||
|
||||
for task in mutation_list:
|
||||
try:
|
||||
permission = task_capability_map.required_permission(task)
|
||||
except KeyError:
|
||||
reasons.append(f"unknown mutation task '{task}'")
|
||||
continue
|
||||
if permission.lower() not in text and task.replace("_", " ") not in text:
|
||||
reasons.append(
|
||||
f"mutation '{task}' missing exact capability proof "
|
||||
f"('{permission}')"
|
||||
)
|
||||
if task == "set_issue_labels":
|
||||
label_proof = (
|
||||
"label change",
|
||||
"labels changed",
|
||||
"set label",
|
||||
"label mutation",
|
||||
"issue labels",
|
||||
"label:",
|
||||
)
|
||||
if not any(p in text for p in label_proof):
|
||||
reasons.append(
|
||||
"label mutation missing label-specific capability proof"
|
||||
)
|
||||
return {
|
||||
"complete": not reasons,
|
||||
"downgraded": bool(reasons),
|
||||
"reasons": reasons,
|
||||
}
|
||||
|
||||
|
||||
def assess_issue_filing_mutation_scope(
|
||||
report_text: str,
|
||||
*,
|
||||
performed_mutations: list[str] | None = None,
|
||||
forbidden_mutations: list[str] | None = None,
|
||||
) -> dict:
|
||||
"""Issue #191: single-mutation runs must state scope and absent mutations."""
|
||||
text = (report_text or "").lower()
|
||||
performed = list(performed_mutations or [])
|
||||
forbidden = list(forbidden_mutations or [
|
||||
"label", "comment", "pr", "review", "merge", "close",
|
||||
])
|
||||
reasons = []
|
||||
|
||||
if len(performed) == 1:
|
||||
only_phrases = ("only mutation", "only mutations", "sole mutation")
|
||||
if not any(p in text for p in only_phrases):
|
||||
reasons.append(
|
||||
"single-mutation run must state 'Only mutation(s): ...'"
|
||||
)
|
||||
for absent in forbidden:
|
||||
if absent == "comment" and performed[0] in (
|
||||
"comment_issue", "create_issue",
|
||||
):
|
||||
continue
|
||||
if absent not in text:
|
||||
continue
|
||||
denial_phrases = (
|
||||
f"no {absent}",
|
||||
f"no {absent}s",
|
||||
f"without {absent}",
|
||||
"none performed",
|
||||
"none.",
|
||||
"confirm no",
|
||||
)
|
||||
if any(phrase in text for phrase in denial_phrases):
|
||||
continue
|
||||
reasons.append(
|
||||
f"single-mutation run mentions '{absent}' without "
|
||||
f"confirming it was not performed"
|
||||
)
|
||||
if performed == ["create_issue"]:
|
||||
for check in ("label", "comment", "pr", "review", "merge"):
|
||||
if check in text and f"no {check}" not in text:
|
||||
if not any(
|
||||
n in text
|
||||
for n in (
|
||||
f"no {check}",
|
||||
f"no {check}s",
|
||||
"none performed",
|
||||
"confirm no",
|
||||
)
|
||||
):
|
||||
reasons.append(
|
||||
f"create-only run must confirm no {check} "
|
||||
"mutations were performed"
|
||||
)
|
||||
|
||||
return {
|
||||
"complete": not reasons,
|
||||
"downgraded": bool(reasons),
|
||||
"reasons": reasons,
|
||||
}
|
||||
|
||||
|
||||
def assess_issue_filing_duplicate_summary(
|
||||
report_text: str,
|
||||
*,
|
||||
issues_searched: int | None = None,
|
||||
closest_matches: list[dict] | None = None,
|
||||
duplicate_result: dict | None = None,
|
||||
) -> dict:
|
||||
"""Issue #191: duplicate-check evidence must be summarized in the report."""
|
||||
text = (report_text or "").lower()
|
||||
reasons = []
|
||||
dup = duplicate_result or {}
|
||||
|
||||
if issues_searched is not None:
|
||||
count_tokens = (
|
||||
str(issues_searched),
|
||||
f"{issues_searched} open",
|
||||
f"{issues_searched} issue",
|
||||
)
|
||||
if not any(t in text for t in count_tokens):
|
||||
reasons.append(
|
||||
"duplicate-check summary missing open-issues searched count"
|
||||
)
|
||||
|
||||
matches = closest_matches if closest_matches is not None else dup.get("matches")
|
||||
for match in matches or []:
|
||||
num = match.get("number")
|
||||
if num is not None and f"#{num}" not in text and str(num) not in text:
|
||||
reasons.append(
|
||||
f"duplicate-check summary missing closest issue #{num}"
|
||||
)
|
||||
break
|
||||
|
||||
justification_phrases = (
|
||||
"why new",
|
||||
"why update",
|
||||
"rejected update",
|
||||
"new issue justified",
|
||||
"not a duplicate",
|
||||
"no duplicate",
|
||||
"justified",
|
||||
"instead of expanding",
|
||||
)
|
||||
if not any(p in text for p in justification_phrases):
|
||||
reasons.append(
|
||||
"duplicate-check summary missing why update was rejected or "
|
||||
"why a new issue was justified"
|
||||
)
|
||||
|
||||
dup_proof = issue_duplicate_gate.assess_duplicate_search_proof(
|
||||
report_text, matches or []
|
||||
)
|
||||
if not dup_proof.get("valid"):
|
||||
reasons.extend(dup_proof.get("reasons") or [])
|
||||
|
||||
return {
|
||||
"complete": not reasons,
|
||||
"downgraded": bool(reasons),
|
||||
"reasons": reasons,
|
||||
}
|
||||
|
||||
|
||||
def assess_issue_filing_final_report(
|
||||
report_text: str,
|
||||
*,
|
||||
issue_number: int | None = None,
|
||||
issue_title: str | None = None,
|
||||
action: str = "created",
|
||||
mutations: list[str] | None = None,
|
||||
resolved_capabilities: dict | None = None,
|
||||
issues_searched: int | None = None,
|
||||
closest_matches: list[dict] | None = None,
|
||||
duplicate_result: dict | None = None,
|
||||
performed_mutations: list[str] | None = None,
|
||||
) -> dict:
|
||||
"""Issue #191: composite A-bar for issue-filing final reports."""
|
||||
checks = {
|
||||
"controller_handoff": assess_controller_handoff(
|
||||
report_text, role="issue_filing"
|
||||
),
|
||||
"issue_reference": assess_issue_filing_issue_reference(
|
||||
report_text,
|
||||
issue_number=issue_number,
|
||||
issue_title=issue_title,
|
||||
action=action,
|
||||
),
|
||||
"sha_evidence": assess_issue_filing_sha_evidence(report_text),
|
||||
"mutation_capability": assess_issue_filing_mutation_capability(
|
||||
report_text,
|
||||
mutations or performed_mutations,
|
||||
resolved_capabilities,
|
||||
),
|
||||
"mutation_scope": assess_issue_filing_mutation_scope(
|
||||
report_text,
|
||||
performed_mutations=performed_mutations or mutations,
|
||||
),
|
||||
"duplicate_summary": assess_issue_filing_duplicate_summary(
|
||||
report_text,
|
||||
issues_searched=issues_searched,
|
||||
closest_matches=closest_matches,
|
||||
duplicate_result=duplicate_result,
|
||||
),
|
||||
}
|
||||
|
||||
reasons = []
|
||||
downgraded = False
|
||||
for name, result in checks.items():
|
||||
verdict = result.get("verdict")
|
||||
if verdict in ("missing", "incomplete"):
|
||||
downgraded = True
|
||||
reasons.extend(result.get("reasons") or [])
|
||||
elif result.get("downgraded") or not result.get("complete", True):
|
||||
downgraded = True
|
||||
reasons.extend(
|
||||
f"{name}: {r}" for r in (result.get("reasons") or [])
|
||||
)
|
||||
|
||||
grade = "A" if not downgraded else "downgraded"
|
||||
return {
|
||||
"grade": grade,
|
||||
"downgraded": downgraded,
|
||||
"checks": checks,
|
||||
"reasons": reasons,
|
||||
"complete": not downgraded,
|
||||
}
|
||||
|
||||
|
||||
def assess_duplicate_search_proof(report_text, matches):
|
||||
"""#207: reject LLM duplicate summaries that omit known title matches."""
|
||||
return issue_duplicate_gate.assess_duplicate_search_proof(
|
||||
|
||||
@@ -462,12 +462,6 @@ Role-specific fields (append to the compact block):
|
||||
- review/merge tasks: `Selected PR:`, `Reviewer eligibility:`,
|
||||
`Pinned reviewed head:`, `Review decision:`, `Merge result:`,
|
||||
`Linked issue status:`, `Cleanup status:`
|
||||
- issue-filing tasks (#191): `Issue created or updated:`, `Related issues:`;
|
||||
body must cite exact issue number/title, duplicate-search summary (issues
|
||||
searched, closest matches, why update rejected / new issue justified), full
|
||||
40-char SHAs when citing commits, exact mutation capability per change, and
|
||||
`Only mutation(s):` when a single mutation was performed
|
||||
(`review_proofs.assess_issue_filing_final_report`).
|
||||
- author tasks: `Selected issue:`, `Claim/comment status:`,
|
||||
`PR number opened:`, `No review/merge:` (explicit confirmation)
|
||||
- continuation tasks (#188): `Continuation mode:`, `Existing PR:`,
|
||||
|
||||
@@ -80,6 +80,14 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = {
|
||||
"permission": "gitea.branch.delete",
|
||||
"role": "author",
|
||||
},
|
||||
"commit_files": {
|
||||
"permission": "gitea.repo.commit",
|
||||
"role": "author",
|
||||
},
|
||||
"gitea_commit_files": {
|
||||
"permission": "gitea.repo.commit",
|
||||
"role": "author",
|
||||
},
|
||||
}
|
||||
|
||||
# Issue-mutating MCP tools and their resolver task keys.
|
||||
@@ -89,6 +97,7 @@ ISSUE_MUTATION_TOOL_TASKS: dict[str, str] = {
|
||||
"gitea_create_issue_comment": "comment_issue",
|
||||
"gitea_mark_issue": "mark_issue",
|
||||
"gitea_set_issue_labels": "set_issue_labels",
|
||||
"gitea_commit_files": "commit_files",
|
||||
}
|
||||
|
||||
|
||||
|
||||
@@ -1,76 +0,0 @@
|
||||
"""Docs checks for agent temp-artifact cleanup guidance (#261).
|
||||
|
||||
Failed subagent runs left throwaway helper scripts (e.g.
|
||||
``_encode_commit_payload.py``, ``_emit_payload.py``) in the shared working
|
||||
tree, polluting ``git status`` and tripping issue-lock / preflight gates.
|
||||
These tests keep the cleanup runbook guidance present and the ignore
|
||||
patterns in place so stray helpers can never dirty a worktree again.
|
||||
"""
|
||||
from pathlib import Path
|
||||
|
||||
REPO_ROOT = Path(__file__).resolve().parent.parent
|
||||
RUNBOOK = REPO_ROOT / "docs" / "llm-workflow-runbooks.md"
|
||||
GITIGNORE = REPO_ROOT / ".gitignore"
|
||||
|
||||
ARTIFACT_PATTERNS = ("_encode_*.py", "_emit_*.py", "_inline_*.py")
|
||||
|
||||
|
||||
def _runbook_text():
|
||||
assert RUNBOOK.is_file(), "missing docs/llm-workflow-runbooks.md"
|
||||
return RUNBOOK.read_text(encoding="utf-8")
|
||||
|
||||
|
||||
def _gitignore_text():
|
||||
assert GITIGNORE.is_file(), "missing .gitignore"
|
||||
return GITIGNORE.read_text(encoding="utf-8")
|
||||
|
||||
|
||||
def test_runbook_has_agent_temp_artifact_cleanup_section():
|
||||
text = _runbook_text()
|
||||
assert "## Agent temp artifact cleanup" in text, (
|
||||
"runbook lacks an 'Agent temp artifact cleanup' section")
|
||||
assert "#261" in text, "cleanup guidance does not reference issue #261"
|
||||
|
||||
|
||||
def test_runbook_names_known_stray_artifacts():
|
||||
text = _runbook_text()
|
||||
for name in ("_encode_commit_payload.py", "_emit_payload.py"):
|
||||
assert name in text, f"runbook does not name stray artifact {name!r}"
|
||||
|
||||
|
||||
def test_runbook_covers_all_artifact_patterns():
|
||||
text = _runbook_text()
|
||||
for pattern in ("_encode_*", "_emit_*", "_inline_*"):
|
||||
assert pattern in text, f"runbook does not cover pattern {pattern!r}"
|
||||
|
||||
|
||||
def test_runbook_requires_cleanup_after_commit_or_abort():
|
||||
text = _runbook_text().lower()
|
||||
assert "delete" in text or "remove" in text
|
||||
for phase in ("completes", "aborts"):
|
||||
assert phase in text, (
|
||||
f"runbook does not require cleanup when the MCP commit {phase}")
|
||||
|
||||
|
||||
def test_runbook_explains_lock_and_preflight_impact():
|
||||
text = _runbook_text().lower()
|
||||
assert "gitea_lock_issue" in text
|
||||
assert "preflight" in text or "pre-flight" in text
|
||||
|
||||
|
||||
def test_gitignore_blocks_agent_helper_patterns():
|
||||
lines = [
|
||||
ln.strip() for ln in _gitignore_text().splitlines()
|
||||
if ln.strip() and not ln.strip().startswith("#")
|
||||
]
|
||||
for pattern in ARTIFACT_PATTERNS:
|
||||
assert pattern in lines, f".gitignore lacks pattern {pattern!r}"
|
||||
|
||||
|
||||
def test_no_stray_agent_helpers_in_repo_root():
|
||||
"""Acceptance: artifacts stay removed from default worktrees."""
|
||||
stray = [
|
||||
p.name for pattern in ARTIFACT_PATTERNS
|
||||
for p in REPO_ROOT.glob(pattern)
|
||||
]
|
||||
assert not stray, f"stray agent helper artifacts present: {stray}"
|
||||
@@ -0,0 +1,251 @@
|
||||
"""Regression tests: gitea_commit_files tool gates match resolver and whoami verification.
|
||||
|
||||
Covers Issue #262 requirements.
|
||||
"""
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
import tempfile
|
||||
import unittest
|
||||
from unittest.mock import patch
|
||||
|
||||
sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
|
||||
|
||||
import mcp_server
|
||||
import task_capability_map
|
||||
import gitea_config
|
||||
|
||||
CONFIG = {
|
||||
"version": 2,
|
||||
"contexts": {
|
||||
"ctx": {
|
||||
"enabled": True,
|
||||
"gitea": {"enabled": True, "base_url": "https://gitea.example.com"},
|
||||
}
|
||||
},
|
||||
"profiles": {
|
||||
"full-author": {
|
||||
"enabled": True,
|
||||
"context": "ctx",
|
||||
"role": "author",
|
||||
"username": "author-user",
|
||||
"auth": {"type": "env", "name": "GITEA_TOKEN_AUTHOR"},
|
||||
"allowed_operations": [
|
||||
"gitea.read", "gitea.issue.create", "gitea.repo.commit"
|
||||
],
|
||||
"forbidden_operations": [],
|
||||
"execution_profile": "full-author",
|
||||
},
|
||||
"reviewer-no-commit": {
|
||||
"enabled": True,
|
||||
"context": "ctx",
|
||||
"role": "reviewer",
|
||||
"username": "reviewer-user",
|
||||
"auth": {"type": "env", "name": "GITEA_TOKEN_REVIEWER"},
|
||||
"allowed_operations": [
|
||||
"gitea.read", "gitea.pr.review", "gitea.pr.approve"
|
||||
],
|
||||
"forbidden_operations": [
|
||||
"gitea.repo.commit", "gitea.pr.create", "gitea.branch.push"
|
||||
],
|
||||
"execution_profile": "reviewer-no-commit",
|
||||
},
|
||||
},
|
||||
"rules": {"allow_runtime_switching": False},
|
||||
}
|
||||
|
||||
|
||||
class TestCommitFilesGate(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self._remotes = patch.dict(mcp_server.REMOTES, {
|
||||
"prgs": {"host": "gitea.example.com", "org": "Example-Org",
|
||||
"repo": "Example-Repo"},
|
||||
})
|
||||
self._remotes.start()
|
||||
mcp_server._IDENTITY_CACHE.clear()
|
||||
gitea_config._active_profile_override = None
|
||||
self._dir = tempfile.TemporaryDirectory()
|
||||
self.config_path = os.path.join(self._dir.name, "profiles.json")
|
||||
with open(self.config_path, "w", encoding="utf-8") as fh:
|
||||
fh.write(json.dumps(CONFIG))
|
||||
|
||||
def tearDown(self):
|
||||
self._remotes.stop()
|
||||
mcp_server._IDENTITY_CACHE.clear()
|
||||
gitea_config._active_profile_override = None
|
||||
self._dir.cleanup()
|
||||
|
||||
def _env(self, profile: str) -> dict:
|
||||
return {
|
||||
"GITEA_MCP_CONFIG": self.config_path,
|
||||
"GITEA_MCP_PROFILE": profile,
|
||||
"GITEA_TOKEN_AUTHOR": "author-pass",
|
||||
"GITEA_TOKEN_REVIEWER": "reviewer-pass",
|
||||
}
|
||||
|
||||
@patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
|
||||
return_value=(True, []))
|
||||
@patch("mcp_server.api_request")
|
||||
@patch("mcp_server.get_auth_header", return_value="token author-pass")
|
||||
def test_allowed_author_proceeds(self, _auth, mock_api, _role):
|
||||
mock_api.return_value = {
|
||||
"commit": {"sha": "abc123commit"},
|
||||
"branch": {"name": "some-branch"},
|
||||
}
|
||||
with patch.dict(os.environ, self._env("full-author"), clear=True):
|
||||
resolve = mcp_server.gitea_resolve_task_capability(
|
||||
task="commit_files", remote="prgs"
|
||||
)
|
||||
self.assertTrue(resolve["allowed_in_current_session"])
|
||||
|
||||
res = mcp_server.gitea_commit_files(
|
||||
files=[{"operation": "create", "path": "x.txt", "content": "YQ=="}],
|
||||
message="Add x",
|
||||
remote="prgs",
|
||||
)
|
||||
self.assertTrue(res["success"])
|
||||
self.assertEqual(res["commit"], "abc123commit")
|
||||
|
||||
@patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
|
||||
return_value=(True, []))
|
||||
@patch("mcp_server.api_request")
|
||||
@patch("mcp_server.get_auth_header", return_value="token reviewer-pass")
|
||||
def test_denied_reviewer_blocked(self, _auth, mock_api, _role):
|
||||
mock_api.return_value = {"login": "reviewer-user"}
|
||||
with patch.dict(os.environ, self._env("reviewer-no-commit"), clear=True):
|
||||
resolve = mcp_server.gitea_resolve_task_capability(
|
||||
task="commit_files", remote="prgs"
|
||||
)
|
||||
self.assertFalse(resolve["allowed_in_current_session"])
|
||||
|
||||
res = mcp_server.gitea_commit_files(
|
||||
files=[{"operation": "create", "path": "x.txt", "content": "YQ=="}],
|
||||
message="Add x",
|
||||
remote="prgs",
|
||||
)
|
||||
self.assertFalse(res["success"])
|
||||
self.assertFalse(res.get("performed", True))
|
||||
self.assertIn("permission_report", res)
|
||||
self.assertEqual(
|
||||
res["permission_report"]["missing_permission"], "gitea.repo.commit"
|
||||
)
|
||||
post_calls = [c for c in mock_api.call_args_list if len(c.args) > 0 and c.args[0] == "POST"]
|
||||
self.assertFalse(post_calls)
|
||||
|
||||
@patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
|
||||
return_value=(True, []))
|
||||
@patch("mcp_server.api_request")
|
||||
@patch("mcp_server.get_auth_header", return_value="token reviewer-pass")
|
||||
def test_unknown_profile_fails_closed(self, _auth, mock_api, _role):
|
||||
mock_api.return_value = {"login": "reviewer-user"}
|
||||
with patch.dict(os.environ, self._env("non-existent"), clear=True):
|
||||
res = mcp_server.gitea_commit_files(
|
||||
files=[{"operation": "create", "path": "x.txt", "content": "YQ=="}],
|
||||
message="Add x",
|
||||
remote="prgs",
|
||||
)
|
||||
self.assertFalse(res["success"])
|
||||
self.assertFalse(res.get("performed", True))
|
||||
post_calls = [c for c in mock_api.call_args_list if len(c.args) > 0 and c.args[0] == "POST"]
|
||||
self.assertFalse(post_calls)
|
||||
|
||||
|
||||
class TestPreflightCommitFilesGate(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self._remotes = patch.dict(mcp_server.REMOTES, {
|
||||
"prgs": {"host": "gitea.example.com", "org": "Example-Org",
|
||||
"repo": "Example-Repo"},
|
||||
})
|
||||
self._remotes.start()
|
||||
mcp_server._IDENTITY_CACHE.clear()
|
||||
|
||||
self.orig_whoami_called = mcp_server._preflight_whoami_called
|
||||
self.orig_capability_called = mcp_server._preflight_capability_called
|
||||
self.orig_whoami_violation = mcp_server._preflight_whoami_violation
|
||||
self.orig_capability_violation = mcp_server._preflight_capability_violation
|
||||
self.orig_resolved_role = mcp_server._preflight_resolved_role
|
||||
self.orig_process_start = mcp_server._process_start_porcelain
|
||||
self.orig_whoami_baseline = mcp_server._preflight_whoami_baseline_porcelain
|
||||
self.orig_capability_baseline = mcp_server._preflight_capability_baseline_porcelain
|
||||
self.orig_whoami_files = mcp_server._preflight_whoami_violation_files
|
||||
self.orig_capability_files = mcp_server._preflight_capability_violation_files
|
||||
self.orig_reviewer_files = mcp_server._preflight_reviewer_violation_files
|
||||
|
||||
mcp_server._preflight_whoami_called = False
|
||||
mcp_server._preflight_capability_called = False
|
||||
mcp_server._preflight_whoami_violation = False
|
||||
mcp_server._preflight_capability_violation = False
|
||||
mcp_server._preflight_resolved_role = None
|
||||
mcp_server._process_start_porcelain = ""
|
||||
mcp_server._preflight_whoami_baseline_porcelain = None
|
||||
mcp_server._preflight_capability_baseline_porcelain = None
|
||||
mcp_server._preflight_whoami_violation_files = []
|
||||
mcp_server._preflight_capability_violation_files = []
|
||||
mcp_server._preflight_reviewer_violation_files = []
|
||||
|
||||
self._dir = tempfile.TemporaryDirectory()
|
||||
self.config_path = os.path.join(self._dir.name, "profiles.json")
|
||||
with open(self.config_path, "w", encoding="utf-8") as fh:
|
||||
fh.write(json.dumps(CONFIG))
|
||||
|
||||
def tearDown(self):
|
||||
self._remotes.stop()
|
||||
mcp_server._IDENTITY_CACHE.clear()
|
||||
|
||||
mcp_server._preflight_whoami_called = self.orig_whoami_called
|
||||
mcp_server._preflight_capability_called = self.orig_capability_called
|
||||
mcp_server._preflight_whoami_violation = self.orig_whoami_violation
|
||||
mcp_server._preflight_capability_violation = self.orig_capability_violation
|
||||
mcp_server._preflight_resolved_role = self.orig_resolved_role
|
||||
mcp_server._process_start_porcelain = self.orig_process_start
|
||||
mcp_server._preflight_whoami_baseline_porcelain = self.orig_whoami_baseline
|
||||
mcp_server._preflight_capability_baseline_porcelain = self.orig_capability_baseline
|
||||
mcp_server._preflight_whoami_violation_files = self.orig_whoami_files
|
||||
mcp_server._preflight_capability_violation_files = self.orig_capability_files
|
||||
mcp_server._preflight_reviewer_violation_files = self.orig_reviewer_files
|
||||
|
||||
self._dir.cleanup()
|
||||
os.environ.pop("GITEA_TEST_FORCE_DIRTY", None)
|
||||
os.environ.pop("GITEA_TEST_PORCELAIN", None)
|
||||
|
||||
def _env(self, profile: str) -> dict:
|
||||
return {
|
||||
"GITEA_MCP_CONFIG": self.config_path,
|
||||
"GITEA_MCP_PROFILE": profile,
|
||||
"GITEA_TOKEN_AUTHOR": "author-pass",
|
||||
}
|
||||
|
||||
@patch("mcp_server.api_request")
|
||||
@patch("mcp_server.get_auth_header", return_value="token author-pass")
|
||||
def test_preflight_not_called_blocks_commit(self, _auth, mock_api):
|
||||
mock_api.return_value = {"login": "author-user"}
|
||||
with patch.dict(os.environ, self._env("full-author"), clear=True):
|
||||
os.environ["GITEA_TEST_PORCELAIN"] = ""
|
||||
with self.assertRaises(RuntimeError) as ctx:
|
||||
mcp_server.gitea_commit_files(
|
||||
files=[{"operation": "create", "path": "x.txt", "content": "YQ=="}],
|
||||
message="Add x",
|
||||
remote="prgs",
|
||||
)
|
||||
self.assertIn("Identity (gitea_whoami) has not been verified", str(ctx.exception))
|
||||
|
||||
@patch("mcp_server.api_request")
|
||||
@patch("mcp_server.get_auth_header", return_value="token author-pass")
|
||||
def test_dirty_workspace_before_whoami_blocks_commit(self, _auth, mock_api):
|
||||
mock_api.return_value = {"login": "author-user"}
|
||||
with patch.dict(os.environ, self._env("full-author"), clear=True):
|
||||
os.environ["GITEA_TEST_FORCE_DIRTY"] = "1"
|
||||
mcp_server.record_preflight_check("whoami")
|
||||
mcp_server.record_preflight_check("capability", resolved_role="author")
|
||||
|
||||
with self.assertRaises(RuntimeError) as ctx:
|
||||
mcp_server.gitea_commit_files(
|
||||
files=[{"operation": "create", "path": "x.txt", "content": "YQ=="}],
|
||||
message="Add x",
|
||||
remote="prgs",
|
||||
)
|
||||
self.assertIn("Workspace file edits occurred before gitea_whoami verification", str(ctx.exception))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -1071,9 +1071,11 @@ class TestGetFile(unittest.TestCase):
|
||||
# ---------------------------------------------------------------------------
|
||||
class TestCommitFiles(unittest.TestCase):
|
||||
|
||||
@patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
|
||||
return_value=(True, []))
|
||||
@patch("mcp_server.api_request")
|
||||
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
||||
def test_commit_files_success(self, _auth, mock_api):
|
||||
def test_commit_files_success(self, _auth, mock_api, _role):
|
||||
mock_api.return_value = {
|
||||
"commit": {"sha": "commit-sha-123"},
|
||||
"branch": {"name": "test-branch"}
|
||||
@@ -1081,11 +1083,15 @@ class TestCommitFiles(unittest.TestCase):
|
||||
files = [
|
||||
{"operation": "create", "path": "test.txt", "content": "SGVsbG8="}
|
||||
]
|
||||
result = gitea_commit_files(
|
||||
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")
|
||||
|
||||
@@ -33,9 +33,6 @@ from review_proofs import ( # noqa: E402
|
||||
assess_empty_queue_report,
|
||||
assess_fresh_issue_selection,
|
||||
assess_inventory_completeness,
|
||||
assess_issue_filing_final_report,
|
||||
assess_issue_filing_mutation_capability,
|
||||
assess_issue_filing_sha_evidence,
|
||||
assess_issue_selection_final_report,
|
||||
assess_queue_target_final_report,
|
||||
assess_reviewer_queue_inventory,
|
||||
@@ -1932,115 +1929,5 @@ class TestIssueSelectionContinuation(unittest.TestCase):
|
||||
self.assertEqual(result["grade"], "A")
|
||||
|
||||
|
||||
class TestIssueFilingFinalReport(unittest.TestCase):
|
||||
"""Issue #191: issue-filing runs need A-bar final report proofs."""
|
||||
|
||||
ISSUE_TITLE = (
|
||||
"Implement fail-closed continuation mode for issues already "
|
||||
"represented by open PRs"
|
||||
)
|
||||
FULL_SHA = "a4060c5de00f2b1c9e88f4f6f0f3f9a7b2c1d0e9"
|
||||
CAPABILITIES = {
|
||||
"create_issue": {
|
||||
"requested_task": "create_issue",
|
||||
"required_operation_permission": "gitea.issue.create",
|
||||
"allowed_in_current_session": True,
|
||||
},
|
||||
}
|
||||
CLOSEST = [{"number": 183, "title": "Harden author-run reporting", "state": "open"}]
|
||||
|
||||
def _good_report(self, *, sha_line=None):
|
||||
lines = [
|
||||
"Created Gitea-Tools #189 — "
|
||||
f"`{self.ISSUE_TITLE}`",
|
||||
"Duplicate check: searched 12 open issues.",
|
||||
"Closest existing: #183 — Harden author-run reporting "
|
||||
"(update rejected — different scope).",
|
||||
"Why new issue justified: continuation mode is distinct from #183.",
|
||||
"Only mutation: issue creation (gitea.issue.create).",
|
||||
"Confirm no labels, comments, PRs, reviews, merges, or closes.",
|
||||
"## Controller Handoff",
|
||||
"- Task: file issue",
|
||||
"- Repo: Scaled-Tech-Consulting/Gitea-Tools",
|
||||
"- Role: author",
|
||||
"- Identity: prgs-author",
|
||||
"- Issue/PR: #189",
|
||||
"- Branch/SHA: n/a",
|
||||
"- Files changed: none",
|
||||
"- Validation: duplicate gate + create_issue capability resolved",
|
||||
"- Mutations: create_issue via gitea.issue.create",
|
||||
"- Workspace mutations: none",
|
||||
"- Current status: issue created",
|
||||
"- Blockers: none",
|
||||
"- Next: implementation",
|
||||
"- Safety: no review/merge",
|
||||
"- Issue created or updated: created #189",
|
||||
"- Related issues: #183, #188",
|
||||
]
|
||||
if sha_line:
|
||||
lines.insert(4, sha_line)
|
||||
return "\n".join(lines)
|
||||
|
||||
def test_complete_issue_filing_report_earns_a(self):
|
||||
result = assess_issue_filing_final_report(
|
||||
self._good_report(),
|
||||
issue_number=189,
|
||||
issue_title=self.ISSUE_TITLE,
|
||||
action="created",
|
||||
mutations=["create_issue"],
|
||||
resolved_capabilities=self.CAPABILITIES,
|
||||
issues_searched=12,
|
||||
closest_matches=self.CLOSEST,
|
||||
performed_mutations=["create_issue"],
|
||||
)
|
||||
self.assertEqual(result["grade"], "A")
|
||||
self.assertFalse(result["downgraded"])
|
||||
|
||||
def test_missing_controller_handoff_downgrades(self):
|
||||
report = self._good_report().replace("## Controller Handoff", "")
|
||||
result = assess_issue_filing_final_report(
|
||||
report,
|
||||
issue_number=189,
|
||||
issue_title=self.ISSUE_TITLE,
|
||||
mutations=["create_issue"],
|
||||
resolved_capabilities=self.CAPABILITIES,
|
||||
issues_searched=12,
|
||||
performed_mutations=["create_issue"],
|
||||
)
|
||||
self.assertTrue(result["downgraded"])
|
||||
|
||||
def test_abbreviated_sha_downgrades(self):
|
||||
result = assess_issue_filing_sha_evidence(
|
||||
f"old head: {self.FULL_SHA[:7]}"
|
||||
)
|
||||
self.assertTrue(result["downgraded"])
|
||||
|
||||
def test_mutation_without_capability_proof_downgrades(self):
|
||||
report = self._good_report().replace("gitea.issue.create", "allowed")
|
||||
result = assess_issue_filing_mutation_capability(
|
||||
report,
|
||||
["create_issue"],
|
||||
self.CAPABILITIES,
|
||||
)
|
||||
self.assertTrue(result["downgraded"])
|
||||
|
||||
def test_label_mutation_without_label_proof_downgrades(self):
|
||||
caps = {
|
||||
"set_issue_labels": {
|
||||
"requested_task": "set_issue_labels",
|
||||
"required_operation_permission": "gitea.issue.comment",
|
||||
"allowed_in_current_session": True,
|
||||
},
|
||||
}
|
||||
report = "\n".join([
|
||||
"Updated labels with gitea.issue.comment capability resolved.",
|
||||
"Only mutations: set_issue_labels",
|
||||
])
|
||||
result = assess_issue_filing_mutation_capability(
|
||||
report, ["set_issue_labels"], caps
|
||||
)
|
||||
self.assertTrue(result["downgraded"])
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
|
||||
Reference in New Issue
Block a user