Harden author-run reporting and mutation capability proof per #183 #187
+5
-1
@@ -4213,7 +4213,11 @@ def gitea_resolve_task_capability(
|
||||
"role": "author",
|
||||
},
|
||||
"claim_issue": {
|
||||
"permission": "gitea.issue.write",
|
||||
"permission": "gitea.issue.comment",
|
||||
"role": "author",
|
||||
},
|
||||
"mark_issue": {
|
||||
"permission": "gitea.issue.comment",
|
||||
"role": "author",
|
||||
},
|
||||
"create_branch": {
|
||||
|
||||
+138
-1
@@ -635,7 +635,9 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
|
||||
issue_status_verified,
|
||||
capability_evidence=None, sweep=None, live_state=None,
|
||||
role_boundary=None, review_mutation=None,
|
||||
report_text=None, review_decision_lock=None):
|
||||
report_text=None, review_decision_lock=None,
|
||||
controller_handoff=None, capability_proof=None,
|
||||
sweep_proof=None):
|
||||
"""Required behavior 6 + acceptance criteria: one report, distinct proofs.
|
||||
|
||||
Combines the individual proof verdicts into the final-report fields the
|
||||
@@ -671,6 +673,16 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
|
||||
"violations": [],
|
||||
}
|
||||
role_status = role_boundary.get("status", "warning")
|
||||
handoff = _normalize_controller_handoff_proof(controller_handoff)
|
||||
capability_proof = capability_proof or {
|
||||
"proven": False,
|
||||
"reasons": ["capability proof not provided"],
|
||||
}
|
||||
sweep_proof = sweep_proof or {
|
||||
"proven": False,
|
||||
"verdict": "invalid",
|
||||
"reasons": ["secret/provenance sweep proof not provided"],
|
||||
}
|
||||
|
||||
capability_evidence = capability_evidence or {
|
||||
"proven": False,
|
||||
@@ -731,6 +743,15 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
|
||||
downgrade_reasons.extend(role_boundary.get("reasons", []))
|
||||
if not issue_status_verified:
|
||||
downgrade_reasons.append("linked issue status not verified")
|
||||
if not handoff["complete"]:
|
||||
downgrade_reasons.append("Controller Handoff missing or incomplete")
|
||||
downgrade_reasons.extend(handoff.get("reasons", []))
|
||||
if not capability_proof.get("proven"):
|
||||
downgrade_reasons.append("exact mutation capability proof missing")
|
||||
downgrade_reasons.extend(capability_proof.get("reasons", []))
|
||||
if not sweep_proof.get("proven"):
|
||||
downgrade_reasons.append("exact secret/provenance sweep proof missing")
|
||||
downgrade_reasons.extend(sweep_proof.get("reasons", []))
|
||||
if not capability_proven:
|
||||
downgrade_reasons.append("exact capability evidence missing (#179)")
|
||||
downgrade_reasons.extend(capability_evidence.get("reasons", []))
|
||||
@@ -795,6 +816,10 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
|
||||
"merge_allowed": merge_allowed,
|
||||
"merge_performed": bool(merge_performed),
|
||||
"issue_status_verified": bool(issue_status_verified),
|
||||
"controller_handoff_present": handoff["present"],
|
||||
"controller_handoff_complete": handoff["complete"],
|
||||
"capability_proof": bool(capability_proof.get("proven")),
|
||||
"secret_sweep_proof": bool(sweep_proof.get("proven")),
|
||||
"capability_evidence_proven": capability_proven,
|
||||
"sweep_verdict": sweep.get("verdict"),
|
||||
"live_state_recheck_proven": live_state_proven,
|
||||
@@ -803,6 +828,118 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
|
||||
}
|
||||
|
||||
|
||||
def _normalize_controller_handoff_proof(controller_handoff):
|
||||
"""Normalize old boolean handoff proof and rich #182 handoff verdicts."""
|
||||
if controller_handoff is None:
|
||||
return {
|
||||
"present": False,
|
||||
"complete": False,
|
||||
"reasons": ["Controller Handoff proof not provided"],
|
||||
}
|
||||
if isinstance(controller_handoff, str):
|
||||
controller_handoff = assess_controller_handoff(controller_handoff)
|
||||
if not isinstance(controller_handoff, dict):
|
||||
return {
|
||||
"present": False,
|
||||
"complete": False,
|
||||
"reasons": ["Controller Handoff proof has invalid type"],
|
||||
}
|
||||
if "verdict" in controller_handoff:
|
||||
verdict = controller_handoff.get("verdict")
|
||||
return {
|
||||
"present": verdict in {"complete", "incomplete"},
|
||||
"complete": verdict == "complete",
|
||||
"reasons": list(controller_handoff.get("reasons") or []),
|
||||
}
|
||||
present = bool(controller_handoff.get("present"))
|
||||
return {
|
||||
"present": present,
|
||||
"complete": present,
|
||||
"reasons": list(controller_handoff.get("reasons") or []),
|
||||
}
|
||||
|
||||
|
||||
def assess_capability_proof(resolved_capabilities: dict) -> dict:
|
||||
"""Every mutation must have exact capability proof (#183)."""
|
||||
reasons = []
|
||||
if not resolved_capabilities:
|
||||
return {
|
||||
"proven": False,
|
||||
"reasons": ["no capability proof resolved; fail closed"],
|
||||
}
|
||||
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 (
|
||||
"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"
|
||||
)
|
||||
return {"proven": not reasons, "reasons": reasons}
|
||||
|
||||
|
||||
def assess_secret_sweep(sweep_report: dict) -> dict:
|
||||
"""Secret/provenance sweeps must state exact command/method and scope."""
|
||||
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"
|
||||
return {
|
||||
"proven": not reasons and sweep_report.get("clean") is True,
|
||||
"verdict": verdict,
|
||||
"reasons": reasons,
|
||||
}
|
||||
|
||||
|
||||
def assess_author_pr_report(pr_report: dict) -> dict:
|
||||
"""Author PR reports must include PR number, branch, and exact head SHA."""
|
||||
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")
|
||||
|
||||
return {
|
||||
"complete": not reasons,
|
||||
"reasons": reasons,
|
||||
}
|
||||
|
||||
|
||||
# ── Controller Handoff validation (Issue #182) ────────────────────────────────
|
||||
#
|
||||
# Every final report must end with a compact section titled exactly
|
||||
|
||||
@@ -341,11 +341,13 @@ Ready-to-copy templates live in [`templates/`](templates/):
|
||||
|
||||
## 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,
|
||||
discussion-only, or blocked planning. It lets a controller LLM understand the
|
||||
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
|
||||
readability, not as a full human status report. PR bodies still carry the
|
||||
full review detail — the handoff never replaces PR documentation.
|
||||
|
||||
@@ -130,6 +130,17 @@ class TestResolveTaskCapability(unittest.TestCase):
|
||||
self.assertIn("author-profile", res["matching_configured_profile"])
|
||||
self.assertIn("reviewer-profile", res["matching_configured_profile"])
|
||||
|
||||
@patch("mcp_server.api_request", return_value={"login": "author-user"})
|
||||
@patch("mcp_server.get_auth_header", return_value="token author-pass")
|
||||
def test_resolve_mark_issue_author_profile_allowed(self, _auth, _api):
|
||||
# #183: mark_issue must map to gitea.issue.comment (prgs-author allowed op).
|
||||
with patch.dict(os.environ, self._env("author-profile")):
|
||||
res = mcp_server.gitea_resolve_task_capability(task="mark_issue", remote="prgs")
|
||||
self.assertEqual(res["requested_task"], "mark_issue")
|
||||
self.assertEqual(res["required_operation_permission"], "gitea.issue.comment")
|
||||
self.assertTrue(res["allowed_in_current_session"])
|
||||
self.assertIn("gitea.issue.comment", res["active_profile_allowed_operations"])
|
||||
|
||||
def test_resolve_unknown_task_fails_closed(self):
|
||||
with patch.dict(os.environ, self._env("author-profile")):
|
||||
with self.assertRaises(ValueError):
|
||||
|
||||
@@ -21,12 +21,15 @@ import unittest
|
||||
sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
|
||||
|
||||
from review_proofs import ( # noqa: E402
|
||||
assess_author_pr_report,
|
||||
assess_capability_evidence,
|
||||
assess_capability_proof,
|
||||
assess_controller_handoff,
|
||||
assess_inventory_completeness,
|
||||
assess_live_state_recheck,
|
||||
assess_review_mutation_final_report,
|
||||
assess_role_boundary,
|
||||
assess_secret_sweep,
|
||||
assess_self_review_contamination,
|
||||
assess_sweep_evidence,
|
||||
assess_validation_report,
|
||||
@@ -190,6 +193,49 @@ def _good_role_boundary_179(**overrides):
|
||||
return assess_role_boundary(**kwargs)
|
||||
|
||||
|
||||
GOOD_HANDOFF = "\n".join([
|
||||
"## Controller Handoff",
|
||||
"",
|
||||
"- Task: review PR #999",
|
||||
"- Repo: Scaled-Tech-Consulting/Gitea-Tools",
|
||||
"- Role: reviewer",
|
||||
"- Identity: sysadmin / prgs-reviewer",
|
||||
"- Issue/PR: #182 / PR #999",
|
||||
"- Branch/SHA: feat/x @ 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9",
|
||||
"- Files changed: review_proofs.py",
|
||||
"- Validation: 700 passed, 6 skipped",
|
||||
"- Mutations: review only",
|
||||
"- Current status: PR open",
|
||||
"- Blockers: none",
|
||||
"- Next: merge if approved",
|
||||
"- Safety: no self-review; no self-merge; no secrets",
|
||||
])
|
||||
|
||||
|
||||
def _good_handoff():
|
||||
return assess_controller_handoff(GOOD_HANDOFF)
|
||||
|
||||
|
||||
def _good_capability_proof():
|
||||
return assess_capability_proof({
|
||||
"comment_issue": {
|
||||
"requested_task": "comment_issue",
|
||||
"required_operation_permission": "gitea.issue.comment",
|
||||
"allowed_in_current_session": True,
|
||||
}
|
||||
})
|
||||
|
||||
|
||||
def _good_secret_sweep(**overrides):
|
||||
sweep = {
|
||||
"method": "git diff prgs/master...HEAD | grep -iE 'token|secret'",
|
||||
"scope": "full PR diff relative to prgs/master",
|
||||
"clean": True,
|
||||
}
|
||||
sweep.update(overrides)
|
||||
return assess_secret_sweep(sweep)
|
||||
|
||||
|
||||
class TestCheckoutProof(unittest.TestCase):
|
||||
"""Required behavior 1 + 2: prove HEAD == pinned PR head or stop."""
|
||||
|
||||
@@ -545,6 +591,9 @@ class TestFinalReport(unittest.TestCase):
|
||||
"live_state": _good_live_state(),
|
||||
"role_boundary": _good_role_boundary(),
|
||||
"review_mutation": _good_review_mutation(),
|
||||
"controller_handoff": _good_handoff(),
|
||||
"capability_proof": _good_capability_proof(),
|
||||
"sweep_proof": _good_secret_sweep(),
|
||||
}
|
||||
kwargs.update(overrides)
|
||||
return build_final_report(**kwargs)
|
||||
@@ -662,6 +711,28 @@ class TestFinalReport(unittest.TestCase):
|
||||
self.assertEqual(report["grade"], "blocked")
|
||||
self.assertFalse(report["merge_allowed"])
|
||||
|
||||
def test_failed_capability_proof_downgrades(self):
|
||||
report = self._report(
|
||||
capability_proof={
|
||||
"proven": False,
|
||||
"reasons": ["task mark_issue not allowed"],
|
||||
}
|
||||
)
|
||||
self.assertNotEqual(report["grade"], "A")
|
||||
self.assertTrue(
|
||||
any("capability" in r for r in report["downgrade_reasons"])
|
||||
)
|
||||
|
||||
def test_failed_sweep_proof_downgrades(self):
|
||||
report = self._report(
|
||||
sweep_proof={
|
||||
"proven": False,
|
||||
"reasons": ["method missing"],
|
||||
}
|
||||
)
|
||||
self.assertNotEqual(report["grade"], "A")
|
||||
self.assertTrue(any("sweep" in r for r in report["downgrade_reasons"]))
|
||||
|
||||
|
||||
class TestStdoutIsolation(unittest.TestCase):
|
||||
"""Regression test for #178: tests must not close or corrupt stdout/stderr
|
||||
@@ -1231,6 +1302,9 @@ class TestFinalReport179Bar(unittest.TestCase):
|
||||
"live_state": _good_live_state(),
|
||||
"role_boundary": _good_role_boundary(),
|
||||
"review_mutation": _good_review_mutation(),
|
||||
"controller_handoff": _good_handoff(),
|
||||
"capability_proof": _good_capability_proof(),
|
||||
"sweep_proof": _good_secret_sweep(),
|
||||
}
|
||||
kwargs.update(overrides)
|
||||
return build_final_report(**kwargs)
|
||||
@@ -1306,5 +1380,74 @@ class TestFinalReport179Bar(unittest.TestCase):
|
||||
self.assertFalse(report["review_mutation_complete"])
|
||||
|
||||
|
||||
class TestAuthorReporting(unittest.TestCase):
|
||||
"""Harness assertions for author reporting and capability/sweep proofs."""
|
||||
|
||||
def test_good_capability_proof_passes(self):
|
||||
result = _good_capability_proof()
|
||||
self.assertTrue(result["proven"])
|
||||
|
||||
def test_missing_capability_proof_fails_closed(self):
|
||||
result = assess_capability_proof({})
|
||||
self.assertFalse(result["proven"])
|
||||
self.assertTrue(any("fail closed" in r for r in result["reasons"]))
|
||||
|
||||
def test_unresolved_capability_proof_fails_closed(self):
|
||||
result = assess_capability_proof({
|
||||
"mark_issue": {
|
||||
"requested_task": "unknown_task",
|
||||
"required_operation_permission": None,
|
||||
"allowed_in_current_session": None,
|
||||
}
|
||||
})
|
||||
self.assertFalse(result["proven"])
|
||||
self.assertTrue(
|
||||
any("could not be resolved" in r for r in result["reasons"])
|
||||
)
|
||||
|
||||
def test_good_secret_sweep_passes(self):
|
||||
result = _good_secret_sweep()
|
||||
self.assertTrue(result["proven"])
|
||||
self.assertEqual(result["verdict"], "strong")
|
||||
|
||||
def test_vague_secret_sweep_without_method_is_weak(self):
|
||||
result = _good_secret_sweep(method="")
|
||||
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_unconfirmed_secret_sweep_is_weak(self):
|
||||
result = _good_secret_sweep(clean=False)
|
||||
self.assertFalse(result["proven"])
|
||||
self.assertEqual(result["verdict"], "weak")
|
||||
|
||||
def test_good_author_pr_report_passes(self):
|
||||
result = assess_author_pr_report({
|
||||
"pr_number": 185,
|
||||
"branch": "feat/issue-184-repo-name-disambiguation",
|
||||
"head_sha": "e2bccbafeeb93124ba068bfb06058d5aa7467cae",
|
||||
})
|
||||
self.assertTrue(result["complete"])
|
||||
|
||||
def test_author_pr_report_without_head_sha_is_incomplete(self):
|
||||
result = assess_author_pr_report({
|
||||
"pr_number": 185,
|
||||
"branch": "feat/issue-184-repo-name-disambiguation",
|
||||
"head_sha": "",
|
||||
})
|
||||
self.assertFalse(result["complete"])
|
||||
self.assertTrue(any("head SHA" in r for r in result["reasons"]))
|
||||
|
||||
def test_author_pr_report_rejects_short_sha(self):
|
||||
result = assess_author_pr_report({
|
||||
"pr_number": 185,
|
||||
"branch": "feat/issue-184-repo-name-disambiguation",
|
||||
"head_sha": "e2bccba",
|
||||
})
|
||||
self.assertFalse(result["complete"])
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
|
||||
Reference in New Issue
Block a user