feat: require canonical next-action state comments (#495)

Add canonical issue/PR/discussion state comment templates and validators,
final-report rules for STATE/WHO_IS_NEXT/NEXT_ACTION/NEXT_PROMPT, and
queue-controller guidance. Posting enforcement remains in #496.

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>

canonical_state_comments.py

@@ -0,0 +1,171 @@
+"""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"],
+    }