Compare commits
12
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
a86e64ff01 | ||
|
|
056a232ef8 | ||
|
|
8fa94a07a8 | ||
|
|
ca3de3da53 | ||
|
|
be6feabf70 | ||
|
|
1071619532 | ||
|
|
1033a22407 | ||
|
|
7966e70db6 | ||
|
|
4f466550ca | ||
|
|
6ac6b9528c | ||
|
|
4dd32bb9f7 | ||
|
|
d5d3331498 |
@@ -120,6 +120,11 @@ def assess_capability_stop_report(
|
|||||||
capability_denied: bool = True,
|
capability_denied: bool = True,
|
||||||
) -> dict:
|
) -> dict:
|
||||||
"""Validate final report purity after reviewer capability denial."""
|
"""Validate final report purity after reviewer capability denial."""
|
||||||
|
from review_proofs import (
|
||||||
|
assess_empty_queue_report,
|
||||||
|
parse_trust_gate_status_from_report,
|
||||||
|
)
|
||||||
|
|
||||||
text = report_text or ""
|
text = report_text or ""
|
||||||
lower = text.lower()
|
lower = text.lower()
|
||||||
violations = []
|
violations = []
|
||||||
@@ -148,13 +153,19 @@ def assess_capability_stop_report(
|
|||||||
r"inventory empty",
|
r"inventory empty",
|
||||||
re.I,
|
re.I,
|
||||||
)
|
)
|
||||||
|
parsed_status = parse_trust_gate_status_from_report(text)
|
||||||
|
effective_status = trust_gate_status or parsed_status
|
||||||
if empty_queue_patterns.search(text):
|
if empty_queue_patterns.search(text):
|
||||||
if trust_gate_status != "trusted_empty":
|
if effective_status != "trusted_empty":
|
||||||
violations.append(
|
violations.append(
|
||||||
"empty-queue claim after capability stop without "
|
"empty-queue claim after capability stop without "
|
||||||
"pr_inventory_trust_gate.status == trusted_empty"
|
"pr_inventory_trust_gate.status == trusted_empty"
|
||||||
)
|
)
|
||||||
|
|
||||||
|
empty_queue = assess_empty_queue_report(text)
|
||||||
|
if empty_queue.get("claimed") and not empty_queue.get("proven"):
|
||||||
|
violations.extend(empty_queue.get("reasons") or [])
|
||||||
|
|
||||||
ok, elig_violations = validate_eligibility_wording(text)
|
ok, elig_violations = validate_eligibility_wording(text)
|
||||||
violations.extend(elig_violations)
|
violations.extend(elig_violations)
|
||||||
|
|
||||||
|
|||||||
@@ -31,7 +31,8 @@ Wiki-related issues cannot be closed until the live Wiki is verified — see
|
|||||||
|
|
||||||
| Repository | `docs/wiki/` source | Gitea Wiki published | Proof |
|
| Repository | `docs/wiki/` source | Gitea Wiki published | Proof |
|
||||||
|---|---|---|---|
|
|---|---|---|---|
|
||||||
| `Scaled-Tech-Consulting/Gitea-Tools` | yes (10 pages) | published (verified 2026-07-06) | [Wiki Home](https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools/wiki/Home); 10 pages; wiki git log head `11549ee` |
|
| `Scaled-Tech-Consulting/Gitea-Tools` | yes (10 pages) | published (verified 2026-07-06) | [Wiki Home](https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools/wiki/Home); 10 pages; wiki git log head `d1f0693` |
|
||||||
| `Scaled-Tech-Consulting/mcp-control-plane` | yes (10 pages) | published (verified 2026-07-06) | [Wiki Home](https://gitea.prgs.cc/Scaled-Tech-Consulting/mcp-control-plane/wiki/Home) |
|
| `Scaled-Tech-Consulting/mcp-control-plane` | yes (10 pages) | published (verified 2026-07-06) | [Wiki Home](https://gitea.prgs.cc/Scaled-Tech-Consulting/mcp-control-plane/wiki/Home); 10 pages (History, Home, Identity-and-Profiles, MCP-Tools, Open-Decisions, Operator-Guide, Repositories, Runbooks, Safety-and-Gates, Workflow); wiki git log head `ef3dec2` |
|
||||||
|
|
||||||
|
|
||||||
Update this table whenever a wiki is published, re-synced, or found stale.
|
Update this table whenever a wiki is published, re-synced, or found stale.
|
||||||
+42
-1
@@ -259,6 +259,7 @@ import issue_duplicate_gate # noqa: E402
|
|||||||
import role_session_router # noqa: E402
|
import role_session_router # noqa: E402
|
||||||
import role_namespace_gate # noqa: E402
|
import role_namespace_gate # noqa: E402
|
||||||
import task_capability_map # noqa: E402
|
import task_capability_map # noqa: E402
|
||||||
|
import review_proofs # noqa: E402
|
||||||
|
|
||||||
|
|
||||||
# Fail-closed exact-issue-lock file (#204): written by gitea_lock_issue,
|
# Fail-closed exact-issue-lock file (#204): written by gitea_lock_issue,
|
||||||
@@ -2274,6 +2275,23 @@ def gitea_merge_pr(
|
|||||||
return result
|
return result
|
||||||
|
|
||||||
|
|
||||||
|
def _local_git_remote_url(remote_name: str) -> str | None:
|
||||||
|
"""Best-effort local ``git remote get-url`` for trust-gate corroboration."""
|
||||||
|
try:
|
||||||
|
proc = subprocess.run(
|
||||||
|
["git", "remote", "get-url", remote_name],
|
||||||
|
capture_output=True,
|
||||||
|
text=True,
|
||||||
|
cwd=PROJECT_ROOT,
|
||||||
|
)
|
||||||
|
if proc.returncode != 0:
|
||||||
|
return None
|
||||||
|
url = (proc.stdout or "").strip()
|
||||||
|
return url or None
|
||||||
|
except Exception:
|
||||||
|
return None
|
||||||
|
|
||||||
|
|
||||||
@mcp.tool()
|
@mcp.tool()
|
||||||
def gitea_review_pr(
|
def gitea_review_pr(
|
||||||
pr_number: int,
|
pr_number: int,
|
||||||
@@ -2341,6 +2359,8 @@ def gitea_review_pr(
|
|||||||
prs_found_count = 0
|
prs_found_count = 0
|
||||||
pr_details_list = []
|
pr_details_list = []
|
||||||
inventory_msg = ""
|
inventory_msg = ""
|
||||||
|
inventory_trust_gate = None
|
||||||
|
prs: list = []
|
||||||
|
|
||||||
h, o, r = _resolve(remote, host, org, repo)
|
h, o, r = _resolve(remote, host, org, repo)
|
||||||
auth = None
|
auth = None
|
||||||
@@ -2425,7 +2445,28 @@ def gitea_review_pr(
|
|||||||
if inventory_msg:
|
if inventory_msg:
|
||||||
report_lines.append(inventory_msg)
|
report_lines.append(inventory_msg)
|
||||||
else:
|
else:
|
||||||
report_lines.append("Open PRs found: 0")
|
inventory_trust_gate = review_proofs.pr_inventory_trust_gate(
|
||||||
|
prs if inventory_attempted else None,
|
||||||
|
remote=remote,
|
||||||
|
org=o,
|
||||||
|
repo=r,
|
||||||
|
state="open",
|
||||||
|
authenticated_profile=profile,
|
||||||
|
local_remote_url=_local_git_remote_url(remote),
|
||||||
|
has_finality_metadata=True,
|
||||||
|
)
|
||||||
|
report_lines.extend(
|
||||||
|
review_proofs.format_pr_inventory_trust_gate_report(
|
||||||
|
inventory_trust_gate
|
||||||
|
)
|
||||||
|
)
|
||||||
|
if inventory_trust_gate.get("status") == "trusted_empty":
|
||||||
|
report_lines.append("Open PRs found: 0 (trusted_empty)")
|
||||||
|
else:
|
||||||
|
report_lines.append(
|
||||||
|
"Empty-queue claim blocked: "
|
||||||
|
"pr_inventory_trust_gate did not return trusted_empty"
|
||||||
|
)
|
||||||
else:
|
else:
|
||||||
report_lines.append(inventory_msg)
|
report_lines.append(inventory_msg)
|
||||||
|
|
||||||
|
|||||||
+423
-1
@@ -17,6 +17,7 @@ here weakens or replaces them.
|
|||||||
import re
|
import re
|
||||||
|
|
||||||
import issue_duplicate_gate
|
import issue_duplicate_gate
|
||||||
|
from reviewer_worktree import assess_reviewer_worktree_proof
|
||||||
|
|
||||||
_FULL_SHA = re.compile(r"^[0-9a-f]{40}$")
|
_FULL_SHA = re.compile(r"^[0-9a-f]{40}$")
|
||||||
|
|
||||||
@@ -637,7 +638,7 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
|
|||||||
role_boundary=None, review_mutation=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,
|
controller_handoff=None, capability_proof=None,
|
||||||
sweep_proof=None):
|
sweep_proof=None, worktree_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
|
||||||
@@ -707,6 +708,14 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
|
|||||||
"downgraded": True,
|
"downgraded": True,
|
||||||
"reasons": ["review mutation proof not provided (#211)"],
|
"reasons": ["review mutation proof not provided (#211)"],
|
||||||
}
|
}
|
||||||
|
if worktree_proof is not None:
|
||||||
|
worktree = assess_reviewer_worktree_proof(worktree_proof)
|
||||||
|
else:
|
||||||
|
worktree = {
|
||||||
|
"proven": False,
|
||||||
|
"block": True,
|
||||||
|
"reasons": ["reviewer worktree proof not provided (#233)"],
|
||||||
|
}
|
||||||
|
|
||||||
capability_proven = bool(capability_evidence.get("proven"))
|
capability_proven = bool(capability_evidence.get("proven"))
|
||||||
sweep_proven = bool(sweep.get("proven"))
|
sweep_proven = bool(sweep.get("proven"))
|
||||||
@@ -719,6 +728,7 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
|
|||||||
"reasons": ["review mutation proof missing"],
|
"reasons": ["review mutation proof missing"],
|
||||||
}
|
}
|
||||||
review_mutation_complete = bool(review_mutation.get("complete"))
|
review_mutation_complete = bool(review_mutation.get("complete"))
|
||||||
|
worktree_proven = bool(worktree.get("proven"))
|
||||||
|
|
||||||
downgrade_reasons = []
|
downgrade_reasons = []
|
||||||
if not identity_eligible:
|
if not identity_eligible:
|
||||||
@@ -772,6 +782,11 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
|
|||||||
if not review_mutation_complete:
|
if not review_mutation_complete:
|
||||||
downgrade_reasons.append("review mutation proof missing or incomplete (#211)")
|
downgrade_reasons.append("review mutation proof missing or incomplete (#211)")
|
||||||
downgrade_reasons.extend(review_mutation.get("reasons", []))
|
downgrade_reasons.extend(review_mutation.get("reasons", []))
|
||||||
|
if not worktree_proven:
|
||||||
|
downgrade_reasons.append(
|
||||||
|
"reviewer worktree safety proof missing or failed (#233)"
|
||||||
|
)
|
||||||
|
downgrade_reasons.extend(worktree.get("reasons", []))
|
||||||
|
|
||||||
merge_allowed = (
|
merge_allowed = (
|
||||||
identity_eligible
|
identity_eligible
|
||||||
@@ -782,6 +797,7 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
|
|||||||
and validation.get("verdict") != "invalid"
|
and validation.get("verdict") != "invalid"
|
||||||
# #179: no merge without a proven final live-state recheck.
|
# #179: no merge without a proven final live-state recheck.
|
||||||
and live_state_proven
|
and live_state_proven
|
||||||
|
and worktree_proven
|
||||||
)
|
)
|
||||||
|
|
||||||
violations = []
|
violations = []
|
||||||
@@ -825,6 +841,11 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
|
|||||||
"live_state_recheck_proven": live_state_proven,
|
"live_state_recheck_proven": live_state_proven,
|
||||||
"role_boundary_clean": role_boundary_clean,
|
"role_boundary_clean": role_boundary_clean,
|
||||||
"review_mutation_complete": review_mutation_complete,
|
"review_mutation_complete": review_mutation_complete,
|
||||||
|
"worktree_proof_proven": worktree_proven,
|
||||||
|
"worktree_scratch_used": bool(worktree.get("scratch_used")),
|
||||||
|
"unrelated_mutations_avoided": bool(
|
||||||
|
worktree.get("unrelated_mutations_avoided")
|
||||||
|
),
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
@@ -971,6 +992,12 @@ HANDOFF_ROLE_FIELDS = {
|
|||||||
("Selected PR", ("selected pr",)),
|
("Selected PR", ("selected pr",)),
|
||||||
("Reviewer eligibility", ("reviewer eligibility", "eligibility")),
|
("Reviewer eligibility", ("reviewer eligibility", "eligibility")),
|
||||||
("Pinned reviewed head", ("pinned reviewed head", "pinned head")),
|
("Pinned reviewed head", ("pinned reviewed head", "pinned head")),
|
||||||
|
("Worktree path", ("worktree path", "starting worktree path")),
|
||||||
|
("Worktree dirty", ("worktree dirty", "whether worktree was dirty")),
|
||||||
|
("Scratch worktree used", ("scratch worktree used", "scratch clone used",
|
||||||
|
"scratch worktree")),
|
||||||
|
("Unrelated local mutations", ("unrelated local mutations",
|
||||||
|
"unrelated files modified")),
|
||||||
("Review decision", ("review decision", "decision")),
|
("Review decision", ("review decision", "decision")),
|
||||||
("Merge result", ("merge result",)),
|
("Merge result", ("merge result",)),
|
||||||
("Linked issue status", ("linked issue status", "linked issue")),
|
("Linked issue status", ("linked issue status", "linked issue")),
|
||||||
@@ -993,6 +1020,16 @@ HANDOFF_ROLE_FIELDS = {
|
|||||||
("Inventory completeness", ("inventory complete", "inventory scoped",
|
("Inventory completeness", ("inventory complete", "inventory scoped",
|
||||||
"inventory completeness")),
|
"inventory completeness")),
|
||||||
),
|
),
|
||||||
|
"issue_filing": (
|
||||||
|
("Issue created or updated", (
|
||||||
|
"issue created or updated",
|
||||||
|
"issue created",
|
||||||
|
"issue updated",
|
||||||
|
"created issue",
|
||||||
|
"updated issue",
|
||||||
|
)),
|
||||||
|
("Related issues", ("related issues",)),
|
||||||
|
),
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
@@ -1284,6 +1321,391 @@ def pr_inventory_trust_gate(
|
|||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
|
def _split_repo_slug(full_repo: str) -> tuple[str | None, str | None]:
|
||||||
|
parts = (full_repo or "").split("/", 1)
|
||||||
|
if len(parts) == 2:
|
||||||
|
return parts[0].strip() or None, parts[1].strip() or None
|
||||||
|
return None, None
|
||||||
|
|
||||||
|
|
||||||
|
def assess_reviewer_queue_inventory(
|
||||||
|
repo_reports: list[dict] | None,
|
||||||
|
required_repos: list[str] | None = None,
|
||||||
|
*,
|
||||||
|
user_context: str | None = None,
|
||||||
|
) -> dict:
|
||||||
|
"""Canonical reviewer queue path: completeness plus per-repo trust gates (#196).
|
||||||
|
|
||||||
|
Any repository reporting ``open_pr_count == 0`` must pass
|
||||||
|
``pr_inventory_trust_gate`` with ``trusted_empty`` before an empty-queue
|
||||||
|
claim is allowed. A bare ``[]`` from ``gitea_list_prs`` is never sufficient.
|
||||||
|
"""
|
||||||
|
required = list(required_repos or [
|
||||||
|
"Scaled-Tech-Consulting/Gitea-Tools",
|
||||||
|
"Scaled-Tech-Consulting/mcp-control-plane",
|
||||||
|
])
|
||||||
|
completeness = assess_inventory_completeness(repo_reports, required)
|
||||||
|
|
||||||
|
trust_gates: dict[str, dict] = {}
|
||||||
|
blockers: list[str] = []
|
||||||
|
can_claim_empty = bool(completeness.get("complete"))
|
||||||
|
|
||||||
|
for report in repo_reports or []:
|
||||||
|
repo = (report.get("repo") or "").strip()
|
||||||
|
count = report.get("open_pr_count")
|
||||||
|
if not isinstance(count, int) or count != 0:
|
||||||
|
continue
|
||||||
|
|
||||||
|
org, repo_name = _split_repo_slug(repo)
|
||||||
|
list_response = report.get("list_prs_response")
|
||||||
|
if list_response is None:
|
||||||
|
list_response = []
|
||||||
|
|
||||||
|
gate = pr_inventory_trust_gate(
|
||||||
|
list_response,
|
||||||
|
remote=report.get("remote"),
|
||||||
|
org=org,
|
||||||
|
repo=repo_name,
|
||||||
|
state=report.get("state_filter"),
|
||||||
|
authenticated_profile=report.get("authenticated_profile"),
|
||||||
|
local_remote_url=report.get("local_remote_url"),
|
||||||
|
user_context=user_context or report.get("user_context"),
|
||||||
|
corroboration_open_pr_counter=report.get(
|
||||||
|
"corroboration_open_pr_counter"
|
||||||
|
),
|
||||||
|
has_finality_metadata=report.get("pagination_complete") is True,
|
||||||
|
)
|
||||||
|
trust_gates[repo] = gate
|
||||||
|
status = gate.get("status")
|
||||||
|
if status != "trusted_empty":
|
||||||
|
can_claim_empty = False
|
||||||
|
blockers.append(
|
||||||
|
f"repository '{repo}': empty PR list trust gate is "
|
||||||
|
f"'{status}'; cannot claim 'no open PRs'"
|
||||||
|
)
|
||||||
|
blockers.extend(gate.get("reasons") or [])
|
||||||
|
|
||||||
|
return {
|
||||||
|
"complete": bool(completeness.get("complete")),
|
||||||
|
"can_claim_empty_queue": can_claim_empty,
|
||||||
|
"can_claim_exhaustive": (
|
||||||
|
bool(completeness.get("can_claim_exhaustive")) and can_claim_empty
|
||||||
|
),
|
||||||
|
"inventory_reasons": list(completeness.get("reasons") or []),
|
||||||
|
"trust_gates": trust_gates,
|
||||||
|
"blockers": blockers,
|
||||||
|
"reasons": list(completeness.get("reasons") or []) + blockers,
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
def format_pr_inventory_trust_gate_report(gate: dict) -> list[str]:
|
||||||
|
"""Render trust-gate lines for MCP inventory output."""
|
||||||
|
lines = [f"pr_inventory_trust_gate.status: {gate.get('status', 'unknown')}"]
|
||||||
|
if gate.get("corroborated"):
|
||||||
|
lines.append("pr_inventory_trust_gate.corroborated: true")
|
||||||
|
for reason in gate.get("reasons") or []:
|
||||||
|
lines.append(f"pr_inventory_trust_gate.reason: {reason}")
|
||||||
|
return lines
|
||||||
|
|
||||||
|
|
||||||
|
_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 in text and f"no {absent}" not in text:
|
||||||
|
if any(
|
||||||
|
phrase in text
|
||||||
|
for phrase in (
|
||||||
|
f"no {absent}",
|
||||||
|
f"no {absent}s",
|
||||||
|
f"without {absent}",
|
||||||
|
"none performed",
|
||||||
|
"none.",
|
||||||
|
)
|
||||||
|
):
|
||||||
|
continue
|
||||||
|
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():
|
||||||
|
if result.get("verdict") == "missing":
|
||||||
|
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):
|
def assess_duplicate_search_proof(report_text, matches):
|
||||||
"""#207: reject LLM duplicate summaries that omit known title matches."""
|
"""#207: reject LLM duplicate summaries that omit known title matches."""
|
||||||
return issue_duplicate_gate.assess_duplicate_search_proof(
|
return issue_duplicate_gate.assess_duplicate_search_proof(
|
||||||
|
|||||||
@@ -0,0 +1,219 @@
|
|||||||
|
"""Fail-closed reviewer worktree and local-git safety proofs (#233).
|
||||||
|
|
||||||
|
Reviewer sessions must never stash, reset, or otherwise manipulate unrelated
|
||||||
|
local changes from another session. When the active worktree has dirty tracked
|
||||||
|
files outside the PR scope, the workflow must stop or switch to a disposable
|
||||||
|
scratch worktree (``scripts/worktree-review``).
|
||||||
|
"""
|
||||||
|
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import re
|
||||||
|
import shlex
|
||||||
|
|
||||||
|
# Subcommands that mutate unrelated local state — forbidden for reviewers.
|
||||||
|
_FORBIDDEN_REVIEWER_GIT = re.compile(
|
||||||
|
r"\bgit\b(?:\s+(?:-C\s+\S+\s+)?)?"
|
||||||
|
r"(?:stash(?:\s+(?:push|pop|drop|apply|clear|list))?|"
|
||||||
|
r"checkout\s+--|"
|
||||||
|
r"restore\s+|"
|
||||||
|
r"reset(?:\s+(?:--hard|--soft|--mixed|--merge))?|"
|
||||||
|
r"clean(?:\s+(?:-f|-fd|-fdx|-x|-d|-n))*|"
|
||||||
|
r"cherry-pick|"
|
||||||
|
r"rebase|"
|
||||||
|
r"merge|"
|
||||||
|
r"commit(?:\s+(?:--amend|-a|-am))?|"
|
||||||
|
r"push(?:\s+(?:--force|--force-with-lease))?|"
|
||||||
|
r"branch\s+-[dD]|"
|
||||||
|
r"worktree\s+remove)",
|
||||||
|
re.IGNORECASE,
|
||||||
|
)
|
||||||
|
|
||||||
|
# Read-only git operations reviewers may use for validation.
|
||||||
|
_READONLY_REVIEWER_GIT = re.compile(
|
||||||
|
r"\bgit\b(?:\s+(?:-C\s+\S+\s+)?)?"
|
||||||
|
r"(?:fetch|status|diff|log|show|rev-parse|branch(?:\s+--show-current)?|"
|
||||||
|
r"worktree\s+list|worktree\s+add)",
|
||||||
|
re.IGNORECASE,
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def parse_dirty_tracked_files(porcelain: str) -> list[str]:
|
||||||
|
"""Return tracked paths with local modifications from ``git status --porcelain``.
|
||||||
|
|
||||||
|
Untracked entries (``??``) are ignored — they do not block reviewer work
|
||||||
|
when a scratch worktree is used, and authors may have unrelated untracked
|
||||||
|
files without implying reviewer interference.
|
||||||
|
"""
|
||||||
|
paths: list[str] = []
|
||||||
|
for line in (porcelain or "").splitlines():
|
||||||
|
if not line or len(line) < 4:
|
||||||
|
continue
|
||||||
|
if line.startswith("??"):
|
||||||
|
continue
|
||||||
|
path = line[3:].strip()
|
||||||
|
if " -> " in path:
|
||||||
|
path = path.split(" -> ", 1)[1].strip()
|
||||||
|
if path:
|
||||||
|
paths.append(path)
|
||||||
|
return paths
|
||||||
|
|
||||||
|
|
||||||
|
def files_outside_pr_scope(
|
||||||
|
dirty_files: list[str] | None,
|
||||||
|
pr_scope_files: list[str] | None,
|
||||||
|
) -> list[str]:
|
||||||
|
"""Dirty tracked files not explained by the PR diff file set."""
|
||||||
|
dirty = [p for p in (dirty_files or []) if p]
|
||||||
|
scope = {p for p in (pr_scope_files or []) if p}
|
||||||
|
if not dirty:
|
||||||
|
return []
|
||||||
|
if not scope:
|
||||||
|
return list(dirty)
|
||||||
|
return [path for path in dirty if path not in scope]
|
||||||
|
|
||||||
|
|
||||||
|
def is_forbidden_reviewer_git_command(command: str) -> bool:
|
||||||
|
"""True when a shell command would mutate unrelated local/remote git state."""
|
||||||
|
text = (command or "").strip()
|
||||||
|
if not text:
|
||||||
|
return False
|
||||||
|
return bool(_FORBIDDEN_REVIEWER_GIT.search(text))
|
||||||
|
|
||||||
|
|
||||||
|
def is_readonly_reviewer_git_command(command: str) -> bool:
|
||||||
|
"""True when the command is an explicitly allowed read-only git operation."""
|
||||||
|
text = (command or "").strip()
|
||||||
|
if not text:
|
||||||
|
return False
|
||||||
|
if is_forbidden_reviewer_git_command(text):
|
||||||
|
return False
|
||||||
|
return bool(_READONLY_REVIEWER_GIT.search(text))
|
||||||
|
|
||||||
|
|
||||||
|
def assess_reviewer_git_command_log(commands: list[str] | None) -> dict:
|
||||||
|
"""Fail closed when reviewer shell history includes forbidden git mutations."""
|
||||||
|
forbidden = [
|
||||||
|
cmd for cmd in (commands or []) if is_forbidden_reviewer_git_command(cmd)
|
||||||
|
]
|
||||||
|
if forbidden:
|
||||||
|
return {
|
||||||
|
"proven": False,
|
||||||
|
"block": True,
|
||||||
|
"forbidden_commands": forbidden,
|
||||||
|
"reasons": [
|
||||||
|
"reviewer workflow executed forbidden local git mutation: "
|
||||||
|
f"{cmd!r}"
|
||||||
|
for cmd in forbidden
|
||||||
|
],
|
||||||
|
"safe_next_action": (
|
||||||
|
"stop; report worktree interference; do not stash/reset/checkout "
|
||||||
|
"unrelated files — use scripts/worktree-review instead"
|
||||||
|
),
|
||||||
|
}
|
||||||
|
return {
|
||||||
|
"proven": True,
|
||||||
|
"block": False,
|
||||||
|
"forbidden_commands": [],
|
||||||
|
"reasons": [],
|
||||||
|
"safe_next_action": "proceed",
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
def assess_reviewer_worktree_proof(proof: dict | None) -> dict:
|
||||||
|
"""Evaluate reviewer worktree safety before checkout/diff/validation/review.
|
||||||
|
|
||||||
|
*proof* keys:
|
||||||
|
- ``worktree_path`` (required)
|
||||||
|
- ``porcelain_status`` or ``dirty_files``
|
||||||
|
- ``pr_scope_files`` (paths in the PR diff)
|
||||||
|
- ``scratch_used`` (bool)
|
||||||
|
- ``scratch_path`` (when scratch_used)
|
||||||
|
- ``git_commands`` (shell commands executed this session)
|
||||||
|
- ``unrelated_mutations_claimed`` (bool) — stash/reset/drop reported
|
||||||
|
"""
|
||||||
|
proof = dict(proof or {})
|
||||||
|
reasons: list[str] = []
|
||||||
|
worktree_path = (proof.get("worktree_path") or "").strip()
|
||||||
|
if not worktree_path:
|
||||||
|
reasons.append("reviewer worktree path not reported; fail closed")
|
||||||
|
|
||||||
|
if proof.get("dirty_files") is not None:
|
||||||
|
dirty_files = list(proof.get("dirty_files") or [])
|
||||||
|
else:
|
||||||
|
dirty_files = parse_dirty_tracked_files(proof.get("porcelain_status") or "")
|
||||||
|
|
||||||
|
pr_scope = list(proof.get("pr_scope_files") or [])
|
||||||
|
unrelated = files_outside_pr_scope(dirty_files, pr_scope)
|
||||||
|
scratch_used = bool(proof.get("scratch_used"))
|
||||||
|
scratch_path = (proof.get("scratch_path") or "").strip()
|
||||||
|
|
||||||
|
is_dirty = bool(dirty_files)
|
||||||
|
unrelated_dirty = bool(unrelated)
|
||||||
|
|
||||||
|
if unrelated_dirty and not scratch_used:
|
||||||
|
reasons.append(
|
||||||
|
"worktree has dirty tracked files outside PR scope "
|
||||||
|
f"({', '.join(unrelated)}); stop or use a scratch worktree"
|
||||||
|
)
|
||||||
|
if scratch_used and not scratch_path:
|
||||||
|
reasons.append(
|
||||||
|
"scratch worktree was used but scratch_path was not reported"
|
||||||
|
)
|
||||||
|
if proof.get("unrelated_mutations_claimed"):
|
||||||
|
reasons.append(
|
||||||
|
"reviewer reported stash/reset/checkout cleanup of unrelated "
|
||||||
|
"local changes; this is forbidden"
|
||||||
|
)
|
||||||
|
|
||||||
|
command_assessment = assess_reviewer_git_command_log(
|
||||||
|
list(proof.get("git_commands") or [])
|
||||||
|
)
|
||||||
|
if command_assessment["block"]:
|
||||||
|
reasons.extend(command_assessment["reasons"])
|
||||||
|
|
||||||
|
proven = not reasons
|
||||||
|
return {
|
||||||
|
"proven": proven,
|
||||||
|
"block": not proven,
|
||||||
|
"reasons": reasons,
|
||||||
|
"worktree_path": worktree_path or None,
|
||||||
|
"is_dirty": is_dirty,
|
||||||
|
"dirty_files": dirty_files,
|
||||||
|
"unrelated_dirty_files": unrelated,
|
||||||
|
"scratch_used": scratch_used,
|
||||||
|
"scratch_path": scratch_path or None,
|
||||||
|
"unrelated_mutations_avoided": not bool(
|
||||||
|
proof.get("unrelated_mutations_claimed")
|
||||||
|
or command_assessment.get("forbidden_commands")
|
||||||
|
),
|
||||||
|
"safe_next_action": (
|
||||||
|
"proceed"
|
||||||
|
if proven
|
||||||
|
else command_assessment.get("safe_next_action")
|
||||||
|
or "stop; use scripts/worktree-review or report dirty worktree"
|
||||||
|
),
|
||||||
|
"forbidden_commands": command_assessment.get("forbidden_commands", []),
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
def assess_author_worktree_continuity(proof: dict | None) -> dict:
|
||||||
|
"""Authors may keep dirty feature worktrees; reviewers may not manipulate them.
|
||||||
|
|
||||||
|
This helper only proves the task role is author when dirty unrelated files
|
||||||
|
exist — it does not grant reviewers an exception.
|
||||||
|
"""
|
||||||
|
proof = dict(proof or {})
|
||||||
|
role = (proof.get("task_role") or "").strip().lower()
|
||||||
|
dirty_files = list(proof.get("dirty_files") or [])
|
||||||
|
if role == "author" and dirty_files:
|
||||||
|
return {
|
||||||
|
"allowed": True,
|
||||||
|
"reasons": [
|
||||||
|
"author task may continue with dirty tracked files in its own "
|
||||||
|
"worktree; reviewer interference rules do not apply"
|
||||||
|
],
|
||||||
|
}
|
||||||
|
if role == "reviewer" and dirty_files:
|
||||||
|
return assess_reviewer_worktree_proof(proof)
|
||||||
|
return {"allowed": True, "reasons": []}
|
||||||
@@ -377,6 +377,12 @@ Role-specific fields (append to the compact block):
|
|||||||
- review/merge tasks: `Selected PR:`, `Reviewer eligibility:`,
|
- review/merge tasks: `Selected PR:`, `Reviewer eligibility:`,
|
||||||
`Pinned reviewed head:`, `Review decision:`, `Merge result:`,
|
`Pinned reviewed head:`, `Review decision:`, `Merge result:`,
|
||||||
`Linked issue status:`, `Cleanup status:`
|
`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:`,
|
- author tasks: `Selected issue:`, `Claim/comment status:`,
|
||||||
`PR number opened:`, `No review/merge:` (explicit confirmation)
|
`PR number opened:`, `No review/merge:` (explicit confirmation)
|
||||||
- queue/inventory tasks: `Repositories checked:`, `Open PR counts:`,
|
- queue/inventory tasks: `Repositories checked:`, `Open PR counts:`,
|
||||||
|
|||||||
@@ -17,9 +17,30 @@ Repo name disambiguation (Gitea-Tools blind review hardening):
|
|||||||
configured repos were not checked. This is not a complete queue inventory."
|
configured repos were not checked. This is not a complete queue inventory."
|
||||||
- A single-repo "no open PRs" result MUST NOT be reported as global "no open PRs"
|
- A single-repo "no open PRs" result MUST NOT be reported as global "no open PRs"
|
||||||
if the other configured repo was not inventoried.
|
if the other configured repo was not inventoried.
|
||||||
|
- PR inventory trust gate (#196): before reporting "no open PRs" or "queue empty",
|
||||||
|
the workflow must run `pr_inventory_trust_gate` (via the live inventory path or
|
||||||
|
`review_proofs.assess_reviewer_queue_inventory`). Only `trusted_empty` allows a
|
||||||
|
clean empty-queue stop. Report `pr_inventory_trust_gate.status`, reasons, and
|
||||||
|
corroboration in the final report. A bare `[]` from `gitea_list_prs` is never
|
||||||
|
sufficient proof.
|
||||||
|
- Empty-queue report wall (#198): if the final report claims "no open PRs",
|
||||||
|
"queue empty", or "nothing to review", it must include verbatim:
|
||||||
|
`pr_inventory_trust_gate.status`, trust-gate reasons, corroboration,
|
||||||
|
remote/owner/repo/state filter, and the inventory MCP profile. A recent merge
|
||||||
|
commit is not valid corroboration. Author-bound sessions must not present
|
||||||
|
reviewer queue inventory as a reviewer decision.
|
||||||
|
|
||||||
Rules (llm-project-workflow):
|
Rules (llm-project-workflow):
|
||||||
- Review in a SEPARATE detached review worktree, never the author's folder.
|
- Review in a SEPARATE detached review worktree, never the author's folder.
|
||||||
|
- Worktree safety (#233): before checkout, diff, validation, review, or merge,
|
||||||
|
report the starting worktree path and whether it was dirty. If unrelated
|
||||||
|
tracked files exist outside the PR scope, STOP or run
|
||||||
|
`scripts/worktree-review <pr-head-branch>` and validate in the scratch path.
|
||||||
|
NEVER run `git stash`, `git stash pop/drop`, `git checkout --`, `git reset`,
|
||||||
|
or `git clean` to manage another session's dirty files.
|
||||||
|
- Final report must state: Worktree path, Worktree dirty (yes/no),
|
||||||
|
Scratch worktree used (yes/no + path if yes), and confirm no unrelated local
|
||||||
|
files were modified, stashed, reset, or dropped.
|
||||||
- You must NOT be the PR author. If the authenticated user == PR author, stop.
|
- You must NOT be the PR author. If the authenticated user == PR author, stop.
|
||||||
A different LLM-Agent-SHA does NOT make you a different actor — only a
|
A different LLM-Agent-SHA does NOT make you a different actor — only a
|
||||||
different authenticated Gitea user does (docs/llm-agent-sha.md).
|
different authenticated Gitea user does (docs/llm-agent-sha.md).
|
||||||
|
|||||||
@@ -148,6 +148,19 @@ class TestCapabilityStopTerminal(unittest.TestCase):
|
|||||||
)
|
)
|
||||||
self.assertFalse(result["pure"])
|
self.assertFalse(result["pure"])
|
||||||
|
|
||||||
|
def test_empty_queue_with_parsed_trusted_status_passes_gate_check(self):
|
||||||
|
report = (
|
||||||
|
"Cannot perform reviewer task under current profile. "
|
||||||
|
"No reviewer mutations performed.\n"
|
||||||
|
"Repository: Scaled-Tech-Consulting/Gitea-Tools\n"
|
||||||
|
"pr_inventory_trust_gate.status: trusted_empty\n"
|
||||||
|
"pr_inventory_trust_gate.corroborated: true\n"
|
||||||
|
"Inventory profile: prgs-reviewer\n"
|
||||||
|
"No open PRs in queue."
|
||||||
|
)
|
||||||
|
result = assess_capability_stop_terminal_report(report)
|
||||||
|
self.assertTrue(result["pure"])
|
||||||
|
|
||||||
def test_pure_terminal_report_passes(self):
|
def test_pure_terminal_report_passes(self):
|
||||||
report = (
|
report = (
|
||||||
"Cannot perform reviewer task under current profile. "
|
"Cannot perform reviewer task under current profile. "
|
||||||
|
|||||||
@@ -261,12 +261,79 @@ class TestPRQueueInventory(unittest.TestCase):
|
|||||||
for call in mock_api.call_args_list:
|
for call in mock_api.call_args_list:
|
||||||
self.assertEqual(call.args[0], "GET")
|
self.assertEqual(call.args[0], "GET")
|
||||||
|
|
||||||
|
@patch("mcp_server._local_git_remote_url")
|
||||||
@patch("mcp_server.api_get_all")
|
@patch("mcp_server.api_get_all")
|
||||||
@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("mcp_server.get_profile")
|
@patch("mcp_server.get_profile")
|
||||||
def test_author_profiles_cannot_approve_request_changes_merge_or_bypass_gates(self, mock_get_profile, _auth, mock_api, mock_get_all):
|
def test_empty_inventory_runs_trust_gate_and_blocks_without_trusted_empty(
|
||||||
|
self, mock_get_profile, _auth, mock_api, mock_get_all, mock_local_url
|
||||||
|
):
|
||||||
|
mock_get_profile.return_value = {
|
||||||
|
"profile_name": "gitea-author",
|
||||||
|
"allowed_operations": ["read"],
|
||||||
|
"forbidden_operations": [],
|
||||||
|
"base_url": None,
|
||||||
|
}
|
||||||
|
mock_get_all.return_value = []
|
||||||
|
mock_api.return_value = {"login": "jcwalker3"}
|
||||||
|
mock_local_url.return_value = None
|
||||||
|
|
||||||
|
result = gitea_review_pr(
|
||||||
|
pr_number=1,
|
||||||
|
event="APPROVE",
|
||||||
|
remote="prgs",
|
||||||
|
org="Scaled-Tech-Consulting",
|
||||||
|
repo="Gitea-Tools",
|
||||||
|
)
|
||||||
|
self.assertFalse(result["success"])
|
||||||
|
msg = result["message"]
|
||||||
|
self.assertIn("pr_inventory_trust_gate.status: untrusted_empty", msg)
|
||||||
|
self.assertIn("Empty-queue claim blocked", msg)
|
||||||
|
self.assertNotIn("Open PRs found: 0 (trusted_empty)", msg)
|
||||||
|
|
||||||
|
@patch("mcp_server._local_git_remote_url")
|
||||||
|
@patch("mcp_server.api_get_all")
|
||||||
|
@patch("mcp_server.api_request")
|
||||||
|
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
||||||
|
@patch("mcp_server.get_profile")
|
||||||
|
def test_empty_inventory_trusted_empty_when_gate_passes(
|
||||||
|
self, mock_get_profile, _auth, mock_api, mock_get_all, mock_local_url
|
||||||
|
):
|
||||||
|
mock_get_profile.return_value = {
|
||||||
|
"profile_name": "gitea-author",
|
||||||
|
"allowed_operations": ["read"],
|
||||||
|
"forbidden_operations": [],
|
||||||
|
"base_url": None,
|
||||||
|
}
|
||||||
|
mock_get_all.return_value = []
|
||||||
|
mock_api.return_value = {"login": "jcwalker3"}
|
||||||
|
mock_local_url.return_value = (
|
||||||
|
"https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools.git"
|
||||||
|
)
|
||||||
|
|
||||||
|
result = gitea_review_pr(
|
||||||
|
pr_number=1,
|
||||||
|
event="APPROVE",
|
||||||
|
remote="prgs",
|
||||||
|
org="Scaled-Tech-Consulting",
|
||||||
|
repo="Gitea-Tools",
|
||||||
|
)
|
||||||
|
self.assertFalse(result["success"])
|
||||||
|
msg = result["message"]
|
||||||
|
self.assertIn("pr_inventory_trust_gate.status: trusted_empty", msg)
|
||||||
|
self.assertIn("Open PRs found: 0 (trusted_empty)", msg)
|
||||||
|
|
||||||
|
@patch("mcp_server._local_git_remote_url")
|
||||||
|
@patch("mcp_server.api_get_all")
|
||||||
|
@patch("mcp_server.api_request")
|
||||||
|
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
||||||
|
@patch("mcp_server.get_profile")
|
||||||
|
def test_author_profiles_cannot_approve_request_changes_merge_or_bypass_gates(self, mock_get_profile, _auth, mock_api, mock_get_all, mock_local_url):
|
||||||
"""Author profiles still cannot approve, request_changes, merge, or bypass gates even with inventory."""
|
"""Author profiles still cannot approve, request_changes, merge, or bypass gates even with inventory."""
|
||||||
|
mock_local_url.return_value = (
|
||||||
|
"https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools.git"
|
||||||
|
)
|
||||||
for event in ["APPROVE", "REQUEST_CHANGES"]:
|
for event in ["APPROVE", "REQUEST_CHANGES"]:
|
||||||
mock_get_profile.return_value = {
|
mock_get_profile.return_value = {
|
||||||
"profile_name": "gitea-author",
|
"profile_name": "gitea-author",
|
||||||
|
|||||||
@@ -26,6 +26,10 @@ from review_proofs import ( # noqa: E402
|
|||||||
assess_capability_proof,
|
assess_capability_proof,
|
||||||
assess_controller_handoff,
|
assess_controller_handoff,
|
||||||
assess_inventory_completeness,
|
assess_inventory_completeness,
|
||||||
|
assess_issue_filing_final_report,
|
||||||
|
assess_issue_filing_mutation_capability,
|
||||||
|
assess_issue_filing_sha_evidence,
|
||||||
|
assess_reviewer_queue_inventory,
|
||||||
assess_live_state_recheck,
|
assess_live_state_recheck,
|
||||||
assess_review_mutation_final_report,
|
assess_review_mutation_final_report,
|
||||||
assess_role_boundary,
|
assess_role_boundary,
|
||||||
@@ -183,6 +187,24 @@ def _good_review_mutation():
|
|||||||
return assess_review_mutation_final_report(report, lock)
|
return assess_review_mutation_final_report(report, lock)
|
||||||
|
|
||||||
|
|
||||||
|
def _good_worktree(**overrides):
|
||||||
|
proof = {
|
||||||
|
"worktree_path": "/repo/branches/review-feat-issue-224",
|
||||||
|
"porcelain_status": "",
|
||||||
|
"pr_scope_files": ["docs/wiki/Repositories.md"],
|
||||||
|
"scratch_used": True,
|
||||||
|
"scratch_path": "/repo/branches/review-feat-issue-224",
|
||||||
|
"git_commands": [
|
||||||
|
"git fetch prgs master feat/issue-224-wiki-proof-refresh",
|
||||||
|
"git diff prgs/master...prgs/feat/issue-224-wiki-proof-refresh",
|
||||||
|
],
|
||||||
|
}
|
||||||
|
proof.update(overrides)
|
||||||
|
from reviewer_worktree import assess_reviewer_worktree_proof # noqa: E402
|
||||||
|
|
||||||
|
return assess_reviewer_worktree_proof(proof)
|
||||||
|
|
||||||
|
|
||||||
def _good_role_boundary_179(**overrides):
|
def _good_role_boundary_179(**overrides):
|
||||||
kwargs = {
|
kwargs = {
|
||||||
"task_role": "reviewer",
|
"task_role": "reviewer",
|
||||||
@@ -595,6 +617,13 @@ class TestFinalReport(unittest.TestCase):
|
|||||||
"controller_handoff": _good_handoff(),
|
"controller_handoff": _good_handoff(),
|
||||||
"capability_proof": _good_capability_proof(),
|
"capability_proof": _good_capability_proof(),
|
||||||
"sweep_proof": _good_secret_sweep(),
|
"sweep_proof": _good_secret_sweep(),
|
||||||
|
"worktree_proof": {
|
||||||
|
"worktree_path": "/repo/branches/review-feat-issue-224",
|
||||||
|
"porcelain_status": "",
|
||||||
|
"pr_scope_files": ["docs/wiki/Repositories.md"],
|
||||||
|
"scratch_used": True,
|
||||||
|
"scratch_path": "/repo/branches/review-feat-issue-224",
|
||||||
|
},
|
||||||
}
|
}
|
||||||
kwargs.update(overrides)
|
kwargs.update(overrides)
|
||||||
return build_final_report(**kwargs)
|
return build_final_report(**kwargs)
|
||||||
@@ -899,12 +928,17 @@ class TestControllerHandoff(unittest.TestCase):
|
|||||||
result = assess_controller_handoff(self.BASE_HANDOFF, role="review")
|
result = assess_controller_handoff(self.BASE_HANDOFF, role="review")
|
||||||
self.assertEqual(result["verdict"], "incomplete")
|
self.assertEqual(result["verdict"], "incomplete")
|
||||||
self.assertIn("Pinned reviewed head", result["missing_fields"])
|
self.assertIn("Pinned reviewed head", result["missing_fields"])
|
||||||
|
self.assertIn("Worktree path", result["missing_fields"])
|
||||||
self.assertIn("Merge result", result["missing_fields"])
|
self.assertIn("Merge result", result["missing_fields"])
|
||||||
|
|
||||||
complete = self.BASE_HANDOFF + "\n" + "\n".join([
|
complete = self.BASE_HANDOFF + "\n" + "\n".join([
|
||||||
"- Selected PR: #999",
|
"- Selected PR: #999",
|
||||||
"- Reviewer eligibility: passed",
|
"- Reviewer eligibility: passed",
|
||||||
"- Pinned reviewed head: 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9",
|
"- Pinned reviewed head: 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9",
|
||||||
|
"- Worktree path: /repo/branches/review-pr-999",
|
||||||
|
"- Worktree dirty: no",
|
||||||
|
"- Scratch worktree used: yes (/repo/branches/review-pr-999)",
|
||||||
|
"- Unrelated local mutations: none",
|
||||||
"- Review decision: approve",
|
"- Review decision: approve",
|
||||||
"- Merge result: merged",
|
"- Merge result: merged",
|
||||||
"- Linked issue status: closed",
|
"- Linked issue status: closed",
|
||||||
@@ -1163,6 +1197,74 @@ class TestPRInventoryTrustGate(unittest.TestCase):
|
|||||||
self.assertTrue(res["corroborated"])
|
self.assertTrue(res["corroborated"])
|
||||||
|
|
||||||
|
|
||||||
|
class TestAssessReviewerQueueInventory(unittest.TestCase):
|
||||||
|
"""Issue #196: trust gate wired into canonical queue inventory."""
|
||||||
|
|
||||||
|
def _repo_report(self, **overrides):
|
||||||
|
report = {
|
||||||
|
"repo": "Scaled-Tech-Consulting/Gitea-Tools",
|
||||||
|
"state_filter": "open",
|
||||||
|
"pagination_complete": True,
|
||||||
|
"open_pr_count": 0,
|
||||||
|
"list_prs_response": [],
|
||||||
|
"remote": "prgs",
|
||||||
|
"authenticated_profile": {
|
||||||
|
"profile_name": "prgs-reviewer",
|
||||||
|
"allowed_operations": ["read", "gitea.read"],
|
||||||
|
},
|
||||||
|
"local_remote_url": (
|
||||||
|
"https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools.git"
|
||||||
|
),
|
||||||
|
}
|
||||||
|
report.update(overrides)
|
||||||
|
return report
|
||||||
|
|
||||||
|
def test_empty_without_corroboration_blocks_empty_queue_claim(self):
|
||||||
|
result = assess_reviewer_queue_inventory([
|
||||||
|
self._repo_report(pagination_complete=False),
|
||||||
|
self._repo_report(
|
||||||
|
repo="Scaled-Tech-Consulting/mcp-control-plane",
|
||||||
|
open_pr_count=0,
|
||||||
|
pagination_complete=False,
|
||||||
|
),
|
||||||
|
])
|
||||||
|
self.assertFalse(result["can_claim_empty_queue"])
|
||||||
|
self.assertIn("untrusted_empty", str(result["trust_gates"]))
|
||||||
|
|
||||||
|
def test_trusted_empty_with_finality_allows_empty_queue_claim(self):
|
||||||
|
result = assess_reviewer_queue_inventory([
|
||||||
|
self._repo_report(),
|
||||||
|
self._repo_report(
|
||||||
|
repo="Scaled-Tech-Consulting/mcp-control-plane",
|
||||||
|
local_remote_url=(
|
||||||
|
"https://gitea.prgs.cc/Scaled-Tech-Consulting/"
|
||||||
|
"mcp-control-plane.git"
|
||||||
|
),
|
||||||
|
),
|
||||||
|
])
|
||||||
|
self.assertTrue(result["can_claim_empty_queue"])
|
||||||
|
self.assertTrue(result["can_claim_exhaustive"])
|
||||||
|
|
||||||
|
def test_user_context_indicating_open_prs_blocks_empty_claim(self):
|
||||||
|
result = assess_reviewer_queue_inventory(
|
||||||
|
[self._repo_report()],
|
||||||
|
user_context="please review open PR #195 in the queue",
|
||||||
|
)
|
||||||
|
self.assertFalse(result["can_claim_empty_queue"])
|
||||||
|
self.assertTrue(result["blockers"])
|
||||||
|
|
||||||
|
def test_nonempty_inventory_skips_empty_trust_gate_block(self):
|
||||||
|
result = assess_reviewer_queue_inventory([
|
||||||
|
self._repo_report(open_pr_count=2),
|
||||||
|
self._repo_report(
|
||||||
|
repo="Scaled-Tech-Consulting/mcp-control-plane",
|
||||||
|
open_pr_count=1,
|
||||||
|
),
|
||||||
|
])
|
||||||
|
self.assertTrue(result["complete"])
|
||||||
|
self.assertEqual(result["trust_gates"], {})
|
||||||
|
|
||||||
|
|
||||||
class TestCapabilityEvidence(unittest.TestCase):
|
class TestCapabilityEvidence(unittest.TestCase):
|
||||||
"""#179 gap 1: capability claims need exact evidence."""
|
"""#179 gap 1: capability claims need exact evidence."""
|
||||||
|
|
||||||
@@ -1306,6 +1408,13 @@ class TestFinalReport179Bar(unittest.TestCase):
|
|||||||
"controller_handoff": _good_handoff(),
|
"controller_handoff": _good_handoff(),
|
||||||
"capability_proof": _good_capability_proof(),
|
"capability_proof": _good_capability_proof(),
|
||||||
"sweep_proof": _good_secret_sweep(),
|
"sweep_proof": _good_secret_sweep(),
|
||||||
|
"worktree_proof": {
|
||||||
|
"worktree_path": "/repo/branches/review-feat-issue-224",
|
||||||
|
"porcelain_status": "",
|
||||||
|
"pr_scope_files": ["docs/wiki/Repositories.md"],
|
||||||
|
"scratch_used": True,
|
||||||
|
"scratch_path": "/repo/branches/review-feat-issue-224",
|
||||||
|
},
|
||||||
}
|
}
|
||||||
kwargs.update(overrides)
|
kwargs.update(overrides)
|
||||||
return build_final_report(**kwargs)
|
return build_final_report(**kwargs)
|
||||||
@@ -1450,5 +1559,115 @@ class TestAuthorReporting(unittest.TestCase):
|
|||||||
self.assertFalse(result["complete"])
|
self.assertFalse(result["complete"])
|
||||||
|
|
||||||
|
|
||||||
|
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__":
|
if __name__ == "__main__":
|
||||||
unittest.main()
|
unittest.main()
|
||||||
|
|||||||
@@ -0,0 +1,166 @@
|
|||||||
|
"""Tests for reviewer worktree safety proofs (Issue #233)."""
|
||||||
|
import sys
|
||||||
|
import unittest
|
||||||
|
|
||||||
|
sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
|
||||||
|
|
||||||
|
from reviewer_worktree import ( # noqa: E402
|
||||||
|
assess_author_worktree_continuity,
|
||||||
|
assess_reviewer_git_command_log,
|
||||||
|
assess_reviewer_worktree_proof,
|
||||||
|
files_outside_pr_scope,
|
||||||
|
is_forbidden_reviewer_git_command,
|
||||||
|
is_readonly_reviewer_git_command,
|
||||||
|
parse_dirty_tracked_files,
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
class TestParseDirtyTrackedFiles(unittest.TestCase):
|
||||||
|
def test_ignores_untracked_files(self):
|
||||||
|
porcelain = "?? untracked.txt\n M tracked.py\n"
|
||||||
|
self.assertEqual(parse_dirty_tracked_files(porcelain), ["tracked.py"])
|
||||||
|
|
||||||
|
def test_parses_renamed_paths(self):
|
||||||
|
porcelain = "R old.py -> new.py\n"
|
||||||
|
self.assertEqual(parse_dirty_tracked_files(porcelain), ["new.py"])
|
||||||
|
|
||||||
|
|
||||||
|
class TestFilesOutsidePrScope(unittest.TestCase):
|
||||||
|
def test_all_dirty_in_scope_is_clean(self):
|
||||||
|
self.assertEqual(
|
||||||
|
files_outside_pr_scope(
|
||||||
|
["docs/wiki/Repositories.md"],
|
||||||
|
["docs/wiki/Repositories.md"],
|
||||||
|
),
|
||||||
|
[],
|
||||||
|
)
|
||||||
|
|
||||||
|
def test_unrelated_dirty_files_detected(self):
|
||||||
|
self.assertEqual(
|
||||||
|
files_outside_pr_scope(
|
||||||
|
["review_proofs.py", "docs/wiki/Repositories.md"],
|
||||||
|
["docs/wiki/Repositories.md"],
|
||||||
|
),
|
||||||
|
["review_proofs.py"],
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
class TestForbiddenGitCommands(unittest.TestCase):
|
||||||
|
def test_blocks_stash_operations(self):
|
||||||
|
self.assertTrue(is_forbidden_reviewer_git_command("git stash"))
|
||||||
|
self.assertTrue(
|
||||||
|
is_forbidden_reviewer_git_command(
|
||||||
|
'git stash push -m "reviewer-temp-stash" -- tests/test_mcp_server.py'
|
||||||
|
)
|
||||||
|
)
|
||||||
|
self.assertTrue(is_forbidden_reviewer_git_command("git stash pop"))
|
||||||
|
self.assertTrue(is_forbidden_reviewer_git_command("git stash drop"))
|
||||||
|
|
||||||
|
def test_blocks_checkout_reset_and_clean(self):
|
||||||
|
self.assertTrue(
|
||||||
|
is_forbidden_reviewer_git_command(
|
||||||
|
"git checkout -- review_proofs.py tests/test_mcp_server.py"
|
||||||
|
)
|
||||||
|
)
|
||||||
|
self.assertTrue(is_forbidden_reviewer_git_command("git reset --hard"))
|
||||||
|
self.assertTrue(is_forbidden_reviewer_git_command("git clean -fd"))
|
||||||
|
|
||||||
|
def test_allows_readonly_commands(self):
|
||||||
|
for cmd in (
|
||||||
|
"git fetch prgs master",
|
||||||
|
"git status --porcelain",
|
||||||
|
"git diff prgs/master...HEAD",
|
||||||
|
"git rev-parse HEAD",
|
||||||
|
"git -C /repo log -1",
|
||||||
|
):
|
||||||
|
with self.subTest(cmd=cmd):
|
||||||
|
self.assertFalse(is_forbidden_reviewer_git_command(cmd))
|
||||||
|
self.assertTrue(is_readonly_reviewer_git_command(cmd))
|
||||||
|
|
||||||
|
|
||||||
|
class TestAssessReviewerWorktreeProof(unittest.TestCase):
|
||||||
|
def test_clean_worktree_proceeds(self):
|
||||||
|
result = assess_reviewer_worktree_proof({
|
||||||
|
"worktree_path": "/repo/branches/review-pr-231",
|
||||||
|
"porcelain_status": "",
|
||||||
|
"pr_scope_files": ["docs/wiki/Repositories.md"],
|
||||||
|
"scratch_used": False,
|
||||||
|
"git_commands": ["git fetch prgs master", "git diff prgs/master...HEAD"],
|
||||||
|
})
|
||||||
|
self.assertTrue(result["proven"])
|
||||||
|
self.assertFalse(result["block"])
|
||||||
|
|
||||||
|
def test_dirty_unrelated_without_scratch_blocks(self):
|
||||||
|
result = assess_reviewer_worktree_proof({
|
||||||
|
"worktree_path": "/repo",
|
||||||
|
"dirty_files": ["review_proofs.py", "tests/test_review_proofs.py"],
|
||||||
|
"pr_scope_files": ["docs/wiki/Repositories.md"],
|
||||||
|
"scratch_used": False,
|
||||||
|
})
|
||||||
|
self.assertFalse(result["proven"])
|
||||||
|
self.assertTrue(result["block"])
|
||||||
|
self.assertIn("review_proofs.py", result["unrelated_dirty_files"][0])
|
||||||
|
|
||||||
|
def test_scratch_worktree_allows_dirty_main_repo(self):
|
||||||
|
result = assess_reviewer_worktree_proof({
|
||||||
|
"worktree_path": "/repo",
|
||||||
|
"dirty_files": ["review_proofs.py"],
|
||||||
|
"pr_scope_files": ["docs/wiki/Repositories.md"],
|
||||||
|
"scratch_used": True,
|
||||||
|
"scratch_path": "/repo/branches/review-feat-issue-224",
|
||||||
|
})
|
||||||
|
self.assertTrue(result["proven"])
|
||||||
|
|
||||||
|
def test_forbidden_command_history_blocks(self):
|
||||||
|
result = assess_reviewer_worktree_proof({
|
||||||
|
"worktree_path": "/repo/branches/review-pr-231",
|
||||||
|
"porcelain_status": "",
|
||||||
|
"git_commands": ["git stash push -m temp -- review_proofs.py"],
|
||||||
|
})
|
||||||
|
self.assertFalse(result["proven"])
|
||||||
|
self.assertTrue(result["forbidden_commands"])
|
||||||
|
|
||||||
|
def test_unrelated_mutations_claimed_blocks(self):
|
||||||
|
result = assess_reviewer_worktree_proof({
|
||||||
|
"worktree_path": "/repo",
|
||||||
|
"porcelain_status": "",
|
||||||
|
"unrelated_mutations_claimed": True,
|
||||||
|
})
|
||||||
|
self.assertFalse(result["proven"])
|
||||||
|
|
||||||
|
|
||||||
|
class TestAuthorContinuity(unittest.TestCase):
|
||||||
|
def test_author_may_keep_dirty_worktree(self):
|
||||||
|
result = assess_author_worktree_continuity({
|
||||||
|
"task_role": "author",
|
||||||
|
"dirty_files": ["feat.py"],
|
||||||
|
})
|
||||||
|
self.assertTrue(result["allowed"])
|
||||||
|
|
||||||
|
def test_reviewer_dirty_worktree_uses_reviewer_gate(self):
|
||||||
|
result = assess_author_worktree_continuity({
|
||||||
|
"task_role": "reviewer",
|
||||||
|
"worktree_path": "/repo",
|
||||||
|
"dirty_files": ["other.py"],
|
||||||
|
"pr_scope_files": ["docs/a.md"],
|
||||||
|
"scratch_used": False,
|
||||||
|
})
|
||||||
|
self.assertFalse(result["proven"])
|
||||||
|
|
||||||
|
|
||||||
|
class TestAssessReviewerGitCommandLog(unittest.TestCase):
|
||||||
|
def test_empty_log_is_clean(self):
|
||||||
|
result = assess_reviewer_git_command_log([])
|
||||||
|
self.assertTrue(result["proven"])
|
||||||
|
|
||||||
|
def test_mixed_log_blocks_on_forbidden(self):
|
||||||
|
result = assess_reviewer_git_command_log([
|
||||||
|
"git fetch prgs",
|
||||||
|
"git stash",
|
||||||
|
])
|
||||||
|
self.assertFalse(result["proven"])
|
||||||
|
self.assertEqual(len(result["forbidden_commands"]), 1)
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
unittest.main()
|
||||||
Reference in New Issue
Block a user