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

Merge pull request 'feat: add workflow-load session boundary proof (Closes #403)' (#422) from feat/issue-403-workflow-load-boundaries into master

Browse Source
This commit was merged in pull request #422.
This commit is contained in:
sysadmin
2026-07-08 10:42:49 -05:00
parent 2bb45c0f80 cefebd7643
commit 7231f27c1c
11 changed files with 589 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files
final_report_validator.py
+65
View File
@@ -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,
gitea_mcp_server.py
+43 -4
View File
@@ -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 []
),
}
review_workflow_boundary.py
+259
View File
@@ -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": [],
}
review_workflow_load.py
+21 -2
View File
@@ -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 []
skills/llm-project-workflow/schemas/review-merge-final-report.md
+6
View File
@@ -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`:
skills/llm-project-workflow/workflows/review-merge-pr.md
+38
View File
@@ -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:
tests/test_audit.py
+4
View File
@@ -294,9 +294,13 @@ class TestGatedToolAudit(_AuditWiringBase):
super().setUp()
from tests.test_mcp_server import _init_reviewer_session, _install_owned_reviewer_lease
import reviewer_pr_lease
import review_workflow_boundary
import review_workflow_load
# Session init clears any prior session lease (#407) and loads workflow (#389).
_init_reviewer_session("prgs")
self.addCleanup(review_workflow_load.clear_review_workflow_load)
self.addCleanup(review_workflow_boundary.clear_pre_review_commands)
self._lease_patch = _install_owned_reviewer_lease(8)
self._lease_patch.start()
self._auth_identity_patch = patch(
tests/test_mcp_server.py
+2 -1
View File
@@ -1057,7 +1057,8 @@ class TestMergePR(unittest.TestCase):
"GITEA_ALLOWED_OPERATIONS": "read,merge"}
with patch.dict(os.environ, env, clear=True):
r = gitea_merge_pr(
pr_number=8, confirmation=self._confirm(8), remote="prgs")
pr_number=8, confirmation=self._confirm(8), remote="prgs",
expected_head_sha=new_sha)
self.assertFalse(r["performed"])
self.assertTrue(r.get("approval_visible"))
self.assertFalse(r.get("approval_at_current_head"))
tests/test_permission_reports.py
+4
View File
@@ -95,8 +95,12 @@ class PermissionReportBase(unittest.TestCase):
self._dir = tempfile.TemporaryDirectory()
self.config_path = os.path.join(self._dir.name, "profiles.json")
self._write_config(CONFIG)
import review_workflow_load
review_workflow_load.record_review_workflow_load(mcp_server.PROJECT_ROOT)
def tearDown(self):
import review_workflow_load
review_workflow_load.clear_review_workflow_load()
self._remotes.stop()
mcp_server._IDENTITY_CACHE.clear()
mcp_server.review_workflow_load.clear_review_workflow_load()
tests/test_review_workflow_boundary.py
+138
View File
@@ -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()
tests/test_terminal_review_hard_stop.py
+9
View File
@@ -36,10 +36,13 @@ def _lock(mutations=None, correction=False):
def _seed(mutations=None, correction=False):
import review_workflow_load
review_workflow_load.record_review_workflow_load(mcp_server.PROJECT_ROOT)
mcp_server._save_review_decision_lock(_lock(mutations, correction))
mcp_server.gitea_load_review_workflow()
APPROVED_A = {"pr_number": 5, "action": "approve", "review_id": 1,
"review_state": "approve"}
RC_A = {"pr_number": 5, "action": "request_changes", "review_id": 2,
@@ -94,6 +97,8 @@ class TestTerminalHardStopReasons(unittest.TestCase):
class TestMergeHardStopWiring(unittest.TestCase):
def tearDown(self):
import review_workflow_load
review_workflow_load.clear_review_workflow_load()
mcp_server._save_review_decision_lock(None)
mcp_server.review_workflow_load.clear_review_workflow_load()
@@ -116,6 +121,8 @@ class TestMergeHardStopWiring(unittest.TestCase):
class TestMarkFinalHardStopWiring(unittest.TestCase):
def tearDown(self):
import review_workflow_load
review_workflow_load.clear_review_workflow_load()
mcp_server._save_review_decision_lock(None)
mcp_server.review_workflow_load.clear_review_workflow_load()
@@ -149,6 +156,8 @@ def _mark(action, pr_number=6, **kwargs):
class TestDuplicateRequestChangesSuppression(unittest.TestCase):
def tearDown(self):
import review_workflow_load
review_workflow_load.clear_review_workflow_load()
mcp_server._save_review_decision_lock(None)
mcp_server.review_workflow_load.clear_review_workflow_load()