Compare commits
2
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
dfcbca7355 | ||
|
|
9ea2707289 |
+4
-206
@@ -16,70 +16,10 @@ Configuration (mcp_config.json):
|
|||||||
import os
|
import os
|
||||||
import re
|
import re
|
||||||
import sys
|
import sys
|
||||||
import json
|
|
||||||
import functools
|
import functools
|
||||||
import contextlib
|
import contextlib
|
||||||
import subprocess
|
import subprocess
|
||||||
|
|
||||||
|
|
||||||
LOCK_FILE = "/tmp/gitea_mutation_authority.lock"
|
|
||||||
ISSUE_LOCK_FILE = "/tmp/gitea_issue_lock.json"
|
|
||||||
|
|
||||||
def record_mutation_authority(profile_name: str | None, identity: str | None, remote: str | None, task: str | None):
|
|
||||||
"""Record the resolved capability context to fail-closed lock file."""
|
|
||||||
data = {
|
|
||||||
"initial_profile": profile_name,
|
|
||||||
"initial_identity": identity,
|
|
||||||
"current_profile": profile_name,
|
|
||||||
"current_identity": identity,
|
|
||||||
"remote": remote,
|
|
||||||
"task": task,
|
|
||||||
"role_pivot_authorized": False,
|
|
||||||
"role_pivot_record": None,
|
|
||||||
}
|
|
||||||
try:
|
|
||||||
with open(LOCK_FILE, "w", encoding="utf-8") as f:
|
|
||||||
json.dump(data, f)
|
|
||||||
except Exception:
|
|
||||||
pass
|
|
||||||
|
|
||||||
def verify_mutation_authority(remote: str | None, host: str | None = None, required_role: str = "reviewer"):
|
|
||||||
"""Verify that the current mutation matches the locked capability context."""
|
|
||||||
if not os.path.exists(LOCK_FILE):
|
|
||||||
raise RuntimeError("Mutation authority lock is missing (fail closed)")
|
|
||||||
|
|
||||||
try:
|
|
||||||
with open(LOCK_FILE, "r", encoding="utf-8") as f:
|
|
||||||
data = json.load(f)
|
|
||||||
except Exception as e:
|
|
||||||
raise RuntimeError(f"Could not read mutation authority lock: {e} (fail closed)")
|
|
||||||
|
|
||||||
if data.get("remote") != remote:
|
|
||||||
raise RuntimeError(
|
|
||||||
f"Mutation remote '{remote}' does not match locked remote '{data.get('remote')}' (fail closed)"
|
|
||||||
)
|
|
||||||
|
|
||||||
profile = get_profile()
|
|
||||||
active_profile = profile.get("profile_name")
|
|
||||||
h = host or (REMOTES.get(remote, {}).get("host") if remote in REMOTES else None)
|
|
||||||
active_identity = _authenticated_username(h) if h else None
|
|
||||||
|
|
||||||
locked_profile = data.get("current_profile")
|
|
||||||
locked_identity = data.get("current_identity")
|
|
||||||
|
|
||||||
if active_profile != locked_profile or active_identity != locked_identity:
|
|
||||||
raise RuntimeError(
|
|
||||||
f"Mutation profile '{active_profile}' or identity '{active_identity}' "
|
|
||||||
f"does not match locked authority (profile: '{locked_profile}', identity: '{locked_identity}') (fail closed)"
|
|
||||||
)
|
|
||||||
|
|
||||||
# Check reviewer/author role pivot boundaries
|
|
||||||
if required_role == "reviewer" and "author" in str(data.get("initial_profile")).lower() and "reviewer" in str(active_profile).lower():
|
|
||||||
if not data.get("role_pivot_authorized"):
|
|
||||||
raise RuntimeError(
|
|
||||||
"Attempted reviewer mutation from author session without authorized role pivot (fail closed)"
|
|
||||||
)
|
|
||||||
|
|
||||||
# Resolve the project root. MCP clients must launch this script directly with
|
# Resolve the project root. MCP clients must launch this script directly with
|
||||||
# the venv interpreter (venv/bin/python3) — see the config example above. We do
|
# the venv interpreter (venv/bin/python3) — see the config example above. We do
|
||||||
# NOT os.execv() to re-point the interpreter: replacing the process after the
|
# NOT os.execv() to re-point the interpreter: replacing the process after the
|
||||||
@@ -400,84 +340,6 @@ def gitea_create_issue(
|
|||||||
return _with_optional_url({"number": data["number"]}, data.get("html_url"))
|
return _with_optional_url({"number": data["number"]}, data.get("html_url"))
|
||||||
|
|
||||||
|
|
||||||
@mcp.tool()
|
|
||||||
def gitea_lock_issue(
|
|
||||||
issue_number: int,
|
|
||||||
branch_name: str,
|
|
||||||
remote: str = "dadeschools",
|
|
||||||
host: str | None = None,
|
|
||||||
org: str | None = None,
|
|
||||||
repo: str | None = None,
|
|
||||||
) -> dict:
|
|
||||||
"""Lock exactly one Gitea issue and its branch name to ensure durable tracking.
|
|
||||||
|
|
||||||
Args:
|
|
||||||
issue_number: The tracking issue number.
|
|
||||||
branch_name: The branch name (must match (fix|feat|docs|chore)/issue-<issue_number>-<desc>).
|
|
||||||
remote: Known instance — 'dadeschools' or 'prgs'.
|
|
||||||
host: Override Gitea host.
|
|
||||||
org: Override Org.
|
|
||||||
repo: Override Repo.
|
|
||||||
"""
|
|
||||||
# 1. Enforce branch name includes issue number
|
|
||||||
expected_pattern = f"issue-{issue_number}"
|
|
||||||
if expected_pattern not in branch_name:
|
|
||||||
raise ValueError(
|
|
||||||
f"Branch name '{branch_name}' must contain locked issue pattern '{expected_pattern}' (fail closed)"
|
|
||||||
)
|
|
||||||
|
|
||||||
# 2. Check if the issue already has an open PR (reuse protection)
|
|
||||||
h, o, r = _resolve(remote, host, org, repo)
|
|
||||||
auth = _auth(h)
|
|
||||||
url = f"{repo_api_url(h, o, r)}/pulls?state=open"
|
|
||||||
|
|
||||||
try:
|
|
||||||
prs = api_get_all(url, auth)
|
|
||||||
except Exception as e:
|
|
||||||
raise RuntimeError(f"Could not list open PRs to verify issue lock: {e}")
|
|
||||||
|
|
||||||
for pr in prs:
|
|
||||||
pr_head = pr.get("head", {}).get("ref", "")
|
|
||||||
pr_title = pr.get("title", "")
|
|
||||||
pr_body = pr.get("body", "")
|
|
||||||
|
|
||||||
if expected_pattern in pr_head:
|
|
||||||
raise ValueError(
|
|
||||||
f"Issue #{issue_number} is already tied to an open PR (PR #{pr.get('number')}, branch '{pr_head}') (fail closed)"
|
|
||||||
)
|
|
||||||
|
|
||||||
patterns = [
|
|
||||||
f"closes #{issue_number}",
|
|
||||||
f"fixes #{issue_number}",
|
|
||||||
]
|
|
||||||
text_to_check = f"{pr_title} {pr_body}".lower()
|
|
||||||
if any(p in text_to_check for p in patterns):
|
|
||||||
raise ValueError(
|
|
||||||
f"Issue #{issue_number} is already tied to an open PR (PR #{pr.get('number')}) via Closes/Fixes reference (fail closed)"
|
|
||||||
)
|
|
||||||
|
|
||||||
data = {
|
|
||||||
"issue_number": issue_number,
|
|
||||||
"branch_name": branch_name,
|
|
||||||
"remote": remote,
|
|
||||||
"org": o,
|
|
||||||
"repo": r,
|
|
||||||
}
|
|
||||||
|
|
||||||
try:
|
|
||||||
with open(ISSUE_LOCK_FILE, "w", encoding="utf-8") as f:
|
|
||||||
json.dump(data, f)
|
|
||||||
except Exception as e:
|
|
||||||
raise RuntimeError(f"Could not write issue lock file: {e}")
|
|
||||||
|
|
||||||
return {
|
|
||||||
"success": True,
|
|
||||||
"message": f"Successfully locked issue #{issue_number} to branch '{branch_name}' (fail-closed check complete).",
|
|
||||||
"issue_number": issue_number,
|
|
||||||
"branch_name": branch_name,
|
|
||||||
}
|
|
||||||
|
|
||||||
|
|
||||||
@mcp.tool()
|
@mcp.tool()
|
||||||
def gitea_create_pr(
|
def gitea_create_pr(
|
||||||
title: str,
|
title: str,
|
||||||
@@ -505,41 +367,6 @@ def gitea_create_pr(
|
|||||||
dict with 'number' of the created PR ('url' only with the reveal opt-in).
|
dict with 'number' of the created PR ('url' only with the reveal opt-in).
|
||||||
"""
|
"""
|
||||||
h, o, r = _resolve(remote, host, org, repo)
|
h, o, r = _resolve(remote, host, org, repo)
|
||||||
|
|
||||||
# ── Issue Lock Validation (Issue #194 / #196) ──
|
|
||||||
if not os.path.exists(ISSUE_LOCK_FILE):
|
|
||||||
raise RuntimeError("Issue lock is missing (fail closed). Call gitea_lock_issue first.")
|
|
||||||
|
|
||||||
try:
|
|
||||||
with open(ISSUE_LOCK_FILE, "r", encoding="utf-8") as f:
|
|
||||||
lock_data = json.load(f)
|
|
||||||
except Exception as e:
|
|
||||||
raise RuntimeError(f"Could not read issue lock file: {e} (fail closed)")
|
|
||||||
|
|
||||||
locked_issue = lock_data.get("issue_number")
|
|
||||||
locked_branch = lock_data.get("branch_name")
|
|
||||||
|
|
||||||
if head != locked_branch:
|
|
||||||
raise ValueError(
|
|
||||||
f"PR head branch '{head}' does not match locked branch '{locked_branch}' (fail closed)"
|
|
||||||
)
|
|
||||||
|
|
||||||
# Check for forbidden terms anywhere in title/body
|
|
||||||
forbidden_terms = ["equivalent", "related", "same as"]
|
|
||||||
text_to_check = f"{title} {body}".lower()
|
|
||||||
for term in forbidden_terms:
|
|
||||||
if term in text_to_check:
|
|
||||||
raise ValueError(
|
|
||||||
f"PR title or body contains forbidden term '{term}' (fail closed)"
|
|
||||||
)
|
|
||||||
|
|
||||||
# Ensure Closes #<locked_issue> or Fixes #<locked_issue> is present exactly
|
|
||||||
closes_pattern = re.compile(rf"\b(closes|fixes)\s+#{locked_issue}\b", re.IGNORECASE)
|
|
||||||
if not closes_pattern.search(text_to_check):
|
|
||||||
raise ValueError(
|
|
||||||
f"PR title or body must contain 'Closes #{locked_issue}' or 'Fixes #{locked_issue}' exactly to ensure durable tracking (fail closed)"
|
|
||||||
)
|
|
||||||
|
|
||||||
auth = _auth(h)
|
auth = _auth(h)
|
||||||
url = f"{repo_api_url(h, o, r)}/pulls"
|
url = f"{repo_api_url(h, o, r)}/pulls"
|
||||||
payload = {"title": title, "body": body, "head": head, "base": base}
|
payload = {"title": title, "body": body, "head": head, "base": base}
|
||||||
@@ -1164,12 +991,6 @@ def gitea_submit_pr_review(
|
|||||||
}
|
}
|
||||||
reasons = result["reasons"]
|
reasons = result["reasons"]
|
||||||
|
|
||||||
try:
|
|
||||||
verify_mutation_authority(remote, host, required_role="reviewer")
|
|
||||||
except RuntimeError as e:
|
|
||||||
reasons.append(str(e))
|
|
||||||
return result
|
|
||||||
|
|
||||||
# Gate 1 — valid review action (no mutation on unknown action).
|
# Gate 1 — valid review action (no mutation on unknown action).
|
||||||
if action not in _REVIEW_ACTIONS:
|
if action not in _REVIEW_ACTIONS:
|
||||||
reasons.append(
|
reasons.append(
|
||||||
@@ -1490,12 +1311,6 @@ def gitea_merge_pr(
|
|||||||
}
|
}
|
||||||
reasons = result["reasons"]
|
reasons = result["reasons"]
|
||||||
|
|
||||||
try:
|
|
||||||
verify_mutation_authority(remote, host, required_role="reviewer")
|
|
||||||
except RuntimeError as e:
|
|
||||||
reasons.append(str(e))
|
|
||||||
return result
|
|
||||||
|
|
||||||
# Gate 1 — valid merge method (no API call on a bad method).
|
# Gate 1 — valid merge method (no API call on a bad method).
|
||||||
if do not in _MERGE_METHODS:
|
if do not in _MERGE_METHODS:
|
||||||
reasons.append(
|
reasons.append(
|
||||||
@@ -3172,25 +2987,6 @@ def gitea_activate_profile(
|
|||||||
after_profile = get_profile()["profile_name"]
|
after_profile = get_profile()["profile_name"]
|
||||||
after_identity = _authenticated_username(h) if h else None
|
after_identity = _authenticated_username(h) if h else None
|
||||||
|
|
||||||
# 4.5 Record pivot in mutation authority lock
|
|
||||||
if os.path.exists(LOCK_FILE):
|
|
||||||
try:
|
|
||||||
with open(LOCK_FILE, "r", encoding="utf-8") as f:
|
|
||||||
lock_data = json.load(f)
|
|
||||||
lock_data["current_profile"] = after_profile
|
|
||||||
lock_data["current_identity"] = after_identity
|
|
||||||
lock_data["role_pivot_authorized"] = True
|
|
||||||
lock_data["role_pivot_record"] = {
|
|
||||||
"from_profile": before_profile,
|
|
||||||
"to_profile": after_profile,
|
|
||||||
"from_identity": before_identity,
|
|
||||||
"to_identity": after_identity
|
|
||||||
}
|
|
||||||
with open(LOCK_FILE, "w", encoding="utf-8") as f:
|
|
||||||
json.dump(lock_data, f)
|
|
||||||
except Exception:
|
|
||||||
pass
|
|
||||||
|
|
||||||
# 5. Audit the switch if auditing is on
|
# 5. Audit the switch if auditing is on
|
||||||
_audit(
|
_audit(
|
||||||
"activate_profile",
|
"activate_profile",
|
||||||
@@ -3541,6 +3337,10 @@ def gitea_resolve_task_capability(
|
|||||||
"permission": "gitea.issue.write",
|
"permission": "gitea.issue.write",
|
||||||
"role": "author",
|
"role": "author",
|
||||||
},
|
},
|
||||||
|
"mark_issue": {
|
||||||
|
"permission": "gitea.issue.write",
|
||||||
|
"role": "author",
|
||||||
|
},
|
||||||
"create_branch": {
|
"create_branch": {
|
||||||
"permission": "gitea.branch.create",
|
"permission": "gitea.branch.create",
|
||||||
"role": "author",
|
"role": "author",
|
||||||
@@ -3680,8 +3480,6 @@ def gitea_resolve_task_capability(
|
|||||||
"STOP: the active profile cannot perform the requested task; "
|
"STOP: the active profile cannot perform the requested task; "
|
||||||
"follow exact_safe_next_action instead of improvising.")
|
"follow exact_safe_next_action instead of improvising.")
|
||||||
|
|
||||||
record_mutation_authority(profile["profile_name"], username, remote if remote in REMOTES else None, task)
|
|
||||||
|
|
||||||
return {
|
return {
|
||||||
"requested_task": task,
|
"requested_task": task,
|
||||||
"required_operation_permission": required_permission,
|
"required_operation_permission": required_permission,
|
||||||
|
|||||||
+1
-24
@@ -24,7 +24,7 @@ venv_python = os.path.join(PROJECT_ROOT, "venv", "bin", "python3")
|
|||||||
if os.path.exists(venv_python) and sys.executable != venv_python:
|
if os.path.exists(venv_python) and sys.executable != venv_python:
|
||||||
os.execv(venv_python, [venv_python] + sys.argv)
|
os.execv(venv_python, [venv_python] + sys.argv)
|
||||||
|
|
||||||
from gitea_auth import get_auth_header, resolve_remote, add_remote_args, api_request, repo_api_url, get_profile
|
from gitea_auth import get_auth_header, resolve_remote, add_remote_args, api_request, repo_api_url
|
||||||
|
|
||||||
|
|
||||||
def main(argv=None):
|
def main(argv=None):
|
||||||
@@ -60,29 +60,6 @@ def main(argv=None):
|
|||||||
|
|
||||||
host, org, repo = resolve_remote(args)
|
host, org, repo = resolve_remote(args)
|
||||||
|
|
||||||
# ── Mutation Authority context wall check (Issue #194) ──
|
|
||||||
LOCK_FILE = "/tmp/gitea_mutation_authority.lock"
|
|
||||||
|
|
||||||
if os.path.exists(LOCK_FILE):
|
|
||||||
try:
|
|
||||||
with open(LOCK_FILE, "r", encoding="utf-8") as f:
|
|
||||||
lock_data = json.load(f)
|
|
||||||
|
|
||||||
# Resolve current CLI profile
|
|
||||||
cli_profile = get_profile().get("profile_name")
|
|
||||||
locked_profile = lock_data.get("current_profile")
|
|
||||||
|
|
||||||
if cli_profile != locked_profile:
|
|
||||||
print(
|
|
||||||
f"Mismatched active profile vs mutation profile (CLI override rejected): "
|
|
||||||
f"CLI profile '{cli_profile}' does not match locked active profile '{locked_profile}' (fail closed)",
|
|
||||||
file=sys.stderr
|
|
||||||
)
|
|
||||||
return 3
|
|
||||||
except Exception as e:
|
|
||||||
print(f"Mutation authority check failed: {e}", file=sys.stderr)
|
|
||||||
return 3
|
|
||||||
|
|
||||||
body = args.body
|
body = args.body
|
||||||
if args.body_file:
|
if args.body_file:
|
||||||
if args.body_file == "-":
|
if args.body_file == "-":
|
||||||
|
|||||||
+116
-138
@@ -202,13 +202,15 @@ def assess_validation_report(report):
|
|||||||
|
|
||||||
*report* keys: ``command``, ``output_read``, ``result`` ('pass'/'fail'),
|
*report* keys: ``command``, ``output_read``, ``result`` ('pass'/'fail'),
|
||||||
``passed``/``failed``/``skipped`` counts, ``ignored_paths`` (each with a
|
``passed``/``failed``/``skipped`` counts, ``ignored_paths`` (each with a
|
||||||
``justification``), ``canonical_command``, ``deviation_justification``.
|
``justification``), ``canonical_command``, ``deviation_justification``,
|
||||||
|
``is_stdout_capture_fix``, ``normal_pytest_summary``.
|
||||||
|
|
||||||
Verdicts:
|
Verdicts:
|
||||||
- 'invalid' — the result may not be claimed at all (no command stated,
|
- 'invalid' — the result may not be claimed at all (no command stated,
|
||||||
or the command output was never read).
|
or the command output was never read).
|
||||||
- 'weak' — claimable but downgraded (missing counts, unjustified
|
- 'weak' — claimable but downgraded (missing counts, unjustified
|
||||||
ignored paths, unexplained deviation from the canonical command).
|
ignored paths, unexplained deviation from the canonical command, or
|
||||||
|
missing normal summary on stdout capture fix).
|
||||||
- 'strong' — full evidence.
|
- 'strong' — full evidence.
|
||||||
"""
|
"""
|
||||||
reasons = []
|
reasons = []
|
||||||
@@ -252,6 +254,14 @@ def assess_validation_report(report):
|
|||||||
"command and the deviation is not justified"
|
"command and the deviation is not justified"
|
||||||
)
|
)
|
||||||
|
|
||||||
|
# For stdout-capture fixes (e.g. Issue #178)
|
||||||
|
if report.get("is_stdout_capture_fix") is True:
|
||||||
|
if not report.get("normal_pytest_summary"):
|
||||||
|
reasons.append(
|
||||||
|
"stdout-capture fix validation is missing normal pytest summary output "
|
||||||
|
"or still requires junitxml workaround"
|
||||||
|
)
|
||||||
|
|
||||||
verdict = "strong" if not reasons else "weak"
|
verdict = "strong" if not reasons else "weak"
|
||||||
return {"verdict": verdict, "claimable": True, "reasons": reasons}
|
return {"verdict": verdict, "claimable": True, "reasons": reasons}
|
||||||
|
|
||||||
@@ -332,7 +342,8 @@ def assess_self_review_contamination(reviewer_identity, pr_author,
|
|||||||
|
|
||||||
def build_final_report(checkout_proof, inventory, validation, contamination,
|
def build_final_report(checkout_proof, inventory, validation, contamination,
|
||||||
identity_eligible, merge_performed,
|
identity_eligible, merge_performed,
|
||||||
issue_status_verified):
|
issue_status_verified, controller_handoff=None,
|
||||||
|
capability_proof=None, sweep_proof=None):
|
||||||
"""Required behavior 6 + acceptance criteria: one report, distinct proofs.
|
"""Required behavior 6 + acceptance criteria: one report, distinct proofs.
|
||||||
|
|
||||||
Combines the individual proof verdicts into the final-report fields the
|
Combines the individual proof verdicts into the final-report fields the
|
||||||
@@ -370,6 +381,19 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
|
|||||||
if not issue_status_verified:
|
if not issue_status_verified:
|
||||||
downgrade_reasons.append("linked issue status not verified")
|
downgrade_reasons.append("linked issue status not verified")
|
||||||
|
|
||||||
|
handoff = assess_controller_handoff(controller_handoff or "")
|
||||||
|
if not handoff.get("present"):
|
||||||
|
downgrade_reasons.append("Controller Handoff section missing or incomplete")
|
||||||
|
downgrade_reasons.extend(handoff.get("reasons", []))
|
||||||
|
|
||||||
|
if capability_proof and not capability_proof.get("proven"):
|
||||||
|
downgrade_reasons.append("capability proof verification failed")
|
||||||
|
downgrade_reasons.extend(capability_proof.get("reasons", []))
|
||||||
|
|
||||||
|
if sweep_proof and not sweep_proof.get("proven"):
|
||||||
|
downgrade_reasons.append("secret/provenance sweep proof verification failed")
|
||||||
|
downgrade_reasons.extend(sweep_proof.get("reasons", []))
|
||||||
|
|
||||||
merge_allowed = (
|
merge_allowed = (
|
||||||
identity_eligible
|
identity_eligible
|
||||||
and checkout_proven
|
and checkout_proven
|
||||||
@@ -408,157 +432,111 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
|
|||||||
"merge_allowed": merge_allowed,
|
"merge_allowed": merge_allowed,
|
||||||
"merge_performed": bool(merge_performed),
|
"merge_performed": bool(merge_performed),
|
||||||
"issue_status_verified": bool(issue_status_verified),
|
"issue_status_verified": bool(issue_status_verified),
|
||||||
|
"controller_handoff_present": handoff.get("present", False),
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
# ── Controller Handoff validation (Issue #182) ────────────────────────────────
|
def assess_controller_handoff(report_text: str) -> dict:
|
||||||
#
|
"""Required for author and reviewer final reports (#183).
|
||||||
# Every final report must end with a compact section titled exactly
|
|
||||||
# "Controller Handoff". Each required field is a (canonical name, aliases)
|
|
||||||
# pair; a field counts as present when any alias starts a bullet/label line
|
|
||||||
# inside the handoff section.
|
|
||||||
|
|
||||||
HANDOFF_HEADING = "Controller Handoff"
|
The report must contain an exactly titled 'Controller Handoff' section
|
||||||
|
(compact format preferred). Missing it downgrades the report.
|
||||||
HANDOFF_BASE_FIELDS = (
|
|
||||||
("Task", ("task",)),
|
|
||||||
("Repo", ("repo", "repository", "repo/state")),
|
|
||||||
("Role", ("role",)),
|
|
||||||
("Identity", ("identity",)),
|
|
||||||
("Issue/PR", ("issue/pr", "issues/prs", "issue", "pr")),
|
|
||||||
("Branch/SHA", ("branch/sha", "branch", "head sha")),
|
|
||||||
("Files changed", ("files changed", "changed", "files")),
|
|
||||||
("Validation", ("validation",)),
|
|
||||||
("Mutations", ("mutations",)),
|
|
||||||
("Current status", ("current status", "status")),
|
|
||||||
("Blockers", ("blockers",)),
|
|
||||||
("Next", ("next",)),
|
|
||||||
("Safety", ("safety",)),
|
|
||||||
)
|
|
||||||
|
|
||||||
HANDOFF_ROLE_FIELDS = {
|
|
||||||
"review": (
|
|
||||||
("Selected PR", ("selected pr",)),
|
|
||||||
("Reviewer eligibility", ("reviewer eligibility", "eligibility")),
|
|
||||||
("Pinned reviewed head", ("pinned reviewed head", "pinned head")),
|
|
||||||
("Review decision", ("review decision", "decision")),
|
|
||||||
("Merge result", ("merge result",)),
|
|
||||||
("Linked issue status", ("linked issue status", "linked issue")),
|
|
||||||
("Cleanup status", ("cleanup status", "cleanup")),
|
|
||||||
),
|
|
||||||
"author": (
|
|
||||||
("Selected issue", ("selected issue",)),
|
|
||||||
("Claim/comment status", ("claim/comment status", "claim status",
|
|
||||||
"claim")),
|
|
||||||
("PR number opened", ("pr number opened", "pr opened", "pr number")),
|
|
||||||
("No review/merge confirmation", ("no review/merge",
|
|
||||||
"no review or merge")),
|
|
||||||
),
|
|
||||||
"inventory": (
|
|
||||||
("Repositories checked", ("repositories checked", "repos checked")),
|
|
||||||
("Open PR counts", ("open pr counts", "open pr count",
|
|
||||||
"open prs per repo")),
|
|
||||||
("Selected PR or reason", ("selected pr", "none selected",
|
|
||||||
"reason none selected")),
|
|
||||||
("Inventory completeness", ("inventory complete", "inventory scoped",
|
|
||||||
"inventory completeness")),
|
|
||||||
),
|
|
||||||
}
|
|
||||||
|
|
||||||
|
|
||||||
def _handoff_section_lines(report_text):
|
|
||||||
"""Return the lines of the Controller Handoff section, or None."""
|
|
||||||
lines = (report_text or "").splitlines()
|
|
||||||
start = None
|
|
||||||
for i, line in enumerate(lines):
|
|
||||||
bare = line.strip().lstrip("#").strip().rstrip(":")
|
|
||||||
if bare == HANDOFF_HEADING:
|
|
||||||
start = i + 1
|
|
||||||
break
|
|
||||||
if start is None:
|
|
||||||
return None
|
|
||||||
return lines[start:]
|
|
||||||
|
|
||||||
|
|
||||||
def assess_controller_handoff(report_text, role=None):
|
|
||||||
"""Issue #182: final reports without a Controller Handoff downgrade.
|
|
||||||
|
|
||||||
Verdicts:
|
|
||||||
- 'missing' — no exactly-titled section; the report is downgraded.
|
|
||||||
- 'incomplete' — section present but required fields absent (listed).
|
|
||||||
- 'complete' — all base fields plus the role-specific fields present.
|
|
||||||
|
|
||||||
*role* is 'review', 'author', 'inventory', or None (base fields only).
|
|
||||||
The handoff supplements the full report; this helper never validates
|
|
||||||
the full report body, only the continuation summary.
|
|
||||||
"""
|
"""
|
||||||
section = _handoff_section_lines(report_text)
|
if not report_text:
|
||||||
if section is None:
|
return {"present": False, "reasons": ["no report text"]}
|
||||||
|
|
||||||
|
text = str(report_text)
|
||||||
|
if "Controller Handoff" not in text:
|
||||||
return {
|
return {
|
||||||
"verdict": "missing",
|
"present": False,
|
||||||
"downgraded": True,
|
|
||||||
"missing_fields": [name for name, _ in HANDOFF_BASE_FIELDS],
|
|
||||||
"reasons": [
|
"reasons": [
|
||||||
"final report has no section titled exactly "
|
"final report missing exactly-titled 'Controller Handoff' section"
|
||||||
f"'{HANDOFF_HEADING}'"
|
|
||||||
],
|
],
|
||||||
}
|
}
|
||||||
|
return {"present": True, "reasons": []}
|
||||||
|
|
||||||
labels = []
|
|
||||||
for line in section:
|
|
||||||
stripped = line.strip().lstrip("-*").strip()
|
|
||||||
if ":" in stripped:
|
|
||||||
labels.append(stripped.split(":", 1)[0].strip().lower())
|
|
||||||
|
|
||||||
required = list(HANDOFF_BASE_FIELDS)
|
def assess_capability_proof(resolved_capabilities: dict) -> dict:
|
||||||
required.extend(HANDOFF_ROLE_FIELDS.get(role or "", ()))
|
"""Required behavior: every mutation must have exact capability proof.
|
||||||
|
|
||||||
missing = []
|
If a mutation task is unknown, unresolved, or lacks explicit resolver evidence,
|
||||||
for name, aliases in required:
|
the workflow must fail closed / be downgraded.
|
||||||
if not any(label.startswith(alias)
|
"""
|
||||||
for label in labels for alias in aliases):
|
reasons = []
|
||||||
missing.append(name)
|
if not resolved_capabilities:
|
||||||
|
|
||||||
if missing:
|
|
||||||
return {
|
return {
|
||||||
"verdict": "incomplete",
|
"proven": False,
|
||||||
"downgraded": True,
|
"reasons": ["no capability proof resolved; fail closed"],
|
||||||
"missing_fields": missing,
|
|
||||||
"reasons": [f"handoff missing required field: {m}"
|
|
||||||
for m in missing],
|
|
||||||
}
|
}
|
||||||
|
for task, cap in resolved_capabilities.items():
|
||||||
|
allowed = cap.get("allowed_in_current_session")
|
||||||
|
op = cap.get("required_operation_permission")
|
||||||
|
if not allowed:
|
||||||
|
reasons.append(f"task '{task}' requires permission '{op}' which is not allowed in current session")
|
||||||
|
# If the resolved result indicates an unknown task or missing flag
|
||||||
|
if "unknown" in str(cap.get("requested_task", "")).lower() or cap.get("allowed_in_current_session") is None:
|
||||||
|
reasons.append(f"task '{task}' could not be resolved to a known capability")
|
||||||
|
proven = not reasons
|
||||||
|
return {"proven": proven, "reasons": reasons}
|
||||||
|
|
||||||
# Validate issue/PR references for exact number and no forbidden terms (Issue #194 / #196)
|
|
||||||
fields_dict = {}
|
|
||||||
for line in section:
|
|
||||||
stripped = line.strip().lstrip("-*").strip()
|
|
||||||
if ":" in stripped:
|
|
||||||
k, v = stripped.split(":", 1)
|
|
||||||
fields_dict[k.strip().lower()] = v.strip()
|
|
||||||
|
|
||||||
for alias in ("selected issue", "pr number opened", "pr opened", "pr number", "selected pr"):
|
def assess_secret_sweep(sweep_report: dict) -> dict:
|
||||||
val = fields_dict.get(alias)
|
"""Required behavior: secret/provenance sweeps must state exact command/method and scope.
|
||||||
if val:
|
|
||||||
numbers = re.findall(r"\d+", val)
|
|
||||||
has_forbidden = any(term in val.lower() for term in ("equivalent", "related", "same", "/"))
|
|
||||||
if len(numbers) != 1 or has_forbidden:
|
|
||||||
field_name = "Selected issue/PR"
|
|
||||||
for name, aliases in list(HANDOFF_BASE_FIELDS) + list(HANDOFF_ROLE_FIELDS.get(role or "", ())):
|
|
||||||
if alias in aliases:
|
|
||||||
field_name = name
|
|
||||||
break
|
|
||||||
return {
|
|
||||||
"verdict": "incomplete",
|
|
||||||
"downgraded": True,
|
|
||||||
"missing_fields": [field_name],
|
|
||||||
"reasons": [
|
|
||||||
f"{field_name} must specify exactly one number and no ambiguous references (got: '{val}')"
|
|
||||||
],
|
|
||||||
}
|
|
||||||
|
|
||||||
|
*sweep_report* keys: ``method``, ``scope``, ``clean`` (bool).
|
||||||
|
Returns dict with ``proven`` (bool), ``verdict`` ('strong', 'weak', 'invalid'), and ``reasons`` list.
|
||||||
|
"""
|
||||||
|
if not sweep_report:
|
||||||
|
return {
|
||||||
|
"proven": False,
|
||||||
|
"verdict": "invalid",
|
||||||
|
"reasons": ["no secret/provenance sweep report provided"],
|
||||||
|
}
|
||||||
|
reasons = []
|
||||||
|
method = (sweep_report.get("method") or "").strip()
|
||||||
|
scope = (sweep_report.get("scope") or "").strip()
|
||||||
|
if not method:
|
||||||
|
reasons.append("secret sweep report missing exact scan command, pattern, or method")
|
||||||
|
if not scope:
|
||||||
|
reasons.append("secret sweep report missing scanned scope")
|
||||||
|
if sweep_report.get("clean") is not True:
|
||||||
|
reasons.append("secret sweep report did not confirm diff is clean")
|
||||||
|
|
||||||
|
verdict = "weak" if reasons else "strong"
|
||||||
|
proven = not reasons and sweep_report.get("clean") is True
|
||||||
return {
|
return {
|
||||||
"verdict": "complete",
|
"proven": proven,
|
||||||
"downgraded": False,
|
"verdict": verdict,
|
||||||
"missing_fields": [],
|
"reasons": reasons,
|
||||||
"reasons": [],
|
}
|
||||||
|
|
||||||
|
|
||||||
|
def assess_author_pr_report(pr_report: dict) -> dict:
|
||||||
|
"""Required behavior: author PR creation report must include PR number, branch, and exact head SHA.
|
||||||
|
|
||||||
|
*pr_report* keys: ``pr_number`` (int), ``branch`` (str), ``head_sha`` (str).
|
||||||
|
"""
|
||||||
|
if not pr_report:
|
||||||
|
return {
|
||||||
|
"complete": False,
|
||||||
|
"reasons": ["no PR report provided"],
|
||||||
|
}
|
||||||
|
reasons = []
|
||||||
|
pr_number = pr_report.get("pr_number")
|
||||||
|
branch = (pr_report.get("branch") or "").strip()
|
||||||
|
head_sha = (pr_report.get("head_sha") or "").strip()
|
||||||
|
|
||||||
|
if not isinstance(pr_number, int) or pr_number <= 0:
|
||||||
|
reasons.append("PR number is missing or invalid")
|
||||||
|
if not branch:
|
||||||
|
reasons.append("PR branch name is missing")
|
||||||
|
if not head_sha:
|
||||||
|
reasons.append("PR head SHA is missing")
|
||||||
|
elif not _FULL_SHA.match(head_sha):
|
||||||
|
reasons.append("PR head SHA is not a full 40-hex commit SHA")
|
||||||
|
|
||||||
|
complete = not reasons
|
||||||
|
return {
|
||||||
|
"complete": complete,
|
||||||
|
"reasons": reasons,
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -40,16 +40,6 @@ start_ref="${2:-prgs/master}"
|
|||||||
|
|
||||||
# Enforce issue-linked, traceable branch names (issue → branch → worktree → PR).
|
# Enforce issue-linked, traceable branch names (issue → branch → worktree → PR).
|
||||||
if [[ "$allow_unlinked" -eq 0 ]]; then
|
if [[ "$allow_unlinked" -eq 0 ]]; then
|
||||||
if [[ ! -f "/tmp/gitea_issue_lock.json" ]]; then
|
|
||||||
echo "Error: Issue lock file '/tmp/gitea_issue_lock.json' is missing. You must lock exactly one issue before branch creation (fail closed)." >&2
|
|
||||||
exit 2
|
|
||||||
fi
|
|
||||||
locked_branch=$(python3 -c "import json; print(json.load(open('/tmp/gitea_issue_lock.json')).get('branch_name', ''))")
|
|
||||||
if [[ "$branch" != "$locked_branch" ]]; then
|
|
||||||
echo "Error: Requested branch '$branch' does not match locked branch '$locked_branch' (fail closed)." >&2
|
|
||||||
exit 2
|
|
||||||
fi
|
|
||||||
|
|
||||||
if [[ "$branch" =~ ^(fix|feat|docs|chore)/issue-[0-9]+-.+ ]] \
|
if [[ "$branch" =~ ^(fix|feat|docs|chore)/issue-[0-9]+-.+ ]] \
|
||||||
|| [[ "$branch" =~ ^review/pr-[0-9]+-.+ ]]; then
|
|| [[ "$branch" =~ ^review/pr-[0-9]+-.+ ]]; then
|
||||||
:
|
:
|
||||||
|
|||||||
@@ -287,52 +287,33 @@ Ready-to-copy templates live in [`templates/`](templates/):
|
|||||||
|
|
||||||
## K. Controller Handoff (required, every task)
|
## K. Controller Handoff (required, every task)
|
||||||
|
|
||||||
Every LLM task **must end with a `Controller Handoff`** — whether the
|
Every LLM task **must end with a `Controller Handoff`** (exact title) — whether the
|
||||||
task was implementation, review, merge, issue triage, documentation,
|
task was implementation, review, merge, issue triage, documentation,
|
||||||
discussion-only, or blocked planning. It lets a controller LLM understand the
|
discussion-only, or blocked planning. It lets a controller LLM understand the
|
||||||
current state immediately, without rereading the conversation.
|
current state immediately, without rereading the conversation.
|
||||||
|
|
||||||
|
The section title must be exactly "Controller Handoff" (or "Controller Handoff Summary" for long form). Reports without it are downgraded (see review_proofs.assess_controller_handoff).
|
||||||
|
|
||||||
**The compact format is the default.** It is written for controller-LLM
|
**The compact format is the default.** It is written for controller-LLM
|
||||||
readability, not as a full human status report. PR bodies still carry the
|
readability, not as a full human status report. PR bodies still carry the
|
||||||
full review detail — the handoff never replaces PR documentation.
|
full review detail — the handoff never replaces PR documentation.
|
||||||
|
|
||||||
Compact format (default, canonical field set per issue #182):
|
Compact format (default):
|
||||||
|
|
||||||
```md
|
```md
|
||||||
## Controller Handoff
|
## Controller Handoff
|
||||||
|
|
||||||
- Task:
|
- Task:
|
||||||
- Repo:
|
- Repo/state:
|
||||||
- Role:
|
- Issues/PRs:
|
||||||
- Identity:
|
- Changed:
|
||||||
- Issue/PR:
|
|
||||||
- Branch/SHA:
|
|
||||||
- Files changed:
|
|
||||||
- Validation:
|
- Validation:
|
||||||
- Mutations:
|
|
||||||
- Current status:
|
|
||||||
- Blockers:
|
- Blockers:
|
||||||
|
- Review:
|
||||||
- Next:
|
- Next:
|
||||||
- Safety:
|
- Safety:
|
||||||
```
|
```
|
||||||
|
|
||||||
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:`
|
|
||||||
- author tasks: `Selected issue:`, `Claim/comment status:`,
|
|
||||||
`PR number opened:`, `No review/merge:` (explicit confirmation)
|
|
||||||
- queue/inventory tasks: `Repositories checked:`, `Open PR counts:`,
|
|
||||||
`Selected PR or reason none selected:`, `Inventory completeness:`
|
|
||||||
|
|
||||||
The section title must be exactly `Controller Handoff`.
|
|
||||||
`review_proofs.assess_controller_handoff()` validates this section; reports
|
|
||||||
missing it (or missing required fields) are downgraded. The handoff never
|
|
||||||
replaces the full report — it is the compact continuation summary at the end,
|
|
||||||
and the full report must still carry exact validation results and mutation
|
|
||||||
confirmation.
|
|
||||||
|
|
||||||
The `Safety:` line is never omitted; it is usually:
|
The `Safety:` line is never omitted; it is usually:
|
||||||
|
|
||||||
```text
|
```text
|
||||||
|
|||||||
@@ -35,11 +35,5 @@ Then run the cleanup template (worktree-cleanup.md):
|
|||||||
- delete remote branch, remove local branch + worktree folder
|
- delete remote branch, remove local branch + worktree folder
|
||||||
- fetch/prune; confirm main checkout is clean and current (0 0).
|
- fetch/prune; confirm main checkout is clean and current (0 0).
|
||||||
|
|
||||||
Handoff: end with a section titled exactly `Controller Handoff` per SKILL.md
|
Handoff: reviewer identity, merge result + commit, cleanup done, issue closed, PR metadata state/merged flag/hash, remote master hash, post-merge verification method used & verification results.
|
||||||
§K (long form — a merge is always high-risk), including the review/merge role
|
|
||||||
fields (Selected PR, Reviewer eligibility, Pinned reviewed head, Review
|
|
||||||
decision, Merge result, Linked issue status, Cleanup status) plus: merge
|
|
||||||
commit, PR metadata state/merged flag/hash, remote master hash, and the
|
|
||||||
post-merge verification method used & verification results. Reports missing
|
|
||||||
the handoff are downgraded (review_proofs.assess_controller_handoff).
|
|
||||||
```
|
```
|
||||||
|
|||||||
@@ -65,10 +65,7 @@ Steps:
|
|||||||
- MCP-Profile: <profile name>
|
- MCP-Profile: <profile name>
|
||||||
- Eligibility: passed/failed
|
- Eligibility: passed/failed
|
||||||
|
|
||||||
Handoff: end with a section titled exactly `Controller Handoff` per SKILL.md
|
Handoff: reviewer identity, PR author, scope verdict, checks + results, decision —
|
||||||
§K (compact by default; long form if a merge happened or a gate blocked you),
|
formatted per SKILL.md §K (compact by default; long form if a merge happened
|
||||||
including the review/merge role fields: Selected PR, Reviewer eligibility,
|
or a gate blocked you); if you could not merge, name the exact gate.
|
||||||
Pinned reviewed head, Review decision, Merge result, Linked issue status,
|
|
||||||
Cleanup status. If you could not merge, name the exact gate. Reports missing
|
|
||||||
the handoff are downgraded (review_proofs.assess_controller_handoff).
|
|
||||||
```
|
```
|
||||||
|
|||||||
@@ -42,10 +42,7 @@ Steps:
|
|||||||
- Self-review allowed: no
|
- Self-review allowed: no
|
||||||
9. Stop before review/merge — you are the author.
|
9. Stop before review/merge — you are the author.
|
||||||
|
|
||||||
Handoff: end with a section titled exactly `Controller Handoff` per SKILL.md
|
Handoff: issue #, branch, worktree path, files changed, checks + results, PR URL —
|
||||||
§K (compact; long form only on the high-risk triggers), including the author
|
formatted as the compact Controller Handoff (SKILL.md §K; long form only on
|
||||||
role fields: Selected issue, Claim/comment status, PR number opened, and an
|
the high-risk triggers); Review line: "Review needed — PR is open".
|
||||||
explicit "No review/merge:" confirmation — plus branch, worktree path, files
|
|
||||||
changed, checks + results. Next line: "Review needed — PR is open". Reports
|
|
||||||
missing the handoff are downgraded (review_proofs.assess_controller_handoff).
|
|
||||||
```
|
```
|
||||||
|
|||||||
+4
-183
@@ -33,16 +33,10 @@ from mcp_server import ( # noqa: E402
|
|||||||
gitea_submit_pr_review,
|
gitea_submit_pr_review,
|
||||||
gitea_list_issue_comments,
|
gitea_list_issue_comments,
|
||||||
gitea_create_issue_comment,
|
gitea_create_issue_comment,
|
||||||
gitea_lock_issue,
|
|
||||||
)
|
)
|
||||||
from gitea_auth import get_profile # noqa: E402
|
from gitea_auth import get_profile # noqa: E402
|
||||||
import gitea_config # noqa: E402
|
import gitea_config # noqa: E402
|
||||||
|
|
||||||
import mcp_server
|
|
||||||
# Globally disable verification check for existing isolated unit tests
|
|
||||||
_real_verify = mcp_server.verify_mutation_authority
|
|
||||||
mcp_server.verify_mutation_authority = lambda *args, **kwargs: None
|
|
||||||
|
|
||||||
FAKE_AUTH = "Basic dGVzdDp0ZXN0"
|
FAKE_AUTH = "Basic dGVzdDp0ZXN0"
|
||||||
|
|
||||||
|
|
||||||
@@ -91,13 +85,10 @@ class TestCreatePR(unittest.TestCase):
|
|||||||
|
|
||||||
@patch("mcp_server.api_request")
|
@patch("mcp_server.api_request")
|
||||||
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
||||||
@patch("os.path.exists", return_value=True)
|
def test_creates_pr(self, _auth, mock_api):
|
||||||
@patch("builtins.open")
|
|
||||||
def test_creates_pr(self, mock_open, mock_exists, _auth, mock_api):
|
|
||||||
mock_open.return_value.__enter__.return_value.read.return_value = '{"issue_number": 123, "branch_name": "feat/x"}'
|
|
||||||
mock_api.return_value = {"number": 3, "html_url": "https://example.com/pulls/3"}
|
mock_api.return_value = {"number": 3, "html_url": "https://example.com/pulls/3"}
|
||||||
with patch.dict(os.environ, {}, clear=True):
|
with patch.dict(os.environ, {}, clear=True):
|
||||||
result = gitea_create_pr(title="feat: X Closes #123", head="feat/x", base="main")
|
result = gitea_create_pr(title="feat: X", head="feat/x", base="main")
|
||||||
self.assertEqual(result["number"], 3)
|
self.assertEqual(result["number"], 3)
|
||||||
self.assertNotIn("url", result)
|
self.assertNotIn("url", result)
|
||||||
payload = mock_api.call_args[0][3]
|
payload = mock_api.call_args[0][3]
|
||||||
@@ -106,13 +97,10 @@ class TestCreatePR(unittest.TestCase):
|
|||||||
|
|
||||||
@patch("mcp_server.api_request")
|
@patch("mcp_server.api_request")
|
||||||
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
||||||
@patch("os.path.exists", return_value=True)
|
def test_create_pr_reveal_opt_in_includes_url(self, _auth, mock_api):
|
||||||
@patch("builtins.open")
|
|
||||||
def test_create_pr_reveal_opt_in_includes_url(self, mock_open, mock_exists, _auth, mock_api):
|
|
||||||
mock_open.return_value.__enter__.return_value.read.return_value = '{"issue_number": 123, "branch_name": "feat/x"}'
|
|
||||||
mock_api.return_value = {"number": 3, "html_url": "https://example.com/pulls/3"}
|
mock_api.return_value = {"number": 3, "html_url": "https://example.com/pulls/3"}
|
||||||
with patch.dict(os.environ, {"GITEA_MCP_REVEAL_ENDPOINTS": "1"}, clear=True):
|
with patch.dict(os.environ, {"GITEA_MCP_REVEAL_ENDPOINTS": "1"}, clear=True):
|
||||||
result = gitea_create_pr(title="feat: X Closes #123", head="feat/x", base="main")
|
result = gitea_create_pr(title="feat: X", head="feat/x", base="main")
|
||||||
self.assertIn("pulls/3", result["url"])
|
self.assertIn("pulls/3", result["url"])
|
||||||
|
|
||||||
|
|
||||||
@@ -2308,170 +2296,3 @@ class TestIssueCommentPermissionSeparation(unittest.TestCase):
|
|||||||
"gitea.issue.comment", reviewer["allowed_operations"],
|
"gitea.issue.comment", reviewer["allowed_operations"],
|
||||||
reviewer.get("forbidden_operations", []))
|
reviewer.get("forbidden_operations", []))
|
||||||
self.assertTrue(ok)
|
self.assertTrue(ok)
|
||||||
|
|
||||||
|
|
||||||
class TestVerifyMutationAuthority(unittest.TestCase):
|
|
||||||
"""Test verification lock logic under various configurations."""
|
|
||||||
|
|
||||||
def setUp(self):
|
|
||||||
self.patch_profile = patch("mcp_server.get_profile")
|
|
||||||
self.mock_profile = self.patch_profile.start()
|
|
||||||
self.patch_username = patch("mcp_server._authenticated_username")
|
|
||||||
self.mock_username = self.patch_username.start()
|
|
||||||
|
|
||||||
# Restore real function for these tests
|
|
||||||
self._old_verify = mcp_server.verify_mutation_authority
|
|
||||||
mcp_server.verify_mutation_authority = _real_verify
|
|
||||||
|
|
||||||
def tearDown(self):
|
|
||||||
self.patch_profile.stop()
|
|
||||||
self.patch_username.stop()
|
|
||||||
mcp_server.verify_mutation_authority = self._old_verify
|
|
||||||
# Clean up lock file
|
|
||||||
if os.path.exists("/tmp/gitea_mutation_authority.lock"):
|
|
||||||
os.remove("/tmp/gitea_mutation_authority.lock")
|
|
||||||
|
|
||||||
def test_missing_lock_fails_closed(self):
|
|
||||||
if os.path.exists("/tmp/gitea_mutation_authority.lock"):
|
|
||||||
os.remove("/tmp/gitea_mutation_authority.lock")
|
|
||||||
with self.assertRaises(RuntimeError) as ctx:
|
|
||||||
_real_verify("prgs")
|
|
||||||
self.assertIn("lock is missing", str(ctx.exception))
|
|
||||||
|
|
||||||
def test_mismatched_remote_fails(self):
|
|
||||||
with open("/tmp/gitea_mutation_authority.lock", "w", encoding="utf-8") as f:
|
|
||||||
json.dump({
|
|
||||||
"remote": "dadeschools",
|
|
||||||
"current_profile": "prgs-reviewer",
|
|
||||||
"current_identity": "sysadmin"
|
|
||||||
}, f)
|
|
||||||
with self.assertRaises(RuntimeError) as ctx:
|
|
||||||
_real_verify("prgs")
|
|
||||||
self.assertIn("does not match locked remote", str(ctx.exception))
|
|
||||||
|
|
||||||
def test_mismatched_profile_fails(self):
|
|
||||||
with open("/tmp/gitea_mutation_authority.lock", "w", encoding="utf-8") as f:
|
|
||||||
json.dump({
|
|
||||||
"remote": "prgs",
|
|
||||||
"current_profile": "prgs-author",
|
|
||||||
"current_identity": "jcwalker3"
|
|
||||||
}, f)
|
|
||||||
self.mock_profile.return_value = {"profile_name": "prgs-reviewer"}
|
|
||||||
self.mock_username.return_value = "sysadmin"
|
|
||||||
|
|
||||||
with self.assertRaises(RuntimeError) as ctx:
|
|
||||||
_real_verify("prgs")
|
|
||||||
self.assertIn("does not match locked authority", str(ctx.exception))
|
|
||||||
|
|
||||||
def test_author_to_reviewer_pivot_blocked_without_authorization(self):
|
|
||||||
with open("/tmp/gitea_mutation_authority.lock", "w", encoding="utf-8") as f:
|
|
||||||
json.dump({
|
|
||||||
"remote": "prgs",
|
|
||||||
"initial_profile": "prgs-author",
|
|
||||||
"initial_identity": "jcwalker3",
|
|
||||||
"current_profile": "prgs-reviewer",
|
|
||||||
"current_identity": "sysadmin",
|
|
||||||
"role_pivot_authorized": False
|
|
||||||
}, f)
|
|
||||||
self.mock_profile.return_value = {"profile_name": "prgs-reviewer"}
|
|
||||||
self.mock_username.return_value = "sysadmin"
|
|
||||||
|
|
||||||
with self.assertRaises(RuntimeError) as ctx:
|
|
||||||
_real_verify("prgs", required_role="reviewer")
|
|
||||||
self.assertIn("without authorized role pivot", str(ctx.exception))
|
|
||||||
|
|
||||||
def test_allowed_when_match(self):
|
|
||||||
with open("/tmp/gitea_mutation_authority.lock", "w", encoding="utf-8") as f:
|
|
||||||
json.dump({
|
|
||||||
"remote": "prgs",
|
|
||||||
"initial_profile": "prgs-reviewer",
|
|
||||||
"initial_identity": "sysadmin",
|
|
||||||
"current_profile": "prgs-reviewer",
|
|
||||||
"current_identity": "sysadmin",
|
|
||||||
"role_pivot_authorized": False
|
|
||||||
}, f)
|
|
||||||
self.mock_profile.return_value = {"profile_name": "prgs-reviewer"}
|
|
||||||
self.mock_username.return_value = "sysadmin"
|
|
||||||
|
|
||||||
# Should pass without exception
|
|
||||||
_real_verify("prgs")
|
|
||||||
|
|
||||||
|
|
||||||
class TestIssueLocking(unittest.TestCase):
|
|
||||||
"""Test issue locking and PR gating constraints."""
|
|
||||||
|
|
||||||
def tearDown(self):
|
|
||||||
if os.path.exists("/tmp/gitea_issue_lock.json"):
|
|
||||||
os.remove("/tmp/gitea_issue_lock.json")
|
|
||||||
|
|
||||||
@patch("mcp_server.api_get_all")
|
|
||||||
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
|
||||||
def test_lock_issue_success(self, _auth, mock_api):
|
|
||||||
mock_api.return_value = [] # no open PRs
|
|
||||||
res = gitea_lock_issue(issue_number=196, branch_name="feat/issue-196-mutations", remote="prgs")
|
|
||||||
self.assertTrue(res["success"])
|
|
||||||
self.assertTrue(os.path.exists("/tmp/gitea_issue_lock.json"))
|
|
||||||
|
|
||||||
def test_lock_issue_mismatch_branch_fails(self):
|
|
||||||
with self.assertRaises(ValueError) as ctx:
|
|
||||||
gitea_lock_issue(issue_number=196, branch_name="feat/issue-195-mutations", remote="prgs")
|
|
||||||
self.assertIn("must contain locked issue pattern", str(ctx.exception))
|
|
||||||
|
|
||||||
@patch("mcp_server.api_get_all")
|
|
||||||
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
|
||||||
def test_lock_issue_reused_by_open_pr_branch(self, _auth, mock_api):
|
|
||||||
mock_api.return_value = [{
|
|
||||||
"number": 200,
|
|
||||||
"head": {"ref": "feat/issue-196-boundary"},
|
|
||||||
"title": "Some PR",
|
|
||||||
"body": "No closes ref"
|
|
||||||
}]
|
|
||||||
with self.assertRaises(ValueError) as ctx:
|
|
||||||
gitea_lock_issue(issue_number=196, branch_name="feat/issue-196-mutations", remote="prgs")
|
|
||||||
self.assertIn("already tied to an open PR", str(ctx.exception))
|
|
||||||
|
|
||||||
@patch("mcp_server.api_get_all")
|
|
||||||
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
|
||||||
def test_lock_issue_reused_by_open_pr_closes_ref(self, _auth, mock_api):
|
|
||||||
mock_api.return_value = [{
|
|
||||||
"number": 200,
|
|
||||||
"head": {"ref": "feat/other-branch"},
|
|
||||||
"title": "Some PR",
|
|
||||||
"body": "fixes #196"
|
|
||||||
}]
|
|
||||||
with self.assertRaises(ValueError) as ctx:
|
|
||||||
gitea_lock_issue(issue_number=196, branch_name="feat/issue-196-mutations", remote="prgs")
|
|
||||||
self.assertIn("already tied to an open PR", str(ctx.exception))
|
|
||||||
|
|
||||||
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
|
||||||
def test_create_pr_missing_lock_fails(self, _auth):
|
|
||||||
if os.path.exists("/tmp/gitea_issue_lock.json"):
|
|
||||||
os.remove("/tmp/gitea_issue_lock.json")
|
|
||||||
with self.assertRaises(RuntimeError) as ctx:
|
|
||||||
gitea_create_pr(title="feat: X Closes #196", head="feat/issue-196-mutations", remote="prgs")
|
|
||||||
self.assertIn("Issue lock is missing", str(ctx.exception))
|
|
||||||
|
|
||||||
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
|
||||||
def test_create_pr_branch_mismatch_fails(self, _auth):
|
|
||||||
with open("/tmp/gitea_issue_lock.json", "w", encoding="utf-8") as f:
|
|
||||||
json.dump({"issue_number": 196, "branch_name": "feat/issue-196-mutations"}, f)
|
|
||||||
with self.assertRaises(ValueError) as ctx:
|
|
||||||
gitea_create_pr(title="feat: X Closes #196", head="feat/issue-196-different", remote="prgs")
|
|
||||||
self.assertIn("does not match locked branch", str(ctx.exception))
|
|
||||||
|
|
||||||
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
|
||||||
def test_create_pr_forbidden_terms_fails(self, _auth):
|
|
||||||
with open("/tmp/gitea_issue_lock.json", "w", encoding="utf-8") as f:
|
|
||||||
json.dump({"issue_number": 196, "branch_name": "feat/issue-196-mutations"}, f)
|
|
||||||
for term in ("equivalent to #196", "related to #196", "same as #196"):
|
|
||||||
with self.assertRaises(ValueError) as ctx:
|
|
||||||
gitea_create_pr(title=f"feat: X {term}", head="feat/issue-196-mutations", remote="prgs")
|
|
||||||
self.assertIn("contains forbidden term", str(ctx.exception))
|
|
||||||
|
|
||||||
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
|
||||||
def test_create_pr_missing_closes_ref_fails(self, _auth):
|
|
||||||
with open("/tmp/gitea_issue_lock.json", "w", encoding="utf-8") as f:
|
|
||||||
json.dump({"issue_number": 196, "branch_name": "feat/issue-196-mutations"}, f)
|
|
||||||
with self.assertRaises(ValueError) as ctx:
|
|
||||||
gitea_create_pr(title="feat: X refs #196", head="feat/issue-196-mutations", remote="prgs")
|
|
||||||
self.assertIn("must contain 'Closes #196' or 'Fixes #196' exactly", str(ctx.exception))
|
|
||||||
|
|||||||
@@ -2,8 +2,6 @@
|
|||||||
|
|
||||||
Mocks api_request and credentials.
|
Mocks api_request and credentials.
|
||||||
"""
|
"""
|
||||||
import io
|
|
||||||
import os
|
|
||||||
import sys
|
import sys
|
||||||
import unittest
|
import unittest
|
||||||
from unittest.mock import patch
|
from unittest.mock import patch
|
||||||
@@ -29,11 +27,6 @@ FAKE_PR_DATA = {
|
|||||||
|
|
||||||
class TestArgParsing(unittest.TestCase):
|
class TestArgParsing(unittest.TestCase):
|
||||||
|
|
||||||
def setUp(self):
|
|
||||||
self.exists_patcher = patch("os.path.exists", return_value=False)
|
|
||||||
self.exists_patcher.start()
|
|
||||||
self.addCleanup(self.exists_patcher.stop)
|
|
||||||
|
|
||||||
@patch("review_pr.get_auth_header", return_value=FAKE_CREDS)
|
@patch("review_pr.get_auth_header", return_value=FAKE_CREDS)
|
||||||
def test_missing_pr_number_exits(self, _auth):
|
def test_missing_pr_number_exits(self, _auth):
|
||||||
with self.assertRaises(SystemExit):
|
with self.assertRaises(SystemExit):
|
||||||
@@ -42,11 +35,6 @@ class TestArgParsing(unittest.TestCase):
|
|||||||
|
|
||||||
class TestAPIPayload(unittest.TestCase):
|
class TestAPIPayload(unittest.TestCase):
|
||||||
|
|
||||||
def setUp(self):
|
|
||||||
self.exists_patcher = patch("os.path.exists", return_value=False)
|
|
||||||
self.exists_patcher.start()
|
|
||||||
self.addCleanup(self.exists_patcher.stop)
|
|
||||||
|
|
||||||
@patch("review_pr.api_request")
|
@patch("review_pr.api_request")
|
||||||
@patch("review_pr.get_auth_header", return_value=FAKE_CREDS)
|
@patch("review_pr.get_auth_header", return_value=FAKE_CREDS)
|
||||||
def test_payload_fields_and_workflow(self, _auth, mock_api):
|
def test_payload_fields_and_workflow(self, _auth, mock_api):
|
||||||
@@ -111,70 +99,5 @@ class TestAPIPayload(unittest.TestCase):
|
|||||||
self.assertIn("gitea_merge_pr", msg)
|
self.assertIn("gitea_merge_pr", msg)
|
||||||
|
|
||||||
|
|
||||||
class TestMutationAuthorityLock(unittest.TestCase):
|
|
||||||
"""Issue #194: verify that the CLI tool rejects profile overrides when mismatched with lock."""
|
|
||||||
|
|
||||||
@patch("review_pr.get_profile")
|
|
||||||
def test_cli_blocked_on_profile_mismatch(self, mock_get_profile):
|
|
||||||
mock_get_profile.return_value = {"profile_name": "prgs-reviewer"}
|
|
||||||
|
|
||||||
original_exists = os.path.exists
|
|
||||||
def conditional_exists(path):
|
|
||||||
if "gitea_mutation_authority.lock" in str(path):
|
|
||||||
return True
|
|
||||||
return original_exists(path)
|
|
||||||
|
|
||||||
original_open = open
|
|
||||||
def conditional_open(file, *args, **kwargs):
|
|
||||||
if "gitea_mutation_authority.lock" in str(file):
|
|
||||||
return io.StringIO('{"current_profile": "prgs-author"}')
|
|
||||||
return original_open(file, *args, **kwargs)
|
|
||||||
|
|
||||||
from _pytest.monkeypatch import MonkeyPatch
|
|
||||||
import io
|
|
||||||
buf = io.StringIO()
|
|
||||||
monkeypatch = MonkeyPatch()
|
|
||||||
monkeypatch.setattr(sys, "stderr", buf)
|
|
||||||
|
|
||||||
with patch("os.path.exists", side_effect=conditional_exists), \
|
|
||||||
patch("builtins.open", side_effect=conditional_open):
|
|
||||||
try:
|
|
||||||
rc = review_pr.main([
|
|
||||||
"--pr-number", "81", "--event", "APPROVE",
|
|
||||||
])
|
|
||||||
finally:
|
|
||||||
monkeypatch.undo()
|
|
||||||
|
|
||||||
self.assertEqual(rc, 3)
|
|
||||||
msg = buf.getvalue().lower()
|
|
||||||
self.assertIn("cli override rejected", msg)
|
|
||||||
|
|
||||||
@patch("review_pr.get_profile")
|
|
||||||
@patch("review_pr.api_request")
|
|
||||||
@patch("review_pr.get_auth_header", return_value=FAKE_CREDS)
|
|
||||||
def test_cli_allowed_on_profile_match(self, _auth, mock_api, mock_get_profile):
|
|
||||||
mock_get_profile.return_value = {"profile_name": "prgs-reviewer"}
|
|
||||||
mock_api.side_effect = [FAKE_PR_DATA, {}]
|
|
||||||
|
|
||||||
original_exists = os.path.exists
|
|
||||||
def conditional_exists(path):
|
|
||||||
if "gitea_mutation_authority.lock" in str(path):
|
|
||||||
return True
|
|
||||||
return original_exists(path)
|
|
||||||
|
|
||||||
original_open = open
|
|
||||||
def conditional_open(file, *args, **kwargs):
|
|
||||||
if "gitea_mutation_authority.lock" in str(file):
|
|
||||||
return io.StringIO('{"current_profile": "prgs-reviewer"}')
|
|
||||||
return original_open(file, *args, **kwargs)
|
|
||||||
|
|
||||||
with patch("os.path.exists", side_effect=conditional_exists), \
|
|
||||||
patch("builtins.open", side_effect=conditional_open):
|
|
||||||
rc = review_pr.main([
|
|
||||||
"--pr-number", "81", "--event", "APPROVE",
|
|
||||||
])
|
|
||||||
self.assertEqual(rc, 0)
|
|
||||||
|
|
||||||
|
|
||||||
if __name__ == "__main__":
|
if __name__ == "__main__":
|
||||||
unittest.main()
|
unittest.main()
|
||||||
|
|||||||
+122
-121
@@ -21,8 +21,11 @@ import unittest
|
|||||||
sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
|
sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
|
||||||
|
|
||||||
from review_proofs import ( # noqa: E402
|
from review_proofs import ( # noqa: E402
|
||||||
|
assess_author_pr_report,
|
||||||
|
assess_capability_proof,
|
||||||
assess_controller_handoff,
|
assess_controller_handoff,
|
||||||
assess_inventory_completeness,
|
assess_inventory_completeness,
|
||||||
|
assess_secret_sweep,
|
||||||
assess_self_review_contamination,
|
assess_self_review_contamination,
|
||||||
assess_validation_report,
|
assess_validation_report,
|
||||||
build_final_report,
|
build_final_report,
|
||||||
@@ -315,6 +318,21 @@ class TestValidationReporting(unittest.TestCase):
|
|||||||
)
|
)
|
||||||
self.assertEqual(result["verdict"], "strong")
|
self.assertEqual(result["verdict"], "strong")
|
||||||
|
|
||||||
|
def test_stdout_capture_fix_without_summary_is_weak(self):
|
||||||
|
result = _good_validation(
|
||||||
|
is_stdout_capture_fix=True,
|
||||||
|
normal_pytest_summary=False
|
||||||
|
)
|
||||||
|
self.assertEqual(result["verdict"], "weak")
|
||||||
|
self.assertTrue(any("stdout-capture" in r for r in result["reasons"]))
|
||||||
|
|
||||||
|
def test_stdout_capture_fix_with_summary_is_strong(self):
|
||||||
|
result = _good_validation(
|
||||||
|
is_stdout_capture_fix=True,
|
||||||
|
normal_pytest_summary=True
|
||||||
|
)
|
||||||
|
self.assertEqual(result["verdict"], "strong")
|
||||||
|
|
||||||
|
|
||||||
class TestSelfReviewContamination(unittest.TestCase):
|
class TestSelfReviewContamination(unittest.TestCase):
|
||||||
"""Required behavior 5: contamination claims need evidence."""
|
"""Required behavior 5: contamination claims need evidence."""
|
||||||
@@ -381,6 +399,7 @@ class TestFinalReport(unittest.TestCase):
|
|||||||
"identity_eligible": True,
|
"identity_eligible": True,
|
||||||
"merge_performed": False,
|
"merge_performed": False,
|
||||||
"issue_status_verified": True,
|
"issue_status_verified": True,
|
||||||
|
"controller_handoff": "Controller Handoff\n- Task: test",
|
||||||
}
|
}
|
||||||
kwargs.update(overrides)
|
kwargs.update(overrides)
|
||||||
return build_final_report(**kwargs)
|
return build_final_report(**kwargs)
|
||||||
@@ -465,6 +484,24 @@ class TestFinalReport(unittest.TestCase):
|
|||||||
self.assertNotEqual(report["grade"], "A")
|
self.assertNotEqual(report["grade"], "A")
|
||||||
self.assertFalse(report["merge_allowed"])
|
self.assertFalse(report["merge_allowed"])
|
||||||
|
|
||||||
|
def test_failed_capability_proof_downgrades(self):
|
||||||
|
cap_proof = {
|
||||||
|
"proven": False,
|
||||||
|
"reasons": ["task mark_issue not allowed"]
|
||||||
|
}
|
||||||
|
report = self._report(capability_proof=cap_proof)
|
||||||
|
self.assertNotEqual(report["grade"], "A")
|
||||||
|
self.assertTrue(any("capability" in r for r in report["downgrade_reasons"]))
|
||||||
|
|
||||||
|
def test_failed_sweep_proof_downgrades(self):
|
||||||
|
sweep_proof = {
|
||||||
|
"proven": False,
|
||||||
|
"reasons": ["method missing"]
|
||||||
|
}
|
||||||
|
report = self._report(sweep_proof=sweep_proof)
|
||||||
|
self.assertNotEqual(report["grade"], "A")
|
||||||
|
self.assertTrue(any("sweep" in r for r in report["downgrade_reasons"]))
|
||||||
|
|
||||||
|
|
||||||
class TestStdoutIsolation(unittest.TestCase):
|
class TestStdoutIsolation(unittest.TestCase):
|
||||||
"""Regression test for #178: tests must not close or corrupt stdout/stderr
|
"""Regression test for #178: tests must not close or corrupt stdout/stderr
|
||||||
@@ -578,139 +615,103 @@ class TestRepoNameDisambiguation(unittest.TestCase):
|
|||||||
|
|
||||||
|
|
||||||
class TestControllerHandoff(unittest.TestCase):
|
class TestControllerHandoff(unittest.TestCase):
|
||||||
"""Issue #182: final reports must end with a Controller Handoff."""
|
"""#183: every final report must contain the Controller Handoff section."""
|
||||||
|
|
||||||
BASE_HANDOFF = "\n".join([
|
def test_report_with_handoff_passes(self):
|
||||||
"## Controller Handoff",
|
report = "some details\n\nController Handoff\n- Task: foo"
|
||||||
"",
|
result = assess_controller_handoff(report)
|
||||||
"- Task: implement issue #182",
|
self.assertTrue(result["present"])
|
||||||
"- Repo: Scaled-Tech-Consulting/Gitea-Tools",
|
|
||||||
"- Role: author",
|
|
||||||
"- Identity: jcwalker3 / prgs-author",
|
|
||||||
"- Issue/PR: #182 / PR #999",
|
|
||||||
"- Branch/SHA: feat/x @ 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9",
|
|
||||||
"- Files changed: review_proofs.py",
|
|
||||||
"- Validation: 700 passed, 6 skipped",
|
|
||||||
"- Mutations: one PR opened",
|
|
||||||
"- Current status: PR open",
|
|
||||||
"- Blockers: none",
|
|
||||||
"- Next: review PR #999",
|
|
||||||
"- Safety: no self-review; no self-merge; no secrets",
|
|
||||||
])
|
|
||||||
|
|
||||||
def test_report_without_handoff_is_downgraded(self):
|
def test_report_without_handoff_downgrades(self):
|
||||||
result = assess_controller_handoff("long report text, no handoff")
|
report = "long details without the section"
|
||||||
self.assertEqual(result["verdict"], "missing")
|
result = assess_controller_handoff(report)
|
||||||
self.assertTrue(result["downgraded"])
|
self.assertFalse(result["present"])
|
||||||
|
self.assertTrue(any("Controller Handoff" in r for r in result["reasons"]))
|
||||||
|
|
||||||
def test_wrong_title_is_downgraded(self):
|
|
||||||
text = self.BASE_HANDOFF.replace(
|
|
||||||
"## Controller Handoff", "## Handoff Summary")
|
|
||||||
result = assess_controller_handoff(text)
|
|
||||||
self.assertEqual(result["verdict"], "missing")
|
|
||||||
|
|
||||||
def test_complete_base_handoff_passes(self):
|
class TestAuthorReporting(unittest.TestCase):
|
||||||
result = assess_controller_handoff(
|
"""Harness assertions for author reporting and capability/sweep proofs (#183)."""
|
||||||
"full report body...\n\n" + self.BASE_HANDOFF)
|
|
||||||
self.assertEqual(result["verdict"], "complete")
|
|
||||||
self.assertFalse(result["downgraded"])
|
|
||||||
|
|
||||||
def test_missing_base_fields_are_listed(self):
|
def test_good_capability_proof_passes(self):
|
||||||
text = "\n".join(
|
resolved = {
|
||||||
line for line in self.BASE_HANDOFF.splitlines()
|
"mark_issue": {
|
||||||
if not line.startswith(("- Mutations:", "- Safety:")))
|
"requested_task": "mark_issue",
|
||||||
result = assess_controller_handoff(text)
|
"required_operation_permission": "gitea.issue.write",
|
||||||
self.assertEqual(result["verdict"], "incomplete")
|
"allowed_in_current_session": True,
|
||||||
self.assertTrue(result["downgraded"])
|
}
|
||||||
self.assertIn("Mutations", result["missing_fields"])
|
}
|
||||||
self.assertIn("Safety", result["missing_fields"])
|
result = assess_capability_proof(resolved)
|
||||||
|
self.assertTrue(result["proven"])
|
||||||
|
|
||||||
def test_review_role_requires_review_fields(self):
|
def test_missing_capability_proof_fails_closed(self):
|
||||||
result = assess_controller_handoff(self.BASE_HANDOFF, role="review")
|
result = assess_capability_proof({})
|
||||||
self.assertEqual(result["verdict"], "incomplete")
|
self.assertFalse(result["proven"])
|
||||||
self.assertIn("Pinned reviewed head", result["missing_fields"])
|
self.assertTrue(any("fail closed" in r for r in result["reasons"]))
|
||||||
self.assertIn("Merge result", result["missing_fields"])
|
|
||||||
|
|
||||||
complete = self.BASE_HANDOFF + "\n" + "\n".join([
|
def test_unresolved_capability_proof_fails_closed(self):
|
||||||
"- Selected PR: #999",
|
# unknown task resolving to None/unknown requested task
|
||||||
"- Reviewer eligibility: passed",
|
resolved = {
|
||||||
"- Pinned reviewed head: 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9",
|
"mark_issue": {
|
||||||
"- Review decision: approve",
|
"requested_task": "unknown_task",
|
||||||
"- Merge result: merged",
|
"required_operation_permission": None,
|
||||||
"- Linked issue status: closed",
|
"allowed_in_current_session": None,
|
||||||
"- Cleanup status: branch deleted",
|
}
|
||||||
])
|
}
|
||||||
result = assess_controller_handoff(complete, role="review")
|
result = assess_capability_proof(resolved)
|
||||||
self.assertEqual(result["verdict"], "complete")
|
self.assertFalse(result["proven"])
|
||||||
|
self.assertTrue(any("could not be resolved" in r for r in result["reasons"]))
|
||||||
|
|
||||||
def test_author_role_requires_author_fields(self):
|
def test_good_secret_sweep_passes(self):
|
||||||
complete = self.BASE_HANDOFF + "\n" + "\n".join([
|
sweep = {
|
||||||
"- Selected issue: #182",
|
"method": "git diff | grep -iE 'token|secret'",
|
||||||
"- Claim/comment status: comment-claimed",
|
"scope": "staged diff relative to master",
|
||||||
"- PR number opened: #999",
|
"clean": True,
|
||||||
"- No review/merge: confirmed",
|
}
|
||||||
])
|
result = assess_secret_sweep(sweep)
|
||||||
result = assess_controller_handoff(complete, role="author")
|
self.assertTrue(result["proven"])
|
||||||
self.assertEqual(result["verdict"], "complete")
|
self.assertEqual(result["verdict"], "strong")
|
||||||
|
|
||||||
result = assess_controller_handoff(self.BASE_HANDOFF, role="author")
|
def test_vague_secret_sweep_without_method_is_weak(self):
|
||||||
self.assertEqual(result["verdict"], "incomplete")
|
sweep = {
|
||||||
self.assertIn("No review/merge confirmation", result["missing_fields"])
|
"method": "",
|
||||||
|
"scope": "staged diff",
|
||||||
|
"clean": True,
|
||||||
|
}
|
||||||
|
result = assess_secret_sweep(sweep)
|
||||||
|
self.assertFalse(result["proven"])
|
||||||
|
self.assertEqual(result["verdict"], "weak")
|
||||||
|
self.assertTrue(any("method" in r or "scan" in r for r in result["reasons"]))
|
||||||
|
|
||||||
def test_author_role_rejects_equivalent_or_multiple_issues(self):
|
def test_unconfirmed_secret_sweep_is_invalid(self):
|
||||||
# 1. equivalent reference blocked
|
sweep = {
|
||||||
incomplete_eq = self.BASE_HANDOFF + "\n" + "\n".join([
|
"method": "git diff | grep",
|
||||||
"- Selected issue: Issue #194 / #196 equivalent",
|
"scope": "staged diff",
|
||||||
"- Claim/comment status: comment-claimed",
|
"clean": False,
|
||||||
"- PR number opened: #999",
|
}
|
||||||
"- No review/merge: confirmed",
|
result = assess_secret_sweep(sweep)
|
||||||
])
|
self.assertFalse(result["proven"])
|
||||||
res = assess_controller_handoff(incomplete_eq, role="author")
|
self.assertEqual(result["verdict"], "weak")
|
||||||
self.assertEqual(res["verdict"], "incomplete")
|
|
||||||
self.assertIn("Selected issue", res["missing_fields"])
|
|
||||||
|
|
||||||
# 2. multiple issues blocked
|
def test_good_author_pr_report_passes(self):
|
||||||
incomplete_multi = self.BASE_HANDOFF + "\n" + "\n".join([
|
pr = {
|
||||||
"- Selected issue: #194, #196",
|
"pr_number": 185,
|
||||||
"- Claim/comment status: comment-claimed",
|
"branch": "feat/issue-184-repo-name-disambiguation",
|
||||||
"- PR number opened: #999",
|
"head_sha": "e2bccbafeeb93124ba068bfb06058d5aa7467cae",
|
||||||
"- No review/merge: confirmed",
|
}
|
||||||
])
|
result = assess_author_pr_report(pr)
|
||||||
res = assess_controller_handoff(incomplete_multi, role="author")
|
self.assertTrue(result["complete"])
|
||||||
self.assertEqual(res["verdict"], "incomplete")
|
|
||||||
self.assertIn("Selected issue", res["missing_fields"])
|
|
||||||
|
|
||||||
def test_author_role_rejects_fuzzy_pr_number(self):
|
def test_incomplete_author_pr_report_fails(self):
|
||||||
incomplete_pr = self.BASE_HANDOFF + "\n" + "\n".join([
|
pr = {
|
||||||
"- Selected issue: #196",
|
"pr_number": 0,
|
||||||
"- Claim/comment status: comment-claimed",
|
"branch": "",
|
||||||
"- PR number opened: PR #203 / #204 equivalent",
|
"head_sha": "e2bccba",
|
||||||
"- No review/merge: confirmed",
|
}
|
||||||
])
|
result = assess_author_pr_report(pr)
|
||||||
res = assess_controller_handoff(incomplete_pr, role="author")
|
self.assertFalse(result["complete"])
|
||||||
self.assertEqual(res["verdict"], "incomplete")
|
self.assertTrue(any("number" in r for r in result["reasons"]))
|
||||||
self.assertIn("PR number opened", res["missing_fields"])
|
self.assertTrue(any("branch" in r for r in result["reasons"]))
|
||||||
|
self.assertTrue(any("SHA" in r for r in result["reasons"]))
|
||||||
def test_inventory_role_requires_inventory_fields(self):
|
|
||||||
complete = self.BASE_HANDOFF + "\n" + "\n".join([
|
|
||||||
"- Repositories checked: Gitea-Tools, mcp-control-plane",
|
|
||||||
"- Open PR counts: 2 / 0",
|
|
||||||
"- Selected PR or reason: none eligible (self-authored)",
|
|
||||||
"- Inventory completeness: complete, no pagination needed",
|
|
||||||
])
|
|
||||||
result = assess_controller_handoff(complete, role="inventory")
|
|
||||||
self.assertEqual(result["verdict"], "complete")
|
|
||||||
|
|
||||||
def test_skill_doc_declares_handoff_requirement(self):
|
|
||||||
# Doc-contract: SKILL.md must keep requiring the exact section and
|
|
||||||
# naming this validator, or the convention silently rots.
|
|
||||||
skill = (
|
|
||||||
__import__("pathlib").Path(__file__).resolve().parent.parent
|
|
||||||
/ "skills" / "llm-project-workflow" / "SKILL.md"
|
|
||||||
).read_text(encoding="utf-8")
|
|
||||||
self.assertIn("## Controller Handoff", skill)
|
|
||||||
self.assertIn("assess_controller_handoff", skill)
|
|
||||||
self.assertIn("issue #182", skill)
|
|
||||||
|
|
||||||
|
|
||||||
if __name__ == "__main__":
|
if __name__ == "__main__":
|
||||||
|
|||||||
+5
-33
@@ -14,39 +14,11 @@ BRANCHES = REPO / "branches"
|
|||||||
|
|
||||||
|
|
||||||
def run(script, *args):
|
def run(script, *args):
|
||||||
branch = None
|
proc = subprocess.run(
|
||||||
for arg in args:
|
["bash", str(SCRIPTS / script), *args],
|
||||||
if not arg.startswith("-"):
|
capture_output=True, text=True, cwd=str(REPO),
|
||||||
branch = arg
|
)
|
||||||
break
|
return proc.returncode, proc.stdout, proc.stderr
|
||||||
|
|
||||||
lock_file = Path("/tmp/gitea_issue_lock.json")
|
|
||||||
created_lock = False
|
|
||||||
if script == "worktree-start" and branch:
|
|
||||||
import re
|
|
||||||
import json
|
|
||||||
m = re.search(r"issue-(\d+)", branch)
|
|
||||||
if not m:
|
|
||||||
m = re.search(r"pr-(\d+)", branch)
|
|
||||||
issue_num = int(m.group(1)) if m else 999
|
|
||||||
lock_file.write_text(json.dumps({
|
|
||||||
"issue_number": issue_num,
|
|
||||||
"branch_name": branch,
|
|
||||||
"remote": "prgs",
|
|
||||||
"org": "Scaled-Tech-Consulting",
|
|
||||||
"repo": "Gitea-Tools"
|
|
||||||
}), encoding="utf-8")
|
|
||||||
created_lock = True
|
|
||||||
|
|
||||||
try:
|
|
||||||
proc = subprocess.run(
|
|
||||||
["bash", str(SCRIPTS / script), *args],
|
|
||||||
capture_output=True, text=True, cwd=str(REPO),
|
|
||||||
)
|
|
||||||
return proc.returncode, proc.stdout, proc.stderr
|
|
||||||
finally:
|
|
||||||
if created_lock and lock_file.exists():
|
|
||||||
lock_file.unlink()
|
|
||||||
|
|
||||||
|
|
||||||
class TestWorktreeStart(unittest.TestCase):
|
class TestWorktreeStart(unittest.TestCase):
|
||||||
|
|||||||
Reference in New Issue
Block a user