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

fix: resolve conflicts for PR #358 against latest master

Browse Source 
Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
This commit is contained in:
sysadmin
2026-07-07 09:27:29 -04:00
co-authored by Claude Opus 4.8
parent af806919b3 b41d9d6006
commit c2226177b6
9 changed files with 1218 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files
review_proofs.py
+328
View File
@@ -1242,6 +1242,102 @@ def assess_workspace_mutation_consistency(report_text, observed_commands=None):
}
# ── Already-landed report wording (Issue #298) ───────────────────────────────
#
# An already-landed PR is reconciliation-only: it must never be called
# eligible, its head SHA must never be called reviewed, and the legacy
# eligible/reviewed/scratch-worktree handoff fields must not appear.
# Reviewed-head claims in any report require validation + diff review to
# have actually passed.
ALREADY_LANDED_FORBIDDEN_PHRASES = (
"oldest eligible pr",
"next eligible pr",
"pinned reviewed head",
"scratch worktree used",
)
ALREADY_LANDED_ELIGIBILITY_CLASS = "ALREADY_LANDED_RECONCILE_REQUIRED"
def assess_already_landed_report_wording(report_text, *, already_landed,
review_validated=False):
"""#298: reject eligible/reviewed wording for already-landed PRs.
*already_landed* states whether the already-landed gate fired for the
selected PR. *review_validated* states whether validation and diff
review actually passed this session; only then may a reviewed head
SHA be populated in a non-already-landed report.
Returns {'complete', 'downgraded', 'reasons'}; fails closed.
"""
text_lower = (report_text or "").lower()
fields = _report_labeled_fields(report_text)
reasons = []
reviewed_head = fields.get("reviewed head sha")
reviewed_head_populated = bool(
reviewed_head and not reviewed_head.strip().lower().startswith("none")
)
pinned_head_present = "pinned reviewed head" in fields
if already_landed:
for phrase in ALREADY_LANDED_FORBIDDEN_PHRASES:
if phrase in text_lower:
reasons.append(
f"already-landed report must not use '{phrase}'; the PR "
"is reconciliation-only, not review/merge eligible"
)
eligibility = fields.get("eligibility class", "")
if ALREADY_LANDED_ELIGIBILITY_CLASS not in eligibility.upper():
reasons.append(
"already-landed report missing 'Eligibility class: "
f"{ALREADY_LANDED_ELIGIBILITY_CLASS}'"
)
for required in ("oldest open pr requiring action",
"candidate head sha"):
if required not in fields:
reasons.append(
f"already-landed report missing '{required}' field"
)
if "reviewed head sha" not in fields:
reasons.append(
"already-landed report must state 'Reviewed head SHA: none'"
)
elif reviewed_head_populated:
reasons.append(
"already-landed report must not claim a reviewed head SHA; "
"no validation or diff review passed for this PR"
)
worktree_used = fields.get("review worktree used", "")
if worktree_used.strip().lower() not in ("false", "no"):
reasons.append(
"already-landed report must state 'Review worktree used: "
"false'"
)
elif not review_validated:
if pinned_head_present:
reasons.append(
"'Pinned reviewed head' populated before validation and "
"diff review passed"
)
if reviewed_head_populated:
reasons.append(
"reviewed head SHA populated before validation and diff "
"review passed"
)
return {
"complete": not reasons,
"downgraded": bool(reasons),
"reasons": reasons,
}
def assess_role_boundary(proof=None, *, task_role=None, namespaces_used=None,
justification=None):
"""Assess reviewer/author role separation for blind queue workflows.
@@ -1422,6 +1518,7 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
report_text=None, review_decision_lock=None,
controller_handoff=None, capability_proof=None,
sweep_proof=None, worktree_proof=None,
validation_environment=None,
queue_ordering=None,
baseline_validation=None, project_root=None):
"""Required behavior 6 + acceptance criteria: one report, distinct proofs.
@@ -1450,6 +1547,22 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
)
empty_queue_report = assess_empty_queue_report(report_text)
if validation_environment is None and report_text:
validation_env_proof = assess_validation_environment_proof(
report_text=report_text,
)
elif validation_environment is not None:
validation_env_proof = assess_validation_environment_proof(
report_text=report_text,
validation_environment=validation_environment,
)
else:
validation_env_proof = {
"proven": True,
"block": False,
"reasons": [],
"violations": [],
}
if queue_ordering is None and report_text:
queue_ordering_proof = assess_queue_ordering_proof(
report_text=report_text,
@@ -1617,6 +1730,11 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
"empty-queue report missing or failed trust-gate proof (#198)"
)
downgrade_reasons.extend(empty_queue_report.get("reasons", []))
if not validation_env_proof.get("proven"):
downgrade_reasons.append(
"validation environment proof missing or failed (#311)"
)
downgrade_reasons.extend(validation_env_proof.get("reasons", []))
if not queue_ordering_proof.get("proven"):
downgrade_reasons.append(
"queue ordering proof missing or failed (#321)"
@@ -1648,11 +1766,13 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
# #179: no merge without a proven final live-state recheck.
and live_state_proven
and worktree_proven
and validation_env_proof.get("proven")
and queue_ordering_proof.get("proven")
and baseline_validation.get("proven")
)
violations = []
violations.extend(validation_env_proof.get("violations", []))
violations.extend(queue_ordering_proof.get("violations", []))
violations.extend(baseline_validation.get("violations", []))
if merge_performed and not merge_allowed:
@@ -1706,6 +1826,12 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
else True
),
"empty_queue_trust_gate_status": empty_queue_report.get("status"),
"validation_environment_proven": bool(
validation_env_proof.get("proven")
),
"validation_environment_violations": list(
validation_env_proof.get("violations") or []
),
"queue_ordering_proven": bool(queue_ordering_proof.get("proven")),
"queue_ordering_policy": queue_ordering_proof.get("policy"),
"queue_ordering_violations": list(
@@ -3322,6 +3448,192 @@ def build_review_mutation_proof(run_log: list[dict]) -> dict:
}
# ---------------------------------------------------------------------------
# Validation environment proof (#311)
# ---------------------------------------------------------------------------
_VALIDATION_CLAIM_RE = re.compile(
r"(?:validation\s+command|pytest\s+(?:executed|passed|failed)|"
r"\d+\s+passed|\d+\s+failed|tests?\s+passed)",
re.IGNORECASE,
)
_VALIDATION_CMD_LINE_RE = re.compile(
r"validation\s+command\s*:\s*(.+)",
re.IGNORECASE,
)
_VALIDATION_CWD_LINE_RE = re.compile(
r"working\s+directory\s*:\s*(\S+)",
re.IGNORECASE,
)
_WHICH_PYTEST_LINE_RE = re.compile(
r"which\s+pytest\s*:\s*(.+)",
re.IGNORECASE,
)
_PYTEST_VERSION_LINE_RE = re.compile(
r"pytest\s+--version\s*:\s*(.+)",
re.IGNORECASE,
)
_BARE_PYTEST_CMD_RE = re.compile(
r"^(?:python\s+-m\s+)?pytest(?:\s|$)",
re.IGNORECASE,
)
_VAGUE_PYTEST_CLAIM_RE = re.compile(
r"pytest\s+(?:executed|passed|failed)\b",
re.IGNORECASE,
)
def _parse_validation_command(report_text: str) -> str:
for line in (report_text or "").splitlines():
match = _VALIDATION_CMD_LINE_RE.search(line)
if match:
return match.group(1).strip().rstrip(".")
return ""
def _parse_validation_cwd(report_text: str) -> str:
for line in (report_text or "").splitlines():
match = _VALIDATION_CWD_LINE_RE.search(line)
if match:
return match.group(1).strip().rstrip(",.;")
return ""
def _is_bare_pytest_command(command: str) -> bool:
text = (command or "").strip()
if not text:
return False
if "/" in text or "\\" in text:
return False
return bool(_BARE_PYTEST_CMD_RE.match(text))
def assess_validation_environment_proof(
*,
report_text: str | None = None,
validation_environment: dict | None = None,
) -> dict:
"""#311: reviewer reports must include exact validation command/environment."""
text = report_text or ""
lower = text.lower()
env = validation_environment or {}
reasons: list[str] = []
violations: list[str] = []
claims_validation = bool(_VALIDATION_CLAIM_RE.search(text))
if not claims_validation and not env.get("command"):
return {
"proven": True,
"block": False,
"claims_validation": False,
"reasons": [],
"violations": [],
"safe_next_action": "proceed",
}
command = (env.get("command") or _parse_validation_command(text)).strip()
working_directory = (
env.get("working_directory") or env.get("cwd")
or _parse_validation_cwd(text)
).strip()
if not command:
reasons.append(
"validation result claimed without exact validation command (#311)"
)
elif _VAGUE_PYTEST_CLAIM_RE.search(text) and command == "":
violations.append(
"vague pytest summary without exact validation command (#311)"
)
if not working_directory:
reasons.append(
"validation result claimed without working directory (#311)"
)
has_result = any(
token in lower
for token in ("passed", "failed", "skipped", "error")
) or any(
env.get(key) is not None
for key in ("passed", "failed", "skipped", "result")
)
if not has_result:
reasons.append(
"validation result claimed without pass/fail result evidence (#311)"
)
if command and _is_bare_pytest_command(command):
which_pytest = (
(env.get("which_pytest") or env.get("executable_path") or "")
.strip()
)
if not which_pytest:
for line in text.splitlines():
match = _WHICH_PYTEST_LINE_RE.search(line)
if match:
which_pytest = match.group(1).strip()
break
if not which_pytest:
reasons.append(
"bare pytest command requires which-pytest/executable proof (#311)"
)
pytest_version = (env.get("pytest_version") or "").strip()
if not pytest_version:
for line in text.splitlines():
match = _PYTEST_VERSION_LINE_RE.search(line)
if match:
pytest_version = match.group(1).strip()
break
if not pytest_version:
reasons.append(
"bare pytest command requires pytest --version proof (#311)"
)
venv_resolved = env.get("venv_resolved")
if venv_resolved is None:
venv_resolved = any(
marker in lower
for marker in (
"project venv",
"venv/bin/pytest",
"resolves to the project venv",
"venv resolved: yes",
)
)
if not venv_resolved and which_pytest and "venv" not in which_pytest:
reasons.append(
"bare pytest command requires proof whether executable "
"resolves to the project venv (#311)"
)
proven = not reasons and not violations
return {
"proven": proven,
"block": bool(violations),
"claims_validation": True,
"command": command or None,
"working_directory": working_directory or None,
"reasons": reasons,
"violations": violations,
"safe_next_action": (
"report exact validation command, working directory, result, and "
"for bare pytest also which-pytest, pytest --version, and venv proof"
if not proven
else "proceed"
),
}
# ---------------------------------------------------------------------------
# Reviewer queue ordering proof (#321)
# ---------------------------------------------------------------------------
@@ -4237,6 +4549,13 @@ def assess_git_ref_mutation_report(report_text, *, command_log=None):
}
def assess_merge_simulation_report(report_text, **kwargs):
"""#317: classify local merge simulation as worktree/index mutations."""
from reviewer_merge_simulation import assess_merge_simulation_report as _assess
return _assess(report_text, **kwargs)
def assess_validation_integrity_report(report_text, **kwargs):
"""#316: separate official PR-head validation from diagnostic experiments."""
from reviewer_validation_integrity import assess_validation_integrity_report as _assess
@@ -4258,6 +4577,15 @@ def assess_non_mergeable_skip_proof(report_text, **kwargs):
return _assess(report_text, **kwargs)
def assess_validation_worktree_edit_report(report_text, **kwargs):
"""#315: block edits in PR validation worktrees; require diagnostic separation."""
from reviewer_validation_worktree_edits import (
assess_validation_worktree_edit_report as _assess,
)
return _assess(report_text, **kwargs)
def assess_worktree_ownership_report(report_text, **kwargs):
"""#312: prove session-owned or safe-reuse worktree before reset/validation."""
from reviewer_worktree_ownership import assess_worktree_ownership_report as _assess
reviewer_merge_simulation.py
+148
View File
@@ -0,0 +1,148 @@
"""Merge-simulation worktree mutation verifier for reviewer reports (#317)."""
from __future__ import annotations
import re
from typing import Any
_WORKTREE_INDEX_FIELD_RE = re.compile(
r"^\s*worktree/index mutations\s*:\s*(.+?)\s*$",
re.IGNORECASE | re.MULTILINE,
)
_READONLY_DIAG_RE = re.compile(
r"read[- ]only diagnostics\s*:\s*(.+)$",
re.IGNORECASE | re.MULTILINE,
)
_WORKTREE_PATH_RE = re.compile(
r"(?:worktree path|diagnostic worktree|simulation worktree)\s*:",
re.IGNORECASE,
)
_PRE_CLEAN_RE = re.compile(
r"(?:pre[- ]simulation|before simulation|clean before).*(?:clean|status)",
re.IGNORECASE,
)
_MERGE_RESULT_RE = re.compile(
r"(?:merge result|simulation result|conflict status|merge conflict)",
re.IGNORECASE,
)
_ABORT_RE = re.compile(
r"(?:merge --abort|abort/reset command|abort command|git merge --abort)",
re.IGNORECASE,
)
_POST_CLEAN_RE = re.compile(
r"(?:post[- ]abort|after abort|clean after).*(?:clean|status)",
re.IGNORECASE,
)
def _command_text(entry: Any) -> str:
if isinstance(entry, dict):
return str(entry.get("command") or "").strip()
return str(entry or "").strip()
def _git_subcommand_line(command: str) -> str | None:
tokens = command.split()
if not tokens or tokens[0] != "git":
return None
return " ".join(tokens[1:])
def merge_simulation_commands(command_log: list | None) -> list[str]:
"""Return logged commands that perform local merge simulation (#317)."""
out: list[str] = []
for entry in command_log or []:
command = _command_text(entry)
rest = _git_subcommand_line(command)
if rest is None:
continue
lower = command.lower()
if rest.startswith("merge"):
if "--no-commit" in lower or (
"--no-ff" in lower and "merge" in lower and "--commit" not in lower
):
out.append(command)
elif "--abort" in lower:
out.append(command)
return out
def assess_merge_simulation_report(
report_text: str,
*,
command_log: list | None = None,
) -> dict[str, Any]:
"""Reject merge simulation classified as read-only; require worktree proof (#317)."""
text = report_text or ""
lower = text.lower()
reasons: list[str] = []
simulation_commands = merge_simulation_commands(command_log)
has_abort = any("--abort" in c.lower() for c in simulation_commands)
has_simulation = any(
"--no-commit" in c.lower() or (
"--no-ff" in c.lower() and "--abort" not in c.lower()
)
for c in simulation_commands
)
if not simulation_commands:
return {
"proven": True,
"block": False,
"reasons": [],
"simulation_commands": [],
"safe_next_action": "proceed",
}
field = _WORKTREE_INDEX_FIELD_RE.search(text)
value = (field.group(1).strip().lower() if field else "") or ""
if not value or value in {"none", "n/a", "no"}:
reasons.append(
"merge simulation commands ran but report lacks "
"'Worktree/index mutations' entry"
)
elif "merge" not in value and "simulation" not in value:
if not any(cmd.lower() in lower for cmd in simulation_commands):
reasons.append(
"Worktree/index mutations must document merge simulation commands"
)
readonly = _READONLY_DIAG_RE.search(text)
if readonly:
readonly_value = readonly.group(1)
for command in simulation_commands:
if command.lower() in readonly_value.lower():
reasons.append(
"merge simulation may not be listed under read-only diagnostics"
)
if has_simulation and "merge simulation" in readonly_value.lower():
reasons.append(
"merge simulation may not be classified as read-only diagnostics"
)
if has_simulation:
if not _WORKTREE_PATH_RE.search(text):
reasons.append("merge simulation report missing worktree path")
if not _PRE_CLEAN_RE.search(text):
reasons.append("merge simulation report missing pre-simulation clean status")
if not _MERGE_RESULT_RE.search(text):
reasons.append("merge simulation report missing merge result/conflict status")
if has_abort or has_simulation:
if has_abort and not _ABORT_RE.search(text):
reasons.append("merge simulation report missing abort/reset command proof")
if has_abort and not _POST_CLEAN_RE.search(text):
reasons.append("merge simulation report missing post-abort clean status")
proven = not reasons
return {
"proven": proven,
"block": not proven,
"reasons": reasons,
"simulation_commands": simulation_commands,
"safe_next_action": (
"classify merge simulation under Worktree/index mutations with full "
"pre/merge/abort/post-clean proof; never list as read-only diagnostics"
if reasons
else "proceed"
),
}
reviewer_validation_worktree_edits.py
+184
View File
@@ -0,0 +1,184 @@
"""PR validation worktree read-only edit verifier (#315)."""
from __future__ import annotations
import re
from typing import Any
_VALIDATION_WORKTREE_RE = re.compile(
r"branches/(?:review-pr\d+[\w/-]*|review-[\w-]+)",
re.IGNORECASE,
)
_DIAGNOSTIC_WORKTREE_RE = re.compile(
r"branches/(?:diagnostic[\w/-]*|scratch[\w/-]*)",
re.IGNORECASE,
)
_FILE_EDITS_NONE_RE = re.compile(
r"file edits by reviewer\s*:\s*none\b",
re.IGNORECASE,
)
_FILE_EDITS_FIELD_RE = re.compile(
r"file edits by reviewer\s*:\s*(.+)",
re.IGNORECASE,
)
_VALIDATION_WORKTREE_PATH_RE = re.compile(
r"(?:validation worktree path|review worktree path)\s*:\s*(\S+)",
re.IGNORECASE,
)
_DIAGNOSTIC_WORKTREE_PATH_RE = re.compile(
r"(?:diagnostic (?:scratch )?worktree path|diagnostic worktree)\s*:\s*(\S+)",
re.IGNORECASE,
)
_DIAGNOSTIC_LABEL_RE = re.compile(
r"diagnostic local experiment|not pr-head validation|diagnostic-only",
re.IGNORECASE,
)
_OFFICIAL_VALIDATION_RE = re.compile(
r"(?:official (?:pr-head )?validation|pr-head validation result)",
re.IGNORECASE,
)
_POST_EDIT_OFFICIAL_RE = re.compile(
r"(?:official validation(?:\s+result)?\s*:\s*pass|validation passed after (?:local )?edit)",
re.IGNORECASE,
)
_PERFORMED_EDIT_KINDS = frozenset({"file_edit", "edit", "write", "edited", "created", "wrote"})
def _normalize_path(path: str) -> str:
return (path or "").replace("\\", "/").strip()
def _is_validation_worktree(path: str) -> bool:
return bool(_VALIDATION_WORKTREE_RE.search(_normalize_path(path)))
def _is_diagnostic_worktree(path: str) -> bool:
normalized = _normalize_path(path)
return bool(
_DIAGNOSTIC_WORKTREE_RE.search(normalized)
or "diagnostic" in normalized.lower()
)
def _performed_edits(action_log: list[dict] | None) -> list[dict[str, str]]:
edits: list[dict[str, str]] = []
for entry in action_log or []:
if entry.get("gated_rejected") or entry.get("performed") is False:
continue
kind = (entry.get("kind") or entry.get("action") or "file_edit").strip().lower()
if kind not in _PERFORMED_EDIT_KINDS:
continue
path = (entry.get("path") or "").strip()
if not path:
continue
worktree = _normalize_path(str(entry.get("worktree_path") or entry.get("cwd") or ""))
edits.append({"path": path, "worktree_path": worktree})
return edits
def _extract_validation_path(text: str, session: dict) -> str:
path = _normalize_path(str(session.get("validation_worktree_path") or ""))
if path:
return path
match = _VALIDATION_WORKTREE_PATH_RE.search(text or "")
return _normalize_path(match.group(1)) if match else ""
def _extract_diagnostic_path(text: str, session: dict) -> str:
path = _normalize_path(str(session.get("diagnostic_worktree_path") or ""))
if path:
return path
match = _DIAGNOSTIC_WORKTREE_PATH_RE.search(text or "")
return _normalize_path(match.group(1)) if match else ""
def assess_validation_worktree_edit_report(
report_text: str,
*,
validation_session: dict | None = None,
action_log: list[dict] | None = None,
) -> dict[str, Any]:
"""Block edits in PR validation worktrees; require diagnostic separation (#315)."""
text = report_text or ""
session = dict(validation_session or {})
reasons: list[str] = []
validation_path = _extract_validation_path(text, session)
diagnostic_path = _extract_diagnostic_path(text, session)
validation_edits = list(session.get("validation_worktree_edits") or [])
diagnostic_edits = list(session.get("diagnostic_edits") or [])
edits = _performed_edits(action_log)
if not validation_edits:
for entry in edits:
wt = entry.get("worktree_path") or validation_path
if wt and _is_validation_worktree(wt) and not _is_diagnostic_worktree(wt):
validation_edits.append(entry["path"])
if not diagnostic_edits:
for entry in edits:
wt = entry.get("worktree_path") or ""
if wt and _is_diagnostic_worktree(wt):
diagnostic_edits.append(entry["path"])
elif entry["path"] and diagnostic_path and not validation_edits:
if _is_diagnostic_worktree(diagnostic_path):
diagnostic_edits.append(entry["path"])
claimed_none = bool(_FILE_EDITS_NONE_RE.search(text))
any_edits = bool(validation_edits or diagnostic_edits or edits)
if validation_edits:
reasons.append(
"reviewer must not edit files in the PR validation worktree; "
f"observed edits: {', '.join(validation_edits)}"
)
if diagnostic_edits or session.get("diagnostic_experiment"):
if not diagnostic_path and not _DIAGNOSTIC_WORKTREE_PATH_RE.search(text):
reasons.append(
"diagnostic experiments require a separate diagnostic scratch worktree path"
)
if not _DIAGNOSTIC_LABEL_RE.search(text):
reasons.append(
"diagnostic experiments must be labeled "
"'Diagnostic local experiment — not PR-head validation'"
)
if not _FILE_EDITS_FIELD_RE.search(text) or claimed_none:
reasons.append(
"diagnostic edits must be reported under 'File edits by reviewer'"
)
if any_edits and claimed_none:
reasons.append(
"report claims 'File edits by reviewer: none' but reviewer file edits occurred"
)
if session.get("official_validation_after_edit") or _POST_EDIT_OFFICIAL_RE.search(text):
if validation_edits or (validation_path and diagnostic_edits):
reasons.append(
"post-edit test results cannot be reported as official PR-head validation"
)
if validation_edits and _OFFICIAL_VALIDATION_RE.search(text):
if not _DIAGNOSTIC_LABEL_RE.search(text):
reasons.append(
"validation worktree was edited; official PR-head validation is no longer valid"
)
proven = not reasons
return {
"proven": proven,
"block": not proven,
"reasons": reasons,
"validation_worktree_path": validation_path or None,
"diagnostic_worktree_path": diagnostic_path or None,
"validation_worktree_edits": validation_edits,
"diagnostic_edits": diagnostic_edits,
"safe_next_action": (
"keep PR validation worktrees read-only; use a separate diagnostic "
"scratch worktree and report all reviewer file edits"
if reasons
else "proceed"
),
}
tests/test_agent_temp_artifacts.py
+20
View File
@@ -62,7 +62,24 @@ class TestPreflightWarnings(unittest.TestCase):
self.assertIn("_encode_commit_payload.py", status["preflight_warnings"][0])
# Issue-write tools are profile-gated (#69); gitea_lock_issue requires
# gitea.issue.comment (see task_capability_map), so the gate must be
# seeded exactly like tests/test_mcp_server.py::TestIssueLocking (#359).
ISSUE_WRITE_ENV = {
"GITEA_ALLOWED_OPERATIONS": (
"gitea.issue.create,gitea.issue.close,gitea.issue.comment"
),
}
class TestIssueLockArtifactWarning(unittest.TestCase):
def setUp(self):
self._env_patcher = patch.dict(os.environ, ISSUE_WRITE_ENV, clear=True)
self._env_patcher.start()
def tearDown(self):
self._env_patcher.stop()
@patch("mcp_server.api_get_all", return_value=[])
@patch("mcp_server._auth", return_value="token x")
@patch("mcp_server._resolve", return_value=("h", "o", "r"))
@@ -72,6 +89,9 @@ class TestIssueLockArtifactWarning(unittest.TestCase):
mock_state.return_value = {
"current_branch": "master",
"porcelain_status": "?? _emit_payload.py\n",
"base_equivalent": True,
"inspected_git_root": "/scratch/wt",
"base_branch": "origin/master",
}
result = mcp_server.gitea_lock_issue(
issue_number=261,
tests/test_already_landed_report_wording.py
+151
View File
@@ -0,0 +1,151 @@
"""Tests for already-landed final-report wording validator (#298).
An already-landed PR is reconciliation-only, never review/merge
eligible, and a head SHA may not be called reviewed unless validation
and diff review actually passed. Final reports and controller handoffs
must use the reconciliation wording, not the legacy eligible/reviewed
fields.
"""
import sys
import unittest
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from review_proofs import assess_already_landed_report_wording # noqa: E402
GOOD_ALREADY_LANDED = (
"Controller Handoff\n"
"- Oldest open PR requiring action: PR #278\n"
"- Eligibility class: ALREADY_LANDED_RECONCILE_REQUIRED\n"
"- Candidate head SHA: 2a544b7\n"
"- Reviewed head SHA: none\n"
"- Review worktree used: false\n"
)
class TestAlreadyLandedWording(unittest.TestCase):
def test_correct_reconciliation_wording_passes(self):
result = assess_already_landed_report_wording(
GOOD_ALREADY_LANDED, already_landed=True
)
self.assertTrue(result["complete"])
self.assertFalse(result["downgraded"])
self.assertEqual(result["reasons"], [])
def test_oldest_eligible_wording_fails(self):
report = GOOD_ALREADY_LANDED + "- oldest eligible PR: PR #278\n"
result = assess_already_landed_report_wording(
report, already_landed=True
)
self.assertFalse(result["complete"])
self.assertTrue(
any("oldest eligible pr" in r.lower() for r in result["reasons"])
)
def test_next_eligible_wording_fails(self):
report = GOOD_ALREADY_LANDED + "- next eligible PR: PR #278\n"
result = assess_already_landed_report_wording(
report, already_landed=True
)
self.assertFalse(result["complete"])
def test_pinned_reviewed_head_fails(self):
report = GOOD_ALREADY_LANDED + "- Pinned reviewed head: 2a544b7\n"
result = assess_already_landed_report_wording(
report, already_landed=True
)
self.assertFalse(result["complete"])
self.assertTrue(
any("pinned reviewed head" in r.lower() for r in result["reasons"])
)
def test_scratch_worktree_field_fails(self):
report = GOOD_ALREADY_LANDED + "- Scratch worktree used: False\n"
result = assess_already_landed_report_wording(
report, already_landed=True
)
self.assertFalse(result["complete"])
def test_missing_eligibility_class_fails(self):
report = (
"Controller Handoff\n"
"- Oldest open PR requiring action: PR #278\n"
"- Candidate head SHA: 2a544b7\n"
"- Reviewed head SHA: none\n"
"- Review worktree used: false\n"
)
result = assess_already_landed_report_wording(
report, already_landed=True
)
self.assertFalse(result["complete"])
self.assertTrue(
any("eligibility class" in r.lower() for r in result["reasons"])
)
def test_populated_reviewed_head_sha_fails(self):
report = GOOD_ALREADY_LANDED.replace(
"- Reviewed head SHA: none\n",
"- Reviewed head SHA: 2a544b7\n",
)
result = assess_already_landed_report_wording(
report, already_landed=True
)
self.assertFalse(result["complete"])
self.assertTrue(
any("reviewed head" in r.lower() for r in result["reasons"])
)
class TestNonAlreadyLandedReports(unittest.TestCase):
def test_normal_reviewed_report_passes(self):
report = (
"- Selected PR: 281\n"
"- Pinned reviewed head: abc1234\n"
"- Review decision: approve\n"
)
result = assess_already_landed_report_wording(
report, already_landed=False, review_validated=True
)
self.assertTrue(result["complete"])
def test_blocked_infra_report_with_pinned_head_fails(self):
report = (
"- Selected PR: 281\n"
"- Pinned reviewed head: abc1234\n"
"- Current status: blocked on infra_stop\n"
)
result = assess_already_landed_report_wording(
report, already_landed=False, review_validated=False
)
self.assertFalse(result["complete"])
self.assertTrue(
any("before validation" in r.lower() for r in result["reasons"])
)
def test_validation_failed_report_with_reviewed_sha_fails(self):
report = (
"- Selected PR: 281\n"
"- Reviewed head SHA: abc1234\n"
"- Validation: full suite failed\n"
)
result = assess_already_landed_report_wording(
report, already_landed=False, review_validated=False
)
self.assertFalse(result["complete"])
def test_validation_failed_report_with_reviewed_none_passes(self):
report = (
"- Selected PR: 281\n"
"- Reviewed head SHA: none\n"
"- Validation: full suite failed\n"
)
result = assess_already_landed_report_wording(
report, already_landed=False, review_validated=False
)
self.assertTrue(result["complete"])
if __name__ == "__main__":
unittest.main()
tests/test_llm_workflow_split.py
+13 -1
View File
@@ -136,6 +136,12 @@ def test_create_issue_final_report_verifier_exported():
assert callable(assess_create_issue_final_report)
def test_merge_simulation_verifier_exported():
from review_proofs import assess_merge_simulation_report
assert callable(assess_merge_simulation_report)
def test_validation_integrity_verifier_exported():
from review_proofs import assess_validation_integrity_report
@@ -154,7 +160,13 @@ def test_non_mergeable_skip_verifier_exported():
assert callable(assess_non_mergeable_skip_proof)
def test_validation_worktree_edit_verifier_exported():
from review_proofs import assess_validation_worktree_edit_report
assert callable(assess_validation_worktree_edit_report)
def test_worktree_ownership_verifier_exported():
from review_proofs import assess_worktree_ownership_report
assert callable(assess_worktree_ownership_report)
assert callable(assess_worktree_ownership_report)
tests/test_review_proofs.py
+105 -3
View File
@@ -24,6 +24,7 @@ from review_proofs import ( # noqa: E402
ISSUE_SELECTION_CONTINUATION_EXPLICIT,
ISSUE_SELECTION_REPRESENTED_BY_OPEN_PR,
assess_author_pr_report,
assess_validation_environment_proof,
assess_full_suite_failure_approval_gate,
assess_queue_ordering_proof,
assess_reviewer_baseline_validation_proof,
@@ -2315,10 +2316,108 @@ class TestCreateIssueFinalReport(unittest.TestCase):
self.assertTrue(result["downgraded"])
class TestQueueOrderingProof(unittest.TestCase):
"""Issue #321: reviewer queue selection must prove ordering."""
class TestValidationEnvironmentProof(unittest.TestCase):
"""Issue #311: exact validation command and execution environment."""
ROOT = "/repo/Gitea-Tools/branches/review-pr-280"
def test_venv_pytest_with_full_report_passes(self):
result = assess_validation_environment_proof(
report_text=(
"Validation command: venv/bin/pytest tests/ --ignore=branches\n"
f"Working directory: {self.ROOT}\n"
"Result: 1014 passed, 6 skipped"
),
)
self.assertTrue(result["proven"])
def test_bare_pytest_without_path_proof_fails(self):
result = assess_validation_environment_proof(
report_text=(
"Validation command: pytest\n"
f"Working directory: {self.ROOT}\n"
"1014 passed, 6 skipped"
),
)
self.assertFalse(result["proven"])
def test_bare_pytest_with_path_and_version_proof_passes(self):
result = assess_validation_environment_proof(
report_text=(
"Validation command: pytest tests/\n"
f"Working directory: {self.ROOT}\n"
"which pytest: /repo/Gitea-Tools/venv/bin/pytest\n"
"pytest --version: pytest 8.3.4\n"
"Resolves to the project venv: yes\n"
"1014 passed, 6 skipped"
),
)
self.assertTrue(result["proven"])
def test_vague_pytest_summary_without_command_fails(self):
result = assess_validation_environment_proof(
report_text=(
f"Working directory: {self.ROOT}\n"
"pytest executed inside the worktree: 1014 passed, 6 skipped"
),
)
self.assertFalse(result["proven"])
def test_structured_environment_proof_passes(self):
result = assess_validation_environment_proof(
validation_environment={
"command": "pytest tests/",
"working_directory": self.ROOT,
"which_pytest": "/repo/Gitea-Tools/venv/bin/pytest",
"pytest_version": "pytest 8.3.4",
"venv_resolved": True,
"passed": 1014,
"skipped": 6,
},
)
self.assertTrue(result["proven"])
def test_missing_working_directory_fails(self):
result = assess_validation_environment_proof(
report_text=(
"Validation command: venv/bin/pytest tests/\n"
"1014 passed, 6 skipped"
),
)
self.assertFalse(result["proven"])
def test_build_final_report_downgrades_missing_environment_proof(self):
report = build_final_report(
checkout_proof=_good_checkout(),
inventory=_good_inventory(),
validation=_good_validation(),
contamination=_good_contamination(),
identity_eligible=True,
merge_performed=False,
issue_status_verified=True,
capability_evidence=_good_capability_evidence(),
sweep=_good_sweep(),
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(),
worktree_proof={
"worktree_path": "/repo/branches/review-feat-issue-224",
"porcelain_status": "",
"pr_scope_files": ["docs/wiki/Repositories.md"],
"scratch_used": True,
},
report_text=(
f"Working directory: {self.ROOT}\n"
"pytest executed: 1014 passed, 6 skipped"
),
)
self.assertNotEqual(report["grade"], "A")
self.assertFalse(report["validation_environment_proven"])
def test_next_eligible_claim_without_policy_fails(self):
result = assess_queue_ordering_proof(
report_text="Selected the next eligible PR: PR #286.",
)
@@ -2707,3 +2806,6 @@ class TestFullSuiteFailureApprovalGate(unittest.TestCase):
if __name__ == "__main__":
unittest.main()
if __name__ == "__main__":
unittest.main()
tests/test_reviewer_merge_simulation.py
+135
View File
@@ -0,0 +1,135 @@
"""Tests for merge-simulation worktree mutation verifier (#317)."""
import sys
import unittest
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from reviewer_merge_simulation import ( # noqa: E402
assess_merge_simulation_report,
merge_simulation_commands,
)
def _good_simulation_report(**overrides) -> str:
lines = [
"Worktree/index mutations: merge simulation in branches/review-pr281",
"Worktree path: branches/review-pr281",
"Pre-simulation clean status: clean (git status --porcelain empty)",
"Merge result: conflicts in review_proofs.py",
"Abort command: git merge --abort",
"Post-abort clean status: clean after abort (git status --porcelain empty)",
]
text = "\n".join(lines)
for key, value in overrides.items():
text = text.replace(f"{key}:", f"{key}: {value}")
return text
class TestMergeSimulationDetection(unittest.TestCase):
def test_detects_no_commit_merge(self):
commands = merge_simulation_commands([
"git merge prgs/master --no-commit --no-ff",
"git status",
])
self.assertEqual(commands, ["git merge prgs/master --no-commit --no-ff"])
def test_detects_merge_abort(self):
commands = merge_simulation_commands([
"git merge prgs/master --no-commit --no-ff",
"git merge --abort",
])
self.assertIn("git merge --abort", commands)
class TestMergeSimulationReport(unittest.TestCase):
def test_no_simulation_passes(self):
result = assess_merge_simulation_report(
"Review complete. Worktree/index mutations: none",
command_log=["git status --porcelain", "git diff prgs/master...HEAD"],
)
self.assertTrue(result["proven"])
def test_documented_simulation_passes(self):
report = _good_simulation_report()
result = assess_merge_simulation_report(
report,
command_log=[
"git merge prgs/master --no-commit --no-ff",
"git merge --abort",
],
)
self.assertTrue(result["proven"], result["reasons"])
def test_readonly_classification_blocks(self):
report = "\n".join([
"Read-only diagnostics: git merge prgs/master --no-commit --no-ff, git status",
"Worktree/index mutations: none",
])
result = assess_merge_simulation_report(
report,
command_log=["git merge prgs/master --no-commit --no-ff"],
)
self.assertFalse(result["proven"])
self.assertTrue(result["block"])
self.assertTrue(any("read-only" in r.lower() for r in result["reasons"]))
def test_missing_worktree_field_blocks(self):
report = "Merge simulation ran but only noted in prose."
result = assess_merge_simulation_report(
report,
command_log=["git merge prgs/master --no-commit --no-ff"],
)
self.assertFalse(result["proven"])
self.assertTrue(any("worktree/index mutations" in r.lower() for r in result["reasons"]))
def test_conflict_simulation_requires_merge_result(self):
report = _good_simulation_report().replace(
"Merge result: conflicts in review_proofs.py",
"",
)
result = assess_merge_simulation_report(
report,
command_log=["git merge prgs/master --no-commit --no-ff"],
)
self.assertFalse(result["proven"])
self.assertTrue(any("merge result" in r.lower() for r in result["reasons"]))
def test_aborted_simulation_requires_post_clean_proof(self):
report = _good_simulation_report().replace(
"Post-abort clean status: clean after abort (git status --porcelain empty)",
"",
)
result = assess_merge_simulation_report(
report,
command_log=[
"git merge prgs/master --no-commit --no-ff",
"git merge --abort",
],
)
self.assertFalse(result["proven"])
self.assertTrue(any("post-abort" in r.lower() for r in result["reasons"]))
def test_successful_simulation_without_abort_omits_abort_fields(self):
report = "\n".join([
"Worktree/index mutations: merge simulation completed cleanly",
"Worktree path: branches/review-pr281",
"Pre-simulation clean status: clean before simulation",
"Merge result: clean fast-forward simulation",
])
result = assess_merge_simulation_report(
report,
command_log=["git merge prgs/master --no-commit --no-ff"],
)
self.assertTrue(result["proven"], result["reasons"])
class TestExport(unittest.TestCase):
def test_review_proofs_reexport(self):
from review_proofs import assess_merge_simulation_report as exported
self.assertTrue(callable(exported))
if __name__ == "__main__":
unittest.main()
tests/test_reviewer_validation_worktree_edits.py
+134
View File
@@ -0,0 +1,134 @@
"""Tests for PR validation worktree no-edit verifier (#315)."""
import sys
import unittest
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from reviewer_validation_worktree_edits import ( # noqa: E402
assess_validation_worktree_edit_report,
)
def _read_only_report() -> str:
return "\n".join([
"Validation worktree path: branches/review-pr280-feat-issue-275-claim",
"Official PR-head validation on unmodified worktree.",
"Official validation result: failed (1 failed)",
"File edits by reviewer: none",
"Review decision: request_changes",
])
def _diagnostic_report() -> str:
return "\n".join([
"Validation worktree path: branches/review-pr280-feat-issue-275-claim",
"Official PR-head validation on unmodified worktree.",
"Official validation result: failed (1 failed)",
"Review decision: request_changes",
"Diagnostic local experiment — not PR-head validation",
"Diagnostic scratch worktree path: branches/diagnostic-pr280-fix-test",
"File edits by reviewer: tests/test_agent_temp_artifacts.py (diagnostic only)",
"Diagnostic result: passed after local fix",
])
class TestValidationWorktreeEdits(unittest.TestCase):
def test_read_only_review_passes(self):
result = assess_validation_worktree_edit_report(
_read_only_report(),
validation_session={
"validation_worktree_path": (
"branches/review-pr280-feat-issue-275-claim"
),
},
)
self.assertTrue(result["proven"], result["reasons"])
def test_validation_worktree_edit_blocks(self):
result = assess_validation_worktree_edit_report(
_read_only_report(),
validation_session={
"validation_worktree_path": (
"branches/review-pr280-feat-issue-275-claim"
),
"validation_worktree_edits": ["tests/test_agent_temp_artifacts.py"],
},
)
self.assertFalse(result["proven"])
self.assertTrue(result["block"])
self.assertTrue(any("validation worktree" in r.lower() for r in result["reasons"]))
def test_file_edits_none_contradiction_blocks(self):
result = assess_validation_worktree_edit_report(
_read_only_report(),
validation_session={
"validation_worktree_path": (
"branches/review-pr280-feat-issue-275-claim"
),
},
action_log=[
{
"path": "tests/test_agent_temp_artifacts.py",
"kind": "file_edit",
"performed": True,
"worktree_path": "branches/review-pr280-feat-issue-275-claim",
}
],
)
self.assertFalse(result["proven"])
self.assertTrue(any("file edits" in r.lower() for r in result["reasons"]))
def test_diagnostic_scratch_with_reporting_passes(self):
result = assess_validation_worktree_edit_report(
_diagnostic_report(),
validation_session={
"validation_worktree_path": (
"branches/review-pr280-feat-issue-275-claim"
),
"diagnostic_worktree_path": "branches/diagnostic-pr280-fix-test",
"diagnostic_edits": ["tests/test_agent_temp_artifacts.py"],
"diagnostic_experiment": True,
},
)
self.assertTrue(result["proven"], result["reasons"])
def test_diagnostic_without_scratch_worktree_blocks(self):
result = assess_validation_worktree_edit_report(
"File edits by reviewer: tests/test_example.py",
validation_session={
"diagnostic_edits": ["tests/test_example.py"],
"diagnostic_experiment": True,
},
)
self.assertFalse(result["proven"])
self.assertTrue(any("diagnostic" in r.lower() for r in result["reasons"]))
def test_post_edit_official_validation_blocks(self):
report = "\n".join([
"Validation worktree path: branches/review-pr280-feat-issue-275-claim",
"Official validation result: passed after local edit",
"File edits by reviewer: tests/test_agent_temp_artifacts.py",
])
result = assess_validation_worktree_edit_report(
report,
validation_session={
"validation_worktree_path": (
"branches/review-pr280-feat-issue-275-claim"
),
"validation_worktree_edits": ["tests/test_agent_temp_artifacts.py"],
"official_validation_after_edit": True,
},
)
self.assertFalse(result["proven"])
class TestExport(unittest.TestCase):
def test_review_proofs_reexport(self):
from review_proofs import assess_validation_worktree_edit_report as exported
self.assertTrue(callable(exported))
if __name__ == "__main__":
unittest.main()