feat: gate reviewer mutations on canonical workflow load proof (Closes #389)
Add gitea_load_review_workflow and in-process session proof required before gitea_submit_pr_review, gitea_mark_final_review_decision, and gitea_merge_pr. Capability and runtime context report workflow-load status; stale or missing proof fails closed with recovery handoff text that forbids approve/merge replay.
This commit is contained in:
@@ -0,0 +1,198 @@
|
||||
"""Canonical review-merge workflow load proof for reviewer mutations (#389)."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import os
|
||||
import re
|
||||
from pathlib import Path
|
||||
|
||||
WORKFLOW_REL_PATH = (
|
||||
"skills/llm-project-workflow/workflows/review-merge-pr.md"
|
||||
)
|
||||
SCHEMA_REL_PATH = (
|
||||
"skills/llm-project-workflow/schemas/review-merge-final-report.md"
|
||||
)
|
||||
TASK_MODE = "review-merge-pr"
|
||||
LOAD_TOOL_NAME = "gitea_load_review_workflow"
|
||||
|
||||
_REVIEW_WORKFLOW_LOAD: dict | None = None
|
||||
|
||||
|
||||
def compute_content_hash(text: str) -> str:
|
||||
"""Short deterministic hash for workflow/schema version proof."""
|
||||
return hashlib.sha256((text or "").encode("utf-8")).hexdigest()[:12]
|
||||
|
||||
|
||||
def _read_text(path: Path) -> str:
|
||||
return path.read_text(encoding="utf-8")
|
||||
|
||||
|
||||
def _canonical_paths(project_root: str) -> tuple[Path, Path]:
|
||||
root = Path(project_root)
|
||||
workflow = root / WORKFLOW_REL_PATH
|
||||
schema = root / SCHEMA_REL_PATH
|
||||
if not workflow.is_file():
|
||||
raise FileNotFoundError(f"canonical workflow missing: {workflow}")
|
||||
if not schema.is_file():
|
||||
raise FileNotFoundError(f"final report schema missing: {schema}")
|
||||
return workflow, schema
|
||||
|
||||
|
||||
def build_canonical_workflow_metadata(
|
||||
project_root: str,
|
||||
*,
|
||||
prompt_text: str | None = None,
|
||||
) -> dict:
|
||||
"""Load workflow + schema from disk and compute proof metadata."""
|
||||
workflow_path, schema_path = _canonical_paths(project_root)
|
||||
workflow_text = _read_text(workflow_path)
|
||||
schema_text = _read_text(schema_path)
|
||||
workflow_hash = compute_content_hash(workflow_text)
|
||||
schema_hash = compute_content_hash(schema_text)
|
||||
conflict, conflict_reasons = assess_prompt_conflict(prompt_text)
|
||||
return {
|
||||
"workflow_source": WORKFLOW_REL_PATH,
|
||||
"workflow_path": str(workflow_path),
|
||||
"task_mode": TASK_MODE,
|
||||
"workflow_hash": workflow_hash,
|
||||
"workflow_version": workflow_hash,
|
||||
"final_report_schema_path": SCHEMA_REL_PATH,
|
||||
"final_report_schema_hash": schema_hash,
|
||||
"prompt_conflicts_with_workflow": conflict,
|
||||
"prompt_conflict_reasons": conflict_reasons,
|
||||
"load_tool": LOAD_TOOL_NAME,
|
||||
}
|
||||
|
||||
|
||||
def assess_prompt_conflict(prompt_text: str | None) -> tuple[bool, list[str]]:
|
||||
"""Detect obvious task-mode conflicts between prompt and review workflow."""
|
||||
if not (prompt_text or "").strip():
|
||||
return False, []
|
||||
text = prompt_text.lower()
|
||||
reasons: list[str] = []
|
||||
conflicting = (
|
||||
(r"\bwork[- ]issue\b", "work-issue author mode"),
|
||||
(r"\bcreate[- ]issue\b", "create-issue mode"),
|
||||
(r"\bauthor/coder\b", "author/coder mode"),
|
||||
(r"\breconcile[- ]landed\b", "reconcile-landed mode"),
|
||||
)
|
||||
for pattern, label in conflicting:
|
||||
if re.search(pattern, text):
|
||||
reasons.append(
|
||||
f"active prompt appears to request {label} while loading "
|
||||
f"{TASK_MODE} workflow"
|
||||
)
|
||||
return bool(reasons), reasons
|
||||
|
||||
|
||||
def record_review_workflow_load(
|
||||
project_root: str,
|
||||
*,
|
||||
prompt_text: str | None = None,
|
||||
) -> dict:
|
||||
"""Record in-process workflow load proof for the current MCP session."""
|
||||
global _REVIEW_WORKFLOW_LOAD
|
||||
meta = build_canonical_workflow_metadata(
|
||||
project_root, prompt_text=prompt_text)
|
||||
_REVIEW_WORKFLOW_LOAD = {
|
||||
**meta,
|
||||
"session_pid": os.getpid(),
|
||||
"loaded": True,
|
||||
}
|
||||
return dict(_REVIEW_WORKFLOW_LOAD)
|
||||
|
||||
|
||||
def clear_review_workflow_load() -> None:
|
||||
"""Test helper and review_pr session reset."""
|
||||
global _REVIEW_WORKFLOW_LOAD
|
||||
_REVIEW_WORKFLOW_LOAD = None
|
||||
|
||||
|
||||
def workflow_load_status(project_root: str | None = None) -> dict:
|
||||
"""Non-throwing status for capability/runtime reports."""
|
||||
load = _REVIEW_WORKFLOW_LOAD
|
||||
if load is None:
|
||||
return {
|
||||
"workflow_load_proof_present": False,
|
||||
"workflow_load_valid": False,
|
||||
"workflow_source": None,
|
||||
"workflow_hash": None,
|
||||
"final_report_schema_path": SCHEMA_REL_PATH,
|
||||
"reasons": [
|
||||
f"{LOAD_TOOL_NAME} has not been called in this session "
|
||||
"(fail closed for reviewer mutations)"
|
||||
],
|
||||
}
|
||||
reasons = _session_validation_reasons(load, project_root)
|
||||
return {
|
||||
"workflow_load_proof_present": True,
|
||||
"workflow_load_valid": not reasons,
|
||||
"workflow_source": load.get("workflow_source"),
|
||||
"workflow_hash": load.get("workflow_hash"),
|
||||
"task_mode": load.get("task_mode"),
|
||||
"final_report_schema_path": load.get("final_report_schema_path"),
|
||||
"final_report_schema_hash": load.get("final_report_schema_hash"),
|
||||
"prompt_conflicts_with_workflow": load.get(
|
||||
"prompt_conflicts_with_workflow"),
|
||||
"session_pid": load.get("session_pid"),
|
||||
"reasons": reasons,
|
||||
}
|
||||
|
||||
|
||||
def _session_validation_reasons(
|
||||
load: dict,
|
||||
project_root: str | None,
|
||||
) -> list[str]:
|
||||
reasons: list[str] = []
|
||||
if load.get("session_pid") != os.getpid():
|
||||
reasons.append(
|
||||
"workflow load proof was recorded in a different process "
|
||||
"(fail closed)"
|
||||
)
|
||||
return reasons
|
||||
if load.get("prompt_conflicts_with_workflow"):
|
||||
reasons.extend(load.get("prompt_conflict_reasons") or [
|
||||
"active prompt conflicts with loaded review-merge workflow"
|
||||
])
|
||||
if project_root:
|
||||
try:
|
||||
current = build_canonical_workflow_metadata(project_root)
|
||||
except OSError as exc:
|
||||
reasons.append(f"cannot re-verify workflow hash: {exc}")
|
||||
return reasons
|
||||
if current["workflow_hash"] != load.get("workflow_hash"):
|
||||
reasons.append(
|
||||
"stored workflow hash is stale; reload via "
|
||||
f"{LOAD_TOOL_NAME} (fail closed)"
|
||||
)
|
||||
if current["final_report_schema_hash"] != load.get(
|
||||
"final_report_schema_hash"):
|
||||
reasons.append(
|
||||
"stored final-report schema hash is stale; reload via "
|
||||
f"{LOAD_TOOL_NAME} (fail closed)"
|
||||
)
|
||||
return reasons
|
||||
|
||||
|
||||
def review_workflow_load_blockers(
|
||||
project_root: str | None = None,
|
||||
) -> list[str]:
|
||||
"""Reasons reviewer mutations must fail closed."""
|
||||
status = workflow_load_status(project_root)
|
||||
if not status.get("workflow_load_proof_present"):
|
||||
return list(status.get("reasons") or [])
|
||||
if not status.get("workflow_load_valid"):
|
||||
return list(status.get("reasons") or [])
|
||||
return []
|
||||
|
||||
|
||||
def recovery_handoff_without_replay() -> list[str]:
|
||||
"""Safe next-step lines that must not include approve/merge replay."""
|
||||
return [
|
||||
"Reload the canonical workflow via gitea_load_review_workflow, then "
|
||||
"rerun the full review-merge workflow from inventory.",
|
||||
"Do not call gitea_submit_pr_review, gitea_mark_final_review_decision, "
|
||||
"or gitea_merge_pr until workflow-load proof is present.",
|
||||
"Do not include approve/merge replay commands in the recovery handoff.",
|
||||
]
|
||||
Reference in New Issue
Block a user