diff --git a/final_report_validator.py b/final_report_validator.py index b3ab6f5..64f2c40 100644 --- a/final_report_validator.py +++ b/final_report_validator.py @@ -118,6 +118,22 @@ _TARGET_BRANCH_SHA_RE = re.compile( r"target branch sha\s*:\s*[0-9a-f]{40}", re.IGNORECASE, ) +_WORKFLOW_LOAD_HELPER_RE = re.compile( + r"workflow[- ]load helper result\s*:", + re.IGNORECASE, +) +_WORKFLOW_LOAD_HASH_RE = re.compile( + r"workflow[- ]load helper result[\s\S]{0,400}?workflow[_ ]hash\s*:\s*[0-9a-f]{12}", + re.IGNORECASE, +) +_WORKFLOW_LOAD_BOUNDARY_RE = re.compile( + r"workflow[- ]load helper result[\s\S]{0,400}?boundary[_ ]status\s*:\s*(?:clean|violation)", + re.IGNORECASE, +) +_WORKFLOW_FILE_VIEW_NARRATIVE_RE = re.compile( + r"(?:read|viewed|loaded)\s+(?:the\s+)?(?:canonical\s+)?(?:workflow|review-merge-pr\.md)", + re.IGNORECASE, +) _FULL_SHA_RE = re.compile(r"\b[0-9a-f]{40}\b", re.IGNORECASE) _RECONCILE_STALE_FIELDS = ( "pr number opened", @@ -1024,6 +1040,54 @@ def _rule_shared_author_reviewer_same_run(report_text: str) -> list[dict[str, st ) +def _rule_reviewer_workflow_load_boundary(report_text: str) -> list[dict[str, str]]: + """#403: require structured workflow-load helper result, not file-view narrative.""" + if not report_text.strip(): + return [] + findings: list[dict[str, str]] = [] + has_helper = bool(_WORKFLOW_LOAD_HELPER_RE.search(report_text)) + has_hash = bool(_WORKFLOW_LOAD_HASH_RE.search(report_text)) + has_boundary = bool(_WORKFLOW_LOAD_BOUNDARY_RE.search(report_text)) + has_narrative_only = bool(_WORKFLOW_FILE_VIEW_NARRATIVE_RE.search(report_text)) + + if has_narrative_only and not has_helper: + findings.append(validator_finding( + "reviewer.workflow_load_boundary", + "block", + "Workflow-load helper result", + ( + "canonical workflow file-view narrative without structured " + "gitea_load_review_workflow helper result" + ), + ( + "include Workflow-load helper result with workflow_hash and " + "boundary_status from gitea_load_review_workflow" + ), + )) + return findings + + if has_helper and (not has_hash or not has_boundary): + missing = [] + if not has_hash: + missing.append("workflow_hash") + if not has_boundary: + missing.append("boundary_status") + findings.append(validator_finding( + "reviewer.workflow_load_boundary", + "block", + "Workflow-load helper result", + ( + "workflow-load helper result incomplete; missing " + + ", ".join(missing) + ), + ( + "copy workflow_load_helper_result fields from " + "gitea_load_review_workflow into the final report" + ), + )) + return findings + + def _rule_audit_reconciliation_boundary(report_text: str) -> list[dict[str, str]]: from audit_reconciliation_mode import assess_audit_reconciliation_report @@ -1100,6 +1164,7 @@ _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = { _rule_reviewer_already_landed_eligible, _rule_reviewer_already_landed_state, _rule_reviewer_target_branch_freshness, + _rule_reviewer_workflow_load_boundary, _rule_reviewer_mutation_ledger, _rule_reviewer_review_mutation, _rule_reviewer_post_merge_cleanup_proof, diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index db3220a..ae4135d 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -736,6 +736,7 @@ import role_session_router # noqa: E402 import role_namespace_gate # noqa: E402 import task_capability_map # noqa: E402 import review_proofs # noqa: E402 +import review_workflow_boundary # noqa: E402 import review_workflow_load # noqa: E402 import agent_temp_artifacts import issue_lock_worktree # noqa: E402 @@ -7032,14 +7033,42 @@ def gitea_get_runtime_context( return result +@mcp.tool() +def gitea_record_pre_review_command( + command: str, + cwd: str | None = None, + classification: str | None = None, +) -> dict: + """Classify and record a command executed before workflow load (#403). + + Read-only with respect to Gitea API. Pre-review inventory/diagnostic commands + may be recorded as allowed; boundary violations block reviewer mutations. + """ + recorded = review_workflow_boundary.record_pre_review_command( + command, + cwd=cwd, + project_root=PROJECT_ROOT, + classification=classification, + ) + boundary_state = review_workflow_boundary.assess_boundary_status(PROJECT_ROOT) + return { + "success": True, + "recorded": recorded, + "boundary_status": boundary_state.get("boundary_status"), + "boundary_clean": boundary_state.get("boundary_clean"), + "reasons": list(boundary_state.get("reasons") or []), + } + + @mcp.tool() def gitea_load_review_workflow( prompt_text: str | None = None, ) -> dict: - """Load and record canonical review-merge workflow proof for this session (#389). + """Load and record canonical review-merge workflow proof for this session (#389, #403). Read-only with respect to Gitea API; records in-process workflow source/hash - proof required before reviewer review or merge mutations. + proof and session boundary state required before reviewer review or merge + mutations. """ try: recorded = review_workflow_load.record_review_workflow_load( @@ -7051,8 +7080,11 @@ def gitea_load_review_workflow( "reasons": [str(exc)], "recovery_handoff": review_workflow_load.recovery_handoff_without_replay(), } + boundary_reasons = review_workflow_boundary.boundary_blockers(PROJECT_ROOT) + helper = review_workflow_boundary.workflow_load_helper_result( + recorded, PROJECT_ROOT) return { - "success": True, + "success": not boundary_reasons, "loaded": True, "workflow_source": recorded["workflow_source"], "task_mode": recorded["task_mode"], @@ -7064,7 +7096,14 @@ def gitea_load_review_workflow( "prompt_conflicts_with_workflow"], "prompt_conflict_reasons": recorded.get("prompt_conflict_reasons") or [], "workflow_load_proof_present": True, - "reasons": [], + "boundary_status": recorded.get("boundary_status"), + "boundary_clean": recorded.get("boundary_clean"), + "workflow_load_helper_result": helper, + "reasons": boundary_reasons, + "recovery_handoff": ( + review_workflow_load.recovery_handoff_without_replay() + if boundary_reasons else [] + ), } diff --git a/review_workflow_boundary.py b/review_workflow_boundary.py new file mode 100644 index 0000000..f96aaab --- /dev/null +++ b/review_workflow_boundary.py @@ -0,0 +1,259 @@ +"""Reviewer session boundary tracking for workflow-load gate (#403). + +Pre-review commands executed before ``gitea_load_review_workflow`` must be +classified. Boundary violations block downstream reviewer mutations even when +workflow hash proof is present. +""" + +from __future__ import annotations + +import os +import re +from typing import Any + +CLASSIFICATION_READ_ONLY_INVENTORY = "read_only_inventory" +CLASSIFICATION_DIAGNOSTIC = "diagnostic" +CLASSIFICATION_BOUNDARY_VIOLATION = "boundary_violation" +CLASSIFICATION_UNCLASSIFIED = "unclassified" + +ALLOWED_CLASSIFICATIONS = frozenset({ + CLASSIFICATION_READ_ONLY_INVENTORY, + CLASSIFICATION_DIAGNOSTIC, + CLASSIFICATION_BOUNDARY_VIOLATION, + CLASSIFICATION_UNCLASSIFIED, +}) + +_PRE_REVIEW_COMMANDS: list[dict[str, Any]] = [] + +_READ_ONLY_INVENTORY_PATTERNS = ( + re.compile( + r"\bgitea[_-](?:list|view|whoami|get[-_]|resolve[-_]task|check[-_]pr|route[-_]task)", + re.I, + ), + re.compile(r"\bgit\s+(?:fetch|remote\s+update|branch\s+-a|log|show|rev-parse)\b", re.I), + re.compile(r"\bgit\s+status\b", re.I), + re.compile(r"\bgit\s+worktree\s+list\b", re.I), +) + +_DIAGNOSTIC_PATTERNS = ( + re.compile(r"\bgit\s+diff(?:\s+--stat)?\b", re.I), + re.compile(r"\bwhich\s+pytest\b", re.I), + re.compile(r"\bpytest\s+--version\b", re.I), +) + +_BOUNDARY_VIOLATION_PATTERNS: tuple[tuple[re.Pattern[str], str], ...] = ( + (re.compile(r"\b(?:pytest|python\s+-m\s+pytest|python\s+-m\s+unittest)\b", re.I), + "validation command before workflow load"), + (re.compile(r"\bprofiles\.json\b", re.I), "local profile config inspection"), + (re.compile(r"\bgitea-mcp(?:\.v2-contexts)?\.json\b", re.I), + "local Gitea MCP config inspection"), + (re.compile(r"\b\.env(?:\.|$|\b)", re.I), "credential file inspection"), + (re.compile(r"\bkeychain\b", re.I), "credential store inspection"), + (re.compile(r"\bpkill\b", re.I), "MCP repair activity"), + (re.compile(r"\b(?:edit|write|modify).{0,40}\bmcp\b", re.I), + "MCP config exploration"), + (re.compile(r"\bgit\s+(?:add|commit|reset|clean|checkout|merge|rebase|push)\b", re.I), + "git mutation before workflow load"), +) + + +def clear_pre_review_commands() -> None: + """Test helper and session reset.""" + global _PRE_REVIEW_COMMANDS + _PRE_REVIEW_COMMANDS = [] + + +def pre_review_commands() -> list[dict[str, Any]]: + """Return a shallow copy of recorded pre-review commands.""" + return [dict(entry) for entry in _PRE_REVIEW_COMMANDS] + + +def _normalize_path(path: str | None) -> str: + return os.path.realpath(os.path.abspath((path or "").strip() or os.getcwd())) + + +def is_main_checkout_path(cwd: str | None, project_root: str | None) -> bool: + """True when *cwd* is the stable control checkout (not under branches/).""" + if not project_root: + return False + root = _normalize_path(project_root) + path = _normalize_path(cwd) + if path != root: + return False + marker = f"{os.sep}branches{os.sep}" + return marker not in path + + +def classify_pre_review_command( + command: str, + *, + cwd: str | None = None, + project_root: str | None = None, +) -> dict[str, Any]: + """Classify a command executed before workflow load.""" + text = (command or "").strip() + path = _normalize_path(cwd) + root = _normalize_path(project_root) if project_root else None + reasons: list[str] = [] + + for pattern, label in _BOUNDARY_VIOLATION_PATTERNS: + if pattern.search(text): + if label.startswith("validation") and root and not is_main_checkout_path(path, root): + continue + if label.startswith("git mutation") and root and not is_main_checkout_path(path, root): + continue + reasons.append(label) + return { + "command": text, + "cwd": path, + "classification": CLASSIFICATION_BOUNDARY_VIOLATION, + "reasons": reasons, + } + + for pattern in _READ_ONLY_INVENTORY_PATTERNS: + if pattern.search(text): + return { + "command": text, + "cwd": path, + "classification": CLASSIFICATION_READ_ONLY_INVENTORY, + "reasons": [], + } + + for pattern in _DIAGNOSTIC_PATTERNS: + if pattern.search(text): + return { + "command": text, + "cwd": path, + "classification": CLASSIFICATION_DIAGNOSTIC, + "reasons": [], + } + + if root and is_main_checkout_path(path, root): + if re.search(r"\b(?:cat|head|less|read)\b", text, re.I): + if re.search(r"workflow|skill|runbook", text, re.I): + return { + "command": text, + "cwd": path, + "classification": CLASSIFICATION_BOUNDARY_VIOLATION, + "reasons": [ + "canonical workflow viewed as local file without " + "gitea_load_review_workflow (narrative load is not proof)" + ], + } + + return { + "command": text, + "cwd": path, + "classification": CLASSIFICATION_UNCLASSIFIED, + "reasons": [ + "pre-review command not classified; record via " + "gitea_record_pre_review_command before workflow load" + ], + } + + +def record_pre_review_command( + command: str, + *, + cwd: str | None = None, + project_root: str | None = None, + classification: str | None = None, +) -> dict[str, Any]: + """Record and classify a pre-review command for the current session.""" + assessed = classify_pre_review_command( + command, cwd=cwd, project_root=project_root) + if classification: + if classification not in ALLOWED_CLASSIFICATIONS: + assessed["classification"] = CLASSIFICATION_UNCLASSIFIED + assessed["reasons"] = [ + f"unknown classification '{classification}'; fail closed" + ] + else: + assessed["classification"] = classification + assessed["reasons"] = [] + entry = { + **assessed, + "session_pid": os.getpid(), + } + _PRE_REVIEW_COMMANDS.append(entry) + return dict(entry) + + +def assess_boundary_status(project_root: str | None = None) -> dict[str, Any]: + """Summarize pre-review boundary state for session proof and reports.""" + violations = [ + entry for entry in _PRE_REVIEW_COMMANDS + if entry.get("classification") == CLASSIFICATION_BOUNDARY_VIOLATION + ] + unclassified = [ + entry for entry in _PRE_REVIEW_COMMANDS + if entry.get("classification") == CLASSIFICATION_UNCLASSIFIED + ] + reasons: list[str] = [] + for entry in violations: + reasons.extend(entry.get("reasons") or [ + f"boundary violation: {entry.get('command', '')[:80]}" + ]) + for entry in unclassified: + reasons.extend(entry.get("reasons") or [ + "unclassified pre-review command blocks reviewer mutations" + ]) + + clean = not reasons + return { + "boundary_status": "clean" if clean else "violation", + "boundary_clean": clean, + "pre_review_command_count": len(_PRE_REVIEW_COMMANDS), + "boundary_violation_count": len(violations), + "unclassified_command_count": len(unclassified), + "violations": [ + { + "command": v.get("command"), + "cwd": v.get("cwd"), + "reasons": list(v.get("reasons") or []), + } + for v in violations + ], + "reasons": reasons, + } + + +def boundary_blockers(project_root: str | None = None) -> list[str]: + """Reasons reviewer mutations must fail closed due to boundary state.""" + status = assess_boundary_status(project_root) + if status.get("boundary_clean"): + return [] + return list(status.get("reasons") or [ + "reviewer session boundary violation before workflow load" + ]) + + +def workflow_load_helper_result( + load: dict | None, + project_root: str | None = None, +) -> dict[str, Any]: + """Structured helper result for final reports (#403).""" + boundary = assess_boundary_status(project_root) + if load is None: + return { + "workflow_load_proof_present": False, + "workflow_source": None, + "workflow_hash": None, + "final_report_schema_hash": None, + "boundary_status": boundary.get("boundary_status"), + "boundary_clean": False, + "reasons": [ + "gitea_load_review_workflow helper result missing from report" + ], + } + return { + "workflow_load_proof_present": True, + "workflow_source": load.get("workflow_source"), + "workflow_hash": load.get("workflow_hash"), + "final_report_schema_path": load.get("final_report_schema_path"), + "final_report_schema_hash": load.get("final_report_schema_hash"), + "boundary_status": load.get("boundary_status", boundary.get("boundary_status")), + "boundary_clean": bool(load.get("boundary_clean", boundary.get("boundary_clean"))), + "pre_review_command_count": boundary.get("pre_review_command_count"), + "reasons": [], + } \ No newline at end of file diff --git a/review_workflow_load.py b/review_workflow_load.py index 070c29a..9cee3d0 100644 --- a/review_workflow_load.py +++ b/review_workflow_load.py @@ -1,4 +1,4 @@ -"""Canonical review-merge workflow load proof for reviewer mutations (#389).""" +"""Canonical review-merge workflow load proof for reviewer mutations (#389, #403).""" from __future__ import annotations @@ -7,6 +7,8 @@ import os import re from pathlib import Path +import review_workflow_boundary as boundary + WORKFLOW_REL_PATH = ( "skills/llm-project-workflow/workflows/review-merge-pr.md" ) @@ -95,10 +97,16 @@ def record_review_workflow_load( global _REVIEW_WORKFLOW_LOAD meta = build_canonical_workflow_metadata( project_root, prompt_text=prompt_text) + boundary_state = boundary.assess_boundary_status(project_root) _REVIEW_WORKFLOW_LOAD = { **meta, "session_pid": os.getpid(), "loaded": True, + "boundary_status": boundary_state.get("boundary_status"), + "boundary_clean": boundary_state.get("boundary_clean"), + "pre_review_command_count": boundary_state.get("pre_review_command_count"), + "boundary_violation_count": boundary_state.get("boundary_violation_count"), + "boundary_reasons": list(boundary_state.get("reasons") or []), } return dict(_REVIEW_WORKFLOW_LOAD) @@ -107,6 +115,7 @@ def clear_review_workflow_load() -> None: """Test helper and review_pr session reset.""" global _REVIEW_WORKFLOW_LOAD _REVIEW_WORKFLOW_LOAD = None + boundary.clear_pre_review_commands() def workflow_load_status(project_root: str | None = None) -> dict: @@ -125,6 +134,9 @@ def workflow_load_status(project_root: str | None = None) -> dict: ], } reasons = _session_validation_reasons(load, project_root) + boundary_reasons = boundary.boundary_blockers(project_root) + if boundary_reasons: + reasons = list(reasons) + boundary_reasons return { "workflow_load_proof_present": True, "workflow_load_valid": not reasons, @@ -136,6 +148,10 @@ def workflow_load_status(project_root: str | None = None) -> dict: "prompt_conflicts_with_workflow": load.get( "prompt_conflicts_with_workflow"), "session_pid": load.get("session_pid"), + "boundary_status": load.get("boundary_status"), + "boundary_clean": load.get("boundary_clean"), + "workflow_load_helper_result": boundary.workflow_load_helper_result( + load, project_root), "reasons": reasons, } @@ -179,9 +195,12 @@ def review_workflow_load_blockers( project_root: str | None = None, ) -> list[str]: """Reasons reviewer mutations must fail closed.""" + boundary_reasons = boundary.boundary_blockers(project_root) + if boundary_reasons and _REVIEW_WORKFLOW_LOAD is None: + return boundary_reasons status = workflow_load_status(project_root) if not status.get("workflow_load_proof_present"): - return list(status.get("reasons") or []) + return list(status.get("reasons") or []) + boundary_reasons if not status.get("workflow_load_valid"): return list(status.get("reasons") or []) return [] diff --git a/skills/llm-project-workflow/schemas/review-merge-final-report.md b/skills/llm-project-workflow/schemas/review-merge-final-report.md index 9eb3592..6ef00bc 100644 --- a/skills/llm-project-workflow/schemas/review-merge-final-report.md +++ b/skills/llm-project-workflow/schemas/review-merge-final-report.md @@ -63,8 +63,14 @@ Do not use legacy fields: `Pinned reviewed head`, `Scratch worktree used`, - Current status: - Safe next action: - Safety statement: +- Workflow-load helper result: ``` +The **Workflow-load helper result** field must carry structured output from +`gitea_load_review_workflow` (workflow_hash, final_report_schema_hash, +boundary_status). Narrative claims that workflow files were viewed locally are +not sufficient (#403). + ### Already-landed handoff overrides When eligibility class is `ALREADY_LANDED_RECONCILE_REQUIRED`: diff --git a/skills/llm-project-workflow/workflows/review-merge-pr.md b/skills/llm-project-workflow/workflows/review-merge-pr.md index f57bc23..029a57c 100644 --- a/skills/llm-project-workflow/workflows/review-merge-pr.md +++ b/skills/llm-project-workflow/workflows/review-merge-pr.md @@ -36,6 +36,44 @@ If available, load it first and report: If the canonical workflow cannot be loaded and the project requires it, stop and produce a recovery handoff only. +## 0A. Workflow-load and session boundary anchor (#403) + +The MCP gate is the authority — not local file viewing. + +Before any reviewer mutation: + +1. Record pre-review commands with `gitea_record_pre_review_command` when they + are not automatically classified (inventory/diagnostic commands may be + recorded explicitly for proof). +2. Call `gitea_load_review_workflow` to establish workflow hash proof **and** + session boundary state in the same in-process session proof. +3. Do not claim the workflow was loaded by reading + `skills/llm-project-workflow/workflows/review-merge-pr.md` as a local file; + that narrative does not satisfy the validator. + +Allowed before workflow load (classify as `read_only_inventory` or +`diagnostic`): + +* `gitea_whoami`, `gitea_resolve_task_capability`, `gitea_list_prs`, + `gitea_view_pr`, `gitea_get_runtime_context` +* `git fetch` / `git remote update` for inventory +* `git status`, `git worktree list` (read-only) + +Boundary violations (block downstream reviewer mutations even after load): + +* validation commands (`pytest`, `python -m unittest`) in the main checkout +* local profile/credential/config inspection (`profiles.json`, `gitea-mcp.json`, + `.env`, keychain dumps) +* MCP repair (`pkill`, MCP config edits) +* git mutations before workflow load + +Final reports must include a structured **Workflow-load helper result** block +copied from `gitea_load_review_workflow`, including at minimum: + +* `workflow_hash` +* `final_report_schema_hash` +* `boundary_status` (`clean` or `violation`) + ## 1. Start with live identity, profile, runtime, and capability checks Prove: diff --git a/tests/test_review_workflow_boundary.py b/tests/test_review_workflow_boundary.py new file mode 100644 index 0000000..2d1fc1a --- /dev/null +++ b/tests/test_review_workflow_boundary.py @@ -0,0 +1,138 @@ +"""Tests for workflow-load session boundary tracking (#403).""" + +import os +import sys +import unittest +from unittest.mock import patch + +sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent)) + +import final_report_validator +import review_workflow_boundary +import review_workflow_load +import mcp_server + + +class TestPreReviewClassification(unittest.TestCase): + def setUp(self): + review_workflow_boundary.clear_pre_review_commands() + review_workflow_load.clear_review_workflow_load() + + def test_inventory_command_allowed(self): + result = review_workflow_boundary.classify_pre_review_command( + "gitea_list_prs remote=prgs", + cwd="/tmp", + project_root="/repo/Gitea-Tools", + ) + self.assertEqual( + result["classification"], + review_workflow_boundary.CLASSIFICATION_READ_ONLY_INVENTORY, + ) + + def test_main_checkout_pytest_is_boundary_violation(self): + root = "/repo/Gitea-Tools" + result = review_workflow_boundary.classify_pre_review_command( + "python -m pytest tests/", + cwd=root, + project_root=root, + ) + self.assertEqual( + result["classification"], + review_workflow_boundary.CLASSIFICATION_BOUNDARY_VIOLATION, + ) + + def test_profiles_json_inspection_is_boundary_violation(self): + result = review_workflow_boundary.classify_pre_review_command( + "cat profiles.json", + cwd="/repo/Gitea-Tools", + project_root="/repo/Gitea-Tools", + ) + self.assertEqual( + result["classification"], + review_workflow_boundary.CLASSIFICATION_BOUNDARY_VIOLATION, + ) + + +class TestWorkflowLoadBoundaryGate(unittest.TestCase): + def setUp(self): + review_workflow_boundary.clear_pre_review_commands() + review_workflow_load.clear_review_workflow_load() + mcp_server.init_review_decision_lock("prgs", "review_pr") + mcp_server.record_preflight_check("whoami") + mcp_server.record_preflight_check("capability", "reviewer") + + def _root(self) -> str: + return str(__import__("pathlib").Path(__file__).resolve().parent.parent) + + def test_boundary_violation_blocks_mutation_after_load(self): + root = self._root() + review_workflow_boundary.record_pre_review_command( + "python -m pytest tests/", + cwd=root, + project_root=root, + ) + res = mcp_server.gitea_load_review_workflow() + self.assertFalse(res["success"]) + self.assertEqual(res["boundary_status"], "violation") + blocked = mcp_server.gitea_mark_final_review_decision( + 42, "approve", remote="prgs") + self.assertFalse(blocked["marked_ready"]) + joined = " ".join(blocked["reasons"]).lower() + self.assertTrue( + "validation" in joined or "boundary" in joined or "workflow" in joined + ) + + def test_clean_inventory_then_load_passes(self): + root = self._root() + review_workflow_boundary.record_pre_review_command( + "gitea_list_prs remote=prgs", + cwd=root, + project_root=root, + ) + res = mcp_server.gitea_load_review_workflow() + self.assertTrue(res["success"]) + self.assertEqual(res["boundary_status"], "clean") + blockers = review_workflow_load.review_workflow_load_blockers(root) + self.assertEqual(blockers, []) + + def test_file_view_narrative_fails_validator_without_helper(self): + report = ( + "## Controller Handoff\n" + "- Task: review-merge-pr\n" + "- I read the canonical workflow review-merge-pr.md before review.\n" + ) + findings = final_report_validator.assess_final_report_validator( + report, + task_kind="review_pr", + ) + self.assertTrue(any( + f["rule_id"] == "reviewer.workflow_load_boundary" + for f in findings.get("findings") or [] + )) + + def test_helper_result_passes_validator(self): + root = self._root() + review_workflow_load.record_review_workflow_load(root) + helper = review_workflow_boundary.workflow_load_helper_result( + review_workflow_load._REVIEW_WORKFLOW_LOAD, + root, + ) + report = ( + "## Controller Handoff\n" + "- Task: review-merge-pr\n" + f"- Workflow-load helper result: workflow_hash: {helper['workflow_hash']}; " + f"boundary_status: {helper['boundary_status']}\n" + ) + findings = final_report_validator.assess_final_report_validator( + report, + task_kind="review_pr", + ) + boundary_findings = [ + f for f in (findings.get("findings") or []) + if f.get("rule_id") == "reviewer.workflow_load_boundary" + ] + self.assertEqual(boundary_findings, []) + + +if __name__ == "__main__": + unittest.main() \ No newline at end of file