scripts/validate_runbook_consistency.py

#!/usr/bin/env python3
"""Strict consistency validator for the Hermes canonical runbook.

The validator treats task headings, internal checklists, checkpoint state,
dependencies, completion evidence, YAML front matter, the Progress Dashboard,
the Current Session Handoff, gap counters, and current evidence hashes as one
ledger.  It intentionally validates *consistency*, not project completeness:
open tasks are allowed, but a task may not be presented as complete unless all
of its recorded requirements and dependencies are complete.

Exit codes:
  0 = consistent
  1 = inconsistencies found
  2 = runbook unreadable / invalid UTF-8 / malformed YAML
"""
from __future__ import annotations

import argparse
import hashlib
import re
import sys
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any, Iterable

try:
    import yaml
except ImportError:  # pragma: no cover - deployment dependency guard
    yaml = None


TASK_HEADING_RE = re.compile(
    r"^### \[([ xX~!])\] "
    r"((?:P\d+(?:A)?-T\d+|S\d+-T\d+|OP-[A-Z0-9-]+))\s+—\s+(.+)$",
    re.MULTILINE,
)
CHECKPOINT_START_RE = re.compile(r"<!-- TASK_CHECKPOINT_START:([^>]+) -->")
CHECKPOINT_END_RE = re.compile(r"<!-- TASK_CHECKPOINT_END:([^>]+) -->")
TERMINAL_STATUSES = {"DONE_VERIFIED", "NOT_APPLICABLE"}
PROVISIONAL_STATUSES = {"WEB_IMPLEMENTED_PENDING_DESKTOP_VERIFICATION"}
INCOMPLETE_CHECKPOINT_STATUSES = {"TODO", "IN_PROGRESS", "BLOCKED"}
REQUIRED_YAML_KEYS = {
    "runbook_version",
    "project_name",
    "current_track",
    "current_phase",
    "current_task",
    "current_task_status",
    "last_done_verified_task",
    "current_release_mode",
    "paper_only_required",
    "open_p0_gaps",
    "open_p1_gaps",
    "operator_actions_open",
}


@dataclass
class Task:
    task_id: str
    title: str
    checked: bool
    phase: str
    start: int
    end: int
    body: str
    checkpoint_status: str | None = None
    dependencies_text: str = "None"
    dependency_ids: list[str] = field(default_factory=list)
    dependency_phases: list[str] = field(default_factory=list)
    dependency_gates: list[str] = field(default_factory=list)
    checklist_unfinished: list[str] = field(default_factory=list)
    unfinished_by_section: dict[str, list[str]] = field(default_factory=dict)
    completion_evidence: str | None = None

    @property
    def complete(self) -> bool:
        return self.checked and self.checkpoint_status in TERMINAL_STATUSES

    @property
    def dependency_ready(self) -> bool:
        return self.complete or self.checkpoint_status in PROVISIONAL_STATUSES


@dataclass
class ValidationResult:
    errors: list[str]
    warnings: list[str]
    metrics: dict[str, Any]

    @property
    def ok(self) -> bool:
        return not self.errors


def sha256_file(path: Path) -> str:
    digest = hashlib.sha256()
    with path.open("rb") as handle:
        for chunk in iter(lambda: handle.read(65536), b""):
            digest.update(chunk)
    return digest.hexdigest()


def parse_yaml_front_matter(text: str) -> dict[str, Any]:
    if not text.startswith("---\n"):
        raise ValueError("YAML front matter missing")
    end = text.find("\n---\n", 4)
    if end < 0:
        raise ValueError("YAML front matter closing delimiter missing")
    if yaml is None:
        raise ValueError("PyYAML is required to parse front matter")
    data = yaml.safe_load(text[4:end])
    if not isinstance(data, dict):
        raise ValueError("YAML front matter must be a mapping")
    return data


PHASE_ORDER = [
    "Phase 0", "Phase 1", "Phase 2", "Phase 3", "Phase 4", "Phase 5",
    "Phase 6", "Phase 7", "Phase 8", "Phase 9", "Phase 10", "Phase 10A",
    "Phase 11", "Phase S1",
]


def _expand_phase_range(start: str, end: str) -> list[str]:
    start_label = f"Phase {start}"
    end_label = f"Phase {end}"
    if start_label not in PHASE_ORDER or end_label not in PHASE_ORDER:
        return []
    left = PHASE_ORDER.index(start_label)
    right = PHASE_ORDER.index(end_label)
    if left > right:
        return []
    return PHASE_ORDER[left:right + 1]


def _parse_phase_dependencies(value: str) -> list[str]:
    if "Phase" not in value:
        return []
    normalized = value.replace("–", "-").replace("—", "-")
    phases: list[str] = []
    for start, end in re.findall(r"\bPhases?\s+([0-9]+A?|S1)\s*-\s*([0-9]+A?|S1)\b", normalized):
        phases.extend(_expand_phase_range(start, end))
    for phase in re.findall(r"\bPhase\s+([0-9]+A?|S1)\b", normalized):
        phases.append(f"Phase {phase}")
    plural = re.search(r"\bPhases\s+([0-9]+A?|S1)((?:\s*,\s*(?:[0-9]+A?|S1))+)", normalized)
    if plural:
        phases.append(f"Phase {plural.group(1)}")
        phases.extend(f"Phase {item}" for item in re.findall(r"[0-9]+A?|S1", plural.group(2)))
    return list(dict.fromkeys(phases))


def _parse_gate_dependencies(value: str) -> list[str]:
    return [f"Gate {gate}" for gate in re.findall(r"\bRelease Gate\s+([A-F])\b", value)]


def _parse_release_gates(text: str) -> dict[str, str]:
    result: dict[str, str] = {}
    pattern = re.compile(
        r"^## Gate ([A-F])\s+—.*?^\*\*Status:\*\*\s*`([^`]+)`",
        re.MULTILINE | re.DOTALL,
    )
    for match in pattern.finditer(text):
        result[f"Gate {match.group(1)}"] = match.group(2).strip().upper()
    return result


def _phase_positions(text: str) -> list[tuple[int, str]]:
    return [(m.start(), f"Phase {m.group(1)}") for m in re.finditer(r"^# Phase ([0-9]+A?|S1)\b", text, re.MULTILINE)]


def _phase_at(position: int, phases: list[tuple[int, str]]) -> str:
    current = "Unknown"
    for start, name in phases:
        if start > position:
            break
        current = name
    return current


def _section(body: str, heading: str) -> str:
    match = re.search(rf"^#### {re.escape(heading)}\s*$", body, re.MULTILINE)
    if not match:
        return ""
    next_heading = re.search(r"^#### ", body[match.end():], re.MULTILINE)
    end = match.end() + next_heading.start() if next_heading else len(body)
    return body[match.end():end]


def parse_tasks(text: str) -> list[Task]:
    matches = list(TASK_HEADING_RE.finditer(text))
    phases = _phase_positions(text)
    tasks: list[Task] = []
    for index, match in enumerate(matches):
        end = matches[index + 1].start() if index + 1 < len(matches) else len(text)
        body = text[match.end():end]
        checkpoint = re.search(r"\*\*Checkpoint:\*\*\s*`([^`]+)`", body)
        dependencies = re.search(r"\*\*Dependencies:\*\*\s*(.+)", body)
        evidence = re.search(r"\*\*Completion evidence:\*\*\s*(.+)", body)
        unfinished: list[str] = []
        unfinished_by_section: dict[str, list[str]] = {}
        for section_name in ("Implementation checklist", "Verification checklist", "Acceptance criteria"):
            section = _section(body, section_name)
            section_unfinished: list[str] = []
            for line in section.splitlines():
                if re.match(r"^- \[(?: |~)\]", line):
                    item = line.strip()
                    unfinished.append(f"{section_name}: {item}")
                    section_unfinished.append(item)
            unfinished_by_section[section_name] = section_unfinished
        dep_text = dependencies.group(1).strip() if dependencies else "None"
        dep_ids = re.findall(r"\b(?:P\d+(?:A)?-T\d+|S\d+-T\d+|OP-[A-Z0-9-]+)\b", dep_text)
        tasks.append(Task(
            task_id=match.group(2),
            title=match.group(3).strip(),
            checked=match.group(1).lower() == "x",
            phase=_phase_at(match.start(), phases),
            start=match.start(),
            end=end,
            body=body,
            checkpoint_status=checkpoint.group(1).strip() if checkpoint else None,
            dependencies_text=dep_text,
            dependency_ids=dep_ids,
            dependency_phases=_parse_phase_dependencies(dep_text),
            dependency_gates=_parse_gate_dependencies(dep_text),
            checklist_unfinished=unfinished,
            unfinished_by_section=unfinished_by_section,
            completion_evidence=evidence.group(1).strip() if evidence else None,
        ))
    return tasks


def _parse_dashboard(text: str) -> tuple[dict[str, tuple[str, int, int]], dict[str, str]]:
    dashboard: dict[str, tuple[str, int, int]] = {}
    section_match = re.search(r"^# 4\. Progress Dashboard\s*$", text, re.MULTILINE)
    if section_match:
        end_match = re.search(r"^# 5\.", text[section_match.end():], re.MULTILINE)
        block_end = section_match.end() + end_match.start() if end_match else len(text)
        block = text[section_match.end():block_end]
        for line in block.splitlines():
            m = re.match(
                r"^\|\s*([0-9]+A?|S1)\s*\|.*?\|\s*([A-Z_]+)\s*\|\s*(\d+)\s*/\s*(\d+)(?:\s+reconciled)?\s*\|",
                line,
            )
            if m:
                dashboard[f"Phase {m.group(1)}"] = (m.group(2), int(m.group(3)), int(m.group(4)))
    operational: dict[str, str] = {}
    state = re.search(r"^## Current operational state\s*$([\s\S]*?)(?=^---$|^# )", text, re.MULTILINE)
    if state:
        for line in state.group(1).splitlines():
            m = re.match(r"^- \*\*(.+?):\*\*\s*(.+)$", line)
            if m:
                operational[m.group(1).strip()] = m.group(2).strip()
    return dashboard, operational


def _parse_handoff(text: str) -> dict[str, str]:
    match = re.search(r"^## Current Session Handoff\s*$([\s\S]*?)(?=^---$|^# Final Rule)", text, re.MULTILINE)
    result: dict[str, str] = {}
    if not match:
        return result
    for line in match.group(1).splitlines():
        field_match = re.match(r"^- \*\*(.+?):\*\*\s*(.+)$", line)
        if field_match:
            result[field_match.group(1).strip()] = field_match.group(2).strip()
    return result


def _find_gaps(text: str) -> list[dict[str, str]]:
    matches = list(re.finditer(r"^### (GAP-\d+)\s+—\s+(.+)$", text, re.MULTILINE))
    gaps: list[dict[str, str]] = []
    for index, match in enumerate(matches):
        end = matches[index + 1].start() if index + 1 < len(matches) else len(text)
        body = text[match.end():end]
        status = re.search(r"^- \*\*Status:\*\*\s*(\w+)", body, re.MULTILINE)
        severity = re.search(r"^- \*\*Severity:\*\*\s*(\w+)", body, re.MULTILINE)
        gaps.append({
            "id": match.group(1),
            "status": status.group(1) if status else "UNKNOWN",
            "severity": severity.group(1) if severity else "UNKNOWN",
        })
    return gaps


def _clean_task_ref(value: str) -> str:
    value = value.strip().strip("`")
    match = re.search(r"\b(?:P\d+(?:A)?-T\d+|S\d+-T\d+|OP-[A-Z0-9-]+)\b", value)
    return match.group(0) if match else value


def _evidence_hash_errors(root: Path, evidence_path: Path | None) -> list[str]:
    if evidence_path is None:
        evidence_path = root / "evidence_package" / "EVIDENCE_LATEST.md"
    if not evidence_path.is_file():
        return []
    errors: list[str] = []
    text = evidence_path.read_text(encoding="utf-8")
    for rel, expected in re.findall(r"^- `([^`]+)`: `([0-9a-fA-F]{64})`\s*$", text, re.MULTILINE):
        path = root / rel
        if not path.is_file():
            errors.append(f"evidence hash target missing: {rel}")
            continue
        actual = sha256_file(path)
        if actual.lower() != expected.lower():
            errors.append(f"evidence hash mismatch for {rel}: expected {expected}, actual {actual}")
    return errors


def validate_text(
    text: str,
    *,
    root: Path | None = None,
    evidence_path: Path | None = None,
    check_evidence_hashes: bool = True,
) -> ValidationResult:
    errors: list[str] = []
    warnings: list[str] = []
    metrics: dict[str, Any] = {}
    root = root or Path.cwd()

    try:
        front = parse_yaml_front_matter(text)
    except ValueError as exc:
        return ValidationResult([str(exc)], warnings, metrics)

    missing_keys = sorted(REQUIRED_YAML_KEYS - set(front))
    errors.extend(f"YAML missing key: {key}" for key in missing_keys)

    tasks = parse_tasks(text)
    by_id = {task.task_id: task for task in tasks}
    metrics["tasks_total"] = len(tasks)
    metrics["tasks_checked"] = sum(task.checked for task in tasks)
    metrics["tasks_open"] = len(tasks) - metrics["tasks_checked"]
    if len(by_id) != len(tasks):
        duplicates = sorted({task.task_id for task in tasks if sum(t.task_id == task.task_id for t in tasks) > 1})
        errors.append(f"duplicate task IDs: {', '.join(duplicates)}")

    starts = CHECKPOINT_START_RE.findall(text)
    ends = CHECKPOINT_END_RE.findall(text)
    if len(starts) != len(set(starts)):
        errors.append("duplicate TASK_CHECKPOINT_START marker IDs")
    if len(ends) != len(set(ends)):
        errors.append("duplicate TASK_CHECKPOINT_END marker IDs")
    if sorted(starts) != sorted(ends):
        errors.append("checkpoint start/end marker IDs are not paired")
    task_ids = sorted(by_id)
    if sorted(starts) != task_ids:
        missing = sorted(set(task_ids) - set(starts))
        extra = sorted(set(starts) - set(task_ids))
        errors.append(f"checkpoint markers do not match tasks; missing={missing}, extra={extra}")

    for task in tasks:
        if task.checked:
            if task.checklist_unfinished:
                errors.append(
                    f"checked task {task.task_id} has {len(task.checklist_unfinished)} unfinished internal checklist item(s)"
                )
            if task.checkpoint_status in INCOMPLETE_CHECKPOINT_STATUSES:
                errors.append(f"checked task {task.task_id} has incomplete checkpoint {task.checkpoint_status}")
            if task.checkpoint_status not in TERMINAL_STATUSES:
                errors.append(
                    f"checked task {task.task_id} checkpoint must be terminal DONE_VERIFIED/NOT_APPLICABLE, got {task.checkpoint_status!r}"
                )
            evidence = task.completion_evidence or ""
            if not evidence:
                errors.append(f"checked task {task.task_id} is missing completion evidence")
            elif re.search(r"\b(?:Pending|Missing|NOT_RUN)\b|^\s*(?:—|-)+\s*$", evidence, re.IGNORECASE):
                errors.append(f"checked task {task.task_id} has pending or missing completion evidence")
        elif task.checkpoint_status in PROVISIONAL_STATUSES:
            evidence = task.completion_evidence or ""
            if task.unfinished_by_section.get("Implementation checklist"):
                errors.append(
                    f"provisional task {task.task_id} has unfinished implementation checklist item(s)"
                )
            if task.unfinished_by_section.get("Acceptance criteria"):
                errors.append(
                    f"provisional task {task.task_id} has unfinished acceptance criteria"
                )
            verification_items = task.unfinished_by_section.get("Verification checklist", [])
            if not verification_items:
                errors.append(
                    f"provisional task {task.task_id} must identify an unfinished Desktop-only verification item"
                )
            elif any(
                not re.search(r"desktop|docker|container|browser|ci environment|external", item, re.IGNORECASE)
                for item in verification_items
            ):
                errors.append(
                    f"provisional task {task.task_id} has a non-Desktop verification item still unfinished"
                )
            if not evidence or re.search(r"\b(?:Pending|Missing|NOT_RUN)\b", evidence, re.IGNORECASE):
                errors.append(
                    f"provisional task {task.task_id} requires concrete implementation evidence"
                )
            if not re.search(r"desktop|docker|container|browser|ci", evidence, re.IGNORECASE):
                errors.append(
                    f"provisional task {task.task_id} evidence must name the Desktop-only limitation"
                )
        elif task.checkpoint_status == "DONE_VERIFIED":
            errors.append(f"open task {task.task_id} has DONE_VERIFIED checkpoint")
        elif task.checkpoint_status in {"BLOCKED", "FAILED", "DEFERRED"}:
            evidence = task.completion_evidence or ""
            if not evidence or re.search(r"\b(?:Pending|Missing|NOT_RUN)\b", evidence, re.IGNORECASE):
                errors.append(
                    f"open task {task.task_id} with checkpoint {task.checkpoint_status} requires concrete evidence"
                )

    # Explicit dependency ordering and completion.
    for task in tasks:
        if not task.checked:
            continue
        for dep_id in task.dependency_ids:
            dep = by_id.get(dep_id)
            if dep is None:
                errors.append(f"task {task.task_id} references unknown dependency {dep_id}")
            elif not dep.dependency_ready:
                errors.append(f"checked task {task.task_id} has unmet dependency {dep_id}")
            elif dep.start > task.start:
                errors.append(f"task {task.task_id} is completed before later-listed dependency {dep_id}")

    phase_tasks: dict[str, list[Task]] = {}
    for task in tasks:
        phase_tasks.setdefault(task.phase, []).append(task)
    gates = _parse_release_gates(text)
    for task in tasks:
        if not task.checked:
            continue
        for phase in task.dependency_phases:
            required = [candidate for candidate in phase_tasks.get(phase, []) if not candidate.task_id.startswith("OP-")]
            incomplete = [candidate.task_id for candidate in required if not candidate.dependency_ready]
            if incomplete:
                errors.append(
                    f"checked task {task.task_id} has unmet phase dependency {phase}: {', '.join(incomplete)}"
                )
        for gate in task.dependency_gates:
            if gates.get(gate) != "APPROVED":
                errors.append(
                    f"checked task {task.task_id} requires {gate} APPROVED, got {gates.get(gate, 'MISSING')}"
                )

    dashboard, operational = _parse_dashboard(text)
    for phase, phase_list in phase_tasks.items():
        if phase == "Unknown":
            continue
        actual_done = sum(task.checked for task in phase_list)
        actual_total = len(phase_list)
        if phase not in dashboard:
            errors.append(f"Progress Dashboard missing row for {phase}")
            continue
        status, declared_done, declared_total = dashboard[phase]
        if (declared_done, declared_total) != (actual_done, actual_total):
            errors.append(
                f"Progress Dashboard {phase} counter {declared_done}/{declared_total} "
                f"does not match actual {actual_done}/{actual_total}"
            )
        expected_status = "DONE_VERIFIED" if actual_done == actual_total and actual_total else (
            "IN_PROGRESS" if str(front.get("current_phase")) == phase else "TODO"
        )
        if status != expected_status:
            errors.append(f"Progress Dashboard {phase} status {status} does not match expected {expected_status}")

    current_task = str(front.get("current_task") or "")
    current_phase = str(front.get("current_phase") or "")
    current_status = str(front.get("current_task_status") or "")
    active = by_id.get(current_task)
    if not active:
        errors.append(f"YAML current_task {current_task!r} does not identify an existing task")
    else:
        if active.phase != current_phase:
            errors.append(f"YAML current_phase {current_phase!r} does not match {current_task} phase {active.phase}")
        if active.checked or active.checkpoint_status in TERMINAL_STATUSES:
            errors.append(f"active task {current_task} is already completed")
        if active.checkpoint_status != current_status:
            errors.append(
                f"YAML current_task_status {current_status!r} disagrees with {current_task} checkpoint {active.checkpoint_status!r}"
            )
        for dep_id in active.dependency_ids:
            dep = by_id.get(dep_id)
            if dep is None or not dep.dependency_ready:
                errors.append(f"active task {current_task} is not eligible; unmet dependency {dep_id}")
        for phase in active.dependency_phases:
            required = [candidate for candidate in phase_tasks.get(phase, []) if not candidate.task_id.startswith("OP-")]
            incomplete = [candidate.task_id for candidate in required if not candidate.dependency_ready]
            if incomplete:
                errors.append(
                    f"active task {current_task} is not eligible; unmet phase dependency {phase}: {', '.join(incomplete)}"
                )
        for gate in active.dependency_gates:
            if gates.get(gate) != "APPROVED":
                errors.append(
                    f"active task {current_task} is not eligible; {gate} is {gates.get(gate, 'MISSING')}"
                )

    op_phase = operational.get("Active phase", "")
    op_task = _clean_task_ref(operational.get("Active task", ""))
    if op_phase and op_phase != current_phase:
        errors.append(f"Dashboard active phase {op_phase!r} disagrees with YAML {current_phase!r}")
    if op_task and op_task != current_task:
        errors.append(f"Dashboard active task {op_task!r} disagrees with YAML {current_task!r}")

    handoff = _parse_handoff(text)
    if not handoff:
        errors.append("Current Session Handoff missing")
    else:
        handoff_phase = handoff.get("Active phase", "")
        handoff_task = _clean_task_ref(handoff.get("Active task", ""))
        handoff_status = _clean_task_ref(handoff.get("Task status at end", ""))
        if handoff_phase and handoff_phase != current_phase:
            errors.append(f"Session Handoff active phase {handoff_phase!r} disagrees with YAML {current_phase!r}")
        if handoff_task and handoff_task != current_task:
            errors.append(f"Session Handoff active task {handoff_task!r} disagrees with YAML {current_task!r}")
        # Status lines may contain prose; require the current checkpoint token to appear.
        if current_status and current_status not in handoff.get("Task status at end", ""):
            errors.append(
                f"Session Handoff task status {handoff.get('Task status at end', '')!r} "
                f"does not contain YAML status {current_status!r}"
            )
        for required in ("Exact next file and symbol", "Exact next command", "Exact expected result", "Tracker save verified"):
            value = handoff.get(required, "")
            if not value or value in {"—", "Pending", "NO"}:
                errors.append(f"Session Handoff missing required field: {required}")

    completed = [task for task in tasks if task.complete]
    last_done = front.get("last_done_verified_task")
    expected_last = completed[-1].task_id if completed else None
    if (last_done or None) != expected_last:
        errors.append(
            f"YAML last_done_verified_task={last_done!r} does not match actual last completed task {expected_last!r}"
        )

    gaps = _find_gaps(text)
    open_p0 = sum(g["severity"] == "P0" and g["status"] in {"OPEN", "IN_PROGRESS", "BLOCKED"} for g in gaps)
    open_p1 = sum(g["severity"] == "P1" and g["status"] in {"OPEN", "IN_PROGRESS", "BLOCKED"} for g in gaps)
    metrics.update({"gaps_total": len(gaps), "open_p0": open_p0, "open_p1": open_p1})
    for key, actual in (("open_p0_gaps", open_p0), ("open_p1_gaps", open_p1)):
        try:
            declared = int(front.get(key))
        except (TypeError, ValueError):
            errors.append(f"YAML {key} is not an integer: {front.get(key)!r}")
        else:
            if declared != actual:
                errors.append(f"YAML {key}={declared} but Gap Register has {actual}")

    # Safety release flags.
    if str(front.get("paper_only_required", "")).lower() not in {"true", "yes", "1"}:
        errors.append("paper_only_required must remain true")
    if str(front.get("current_release_mode", "")).upper() != "PAPER_ONLY":
        errors.append("current_release_mode must remain PAPER_ONLY during remediation")
    release_flag_gates = {
        "testnet_execution_approved": "Gate C",
        "live_shadow_approved": "Gate D",
        "limited_live_approved": "Gate E",
        "autonomous_execution_approved": "Gate F",
    }
    for flag, gate in release_flag_gates.items():
        if bool(front.get(flag)) and gates.get(gate) != "APPROVED":
            errors.append(
                f"unsafe release flag enabled without {gate} APPROVED: {flag}"
            )

    if check_evidence_hashes:
        errors.extend(_evidence_hash_errors(root, evidence_path))

    return ValidationResult(errors, warnings, metrics)


def _build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("runbook", nargs="?", help="Runbook path (defaults to canonical file at repository root)")
    parser.add_argument("--evidence", help="Evidence markdown to validate (defaults to evidence_package/EVIDENCE_LATEST.md)")
    parser.add_argument("--skip-evidence-hashes", action="store_true", help="Skip evidence hash checks while regenerating evidence")
    return parser


def main(argv: Iterable[str] | None = None) -> int:
    args = _build_parser().parse_args(list(argv) if argv is not None else None)
    root = Path(__file__).resolve().parents[1]
    path = Path(args.runbook) if args.runbook else root / "HERMES_V5_CLOUD_DESKTOP_MASTER_RUNBOOK.md"
    if not path.is_absolute():
        path = (Path.cwd() / path).resolve()
    try:
        text = path.read_text(encoding="utf-8")
    except (OSError, UnicodeError) as exc:
        print(f"ERROR: unable to read runbook as UTF-8: {exc}", file=sys.stderr)
        return 2

    evidence_path = Path(args.evidence).resolve() if args.evidence else None
    result = validate_text(
        text,
        root=path.parent,
        evidence_path=evidence_path,
        check_evidence_hashes=not args.skip_evidence_hashes,
    )
    for key, value in result.metrics.items():
        print(f"{key}={value}")
    for warning in result.warnings:
        print(f"WARNING: {warning}")
    for error in result.errors:
        print(f"ERROR: {error}")
    print("RESULT: CONSISTENT" if result.ok else "RESULT: INCONSISTENT")
    return 0 if result.ok else 1


if __name__ == "__main__":
    raise SystemExit(main())