Gitea-Tools/canonical_state_comments.py

"""Canonical state comment validation helpers (#495).

These helpers validate durable issue/PR/discussion state handoff comments.
They are intentionally pure and do not post comments; MCP mutation hooks are
owned by #496.
"""

from __future__ import annotations

import re

CANONICAL_HEADINGS = (
    "canonical issue state",
    "canonical pr state",
    "canonical discussion summary",
)

REQUIRED_FIELDS = ("STATE", "WHO_IS_NEXT", "NEXT_ACTION", "NEXT_PROMPT")
ALLOWED_NEXT_ACTORS = {
    "controller",
    "author",
    "reviewer",
    "merger",
    "reconciler",
    "user",
}

VAGUE_NEXT_ACTIONS = {
    "continue",
    "handle this",
    "do it",
    "fix it",
    "proceed",
    "follow up",
    "next",
    "tbd",
    "todo",
    "n/a",
    "none",
}

_FIELD_RE = re.compile(r"^\s*(?:[-*]\s*)?([A-Z][A-Z0-9_ ]+)\s*:\s*(.*)$")
_FULL_SHA_RE = re.compile(r"\b[0-9a-f]{40}\b", re.IGNORECASE)
_CLAIMS_STATE_UPDATE_RE = re.compile(
    r"canonical\s+(?:issue|pr|discussion)?\s*state|"
    r"state\s+comment\s+(?:posted|created|updated)|"
    r"next[- ]action\s+comment\s+(?:posted|created|updated)",
    re.IGNORECASE,
)


def extract_state_fields(text: str | None) -> dict[str, str]:
    """Return upper-case labeled fields from a canonical state block."""
    fields: dict[str, str] = {}
    current_key: str | None = None
    for line in (text or "").splitlines():
        match = _FIELD_RE.match(line)
        if match:
            current_key = match.group(1).strip().upper().replace(" ", "_")
            fields[current_key] = match.group(2).strip()
            continue
        stripped = line.strip()
        if current_key and stripped and not stripped.startswith("#"):
            existing = fields.get(current_key, "")
            fields[current_key] = (
                f"{existing}\n{stripped}" if existing else stripped
            )
    return fields


def contains_canonical_state_block(text: str | None) -> bool:
    lower = (text or "").lower()
    return any(heading in lower for heading in CANONICAL_HEADINGS)


def claims_canonical_state_update(text: str | None) -> bool:
    """Return True when text claims a canonical state/next-action update."""
    return bool(_CLAIMS_STATE_UPDATE_RE.search(text or ""))


def _empty_or_placeholder(value: str | None) -> bool:
    value = (value or "").strip().lower()
    return not value or value in {"none", "n/a", "unknown", "tbd", "<...>"}


def _vague_next_action(value: str | None) -> bool:
    normalized = re.sub(r"\s+", " ", (value or "").strip().lower())
    return normalized in VAGUE_NEXT_ACTIONS


def validate_canonical_state_comment(text: str | None) -> dict:
    """Validate a canonical state comment or embedded final-report block.

    Returns a dict with ``valid`` and ``reasons``. The validator focuses on
    fields that make continuation possible: current state, next actor, next
    action, and paste-ready next prompt, plus a few contradiction checks.
    """
    fields = extract_state_fields(text)
    reasons: list[str] = []

    for field in REQUIRED_FIELDS:
        if _empty_or_placeholder(fields.get(field)):
            reasons.append(f"missing required canonical state field: {field}")

    actor = (fields.get("WHO_IS_NEXT") or "").strip().lower()
    if actor and actor not in ALLOWED_NEXT_ACTORS:
        reasons.append(
            "WHO_IS_NEXT must be one of: "
            + ", ".join(sorted(ALLOWED_NEXT_ACTORS))
        )

    if _vague_next_action(fields.get("NEXT_ACTION")):
        reasons.append("NEXT_ACTION is too vague for durable continuation")

    state = (fields.get("STATE") or "").strip().lower().replace("-", "_")
    if "ready_to_merge" in state or (
        state == "approved" and "pr" in (text or "").lower()
    ):
        review_status = (fields.get("REVIEW_STATUS") or "").lower()
        head_sha = fields.get("HEAD_SHA") or ""
        merge_ready = (fields.get("MERGE_READY") or "").lower()
        if "approved" not in review_status:
            reasons.append("ready-to-merge state requires approved REVIEW_STATUS")
        if not _FULL_SHA_RE.search(head_sha):
            reasons.append("ready-to-merge state requires full HEAD_SHA proof")
        if merge_ready and not merge_ready.startswith(("yes", "true")):
            reasons.append("ready-to-merge state contradicts MERGE_READY")

    if "superseded" in state:
        canonical = fields.get("CANONICAL_ITEM") or fields.get("SUPERSEDED_BY")
        if _empty_or_placeholder(canonical):
            reasons.append("superseded state requires canonical item proof")

    if "blocked" in state:
        blockers = fields.get("BLOCKERS") or ""
        if _empty_or_placeholder(blockers):
            reasons.append("blocked state requires BLOCKERS/unblock condition")

    return {
        "valid": not reasons,
        "fields": fields,
        "reasons": reasons,
    }


def validate_final_report_state_update(report_text: str | None) -> dict:
    """Validate canonical state-update claims inside a final report."""
    text = report_text or ""
    if not claims_canonical_state_update(text) and not contains_canonical_state_block(text):
        return {
            "applicable": False,
            "valid": True,
            "reasons": [],
        }

    if not contains_canonical_state_block(text):
        return {
            "applicable": True,
            "valid": False,
            "reasons": [
                "final report claims a canonical state update but includes no canonical state block"
            ],
        }

    result = validate_canonical_state_comment(text)
    return {
        "applicable": True,
        "valid": result["valid"],
        "fields": result["fields"],
        "reasons": result["reasons"],
    }