Gitea-Tools/canonical_thread_handoff.py

"""Canonical Thread Handoff (CTH) protocol for issue and PR comments (#505).

A CTH comment is the authoritative workflow handoff in a Gitea issue or PR
thread. It records current state, decisions, blockers, proof, and the exact
next prompt/action for the next LLM or person.
"""

from __future__ import annotations

import re
from datetime import datetime, timezone
from typing import Any

MARKER = "<!-- cth:v1 -->"

CTH_TERM = "Canonical Thread Handoff"
CTH_ABBREV = "CTH"

CTH_TYPES = frozenset({
    "State Handoff",
    "Controller Decision",
    "Author Handoff",
    "Reviewer Handoff",
    "Merger Handoff",
    "Supersession Notice",
    "Blocker",
})

_REQUIRED_BASE_FIELDS = (
    "status",
    "next owner",
    "current blocker",
    "decision",
    "proof",
    "next action",
    "ready-to-paste prompt",
)

_HEADING_RE = re.compile(
    r"^##\s+CTH:\s*(.+?)\s*$",
    re.IGNORECASE | re.MULTILINE,
)
_FIELD_RE = re.compile(
    r"^([A-Za-z][A-Za-z0-9 /-]*):\s*(.+?)\s*$",
    re.MULTILINE,
)


def format_cth_body(
    *,
    cth_type: str,
    status: str,
    next_owner: str,
    current_blocker: str = "none",
    decision: str = "none",
    proof: str = "none",
    next_action: str = "none",
    ready_to_paste_prompt: str = "none",
    extra_fields: dict[str, str] | None = None,
) -> str:
    """Render a canonical CTH comment body."""
    normalized_type = (cth_type or "").strip()
    if normalized_type not in CTH_TYPES:
        raise ValueError(
            f"unknown CTH type '{cth_type}'; expected one of {sorted(CTH_TYPES)}"
        )
    lines = [
        MARKER,
        f"## CTH: {normalized_type}",
        "",
        f"Status: {(status or 'unknown').strip() or 'unknown'}",
        f"Next owner: {(next_owner or 'unknown').strip() or 'unknown'}",
        f"Current blocker: {(current_blocker or 'none').strip() or 'none'}",
        f"Decision: {(decision or 'none').strip() or 'none'}",
        f"Proof: {(proof or 'none').strip() or 'none'}",
        f"Next action: {(next_action or 'none').strip() or 'none'}",
        f"Ready-to-paste prompt: {(ready_to_paste_prompt or 'none').strip() or 'none'}",
    ]
    for key, value in (extra_fields or {}).items():
        label = (key or "").strip()
        if not label:
            continue
        lines.append(f"{label}: {(value or 'none').strip() or 'none'}")
    return "\n".join(lines)


def parse_cth_comment(body: str) -> dict[str, Any] | None:
    """Parse one CTH comment body, or None when not a CTH comment."""
    text = body or ""
    if MARKER not in text and not _HEADING_RE.search(text):
        return None
    heading = _HEADING_RE.search(text)
    if not heading:
        return None
    cth_type = heading.group(1).strip()
    fields: dict[str, str] = {}
    for match in _FIELD_RE.finditer(text):
        key = match.group(1).strip().lower()
        if key.startswith("cth"):
            continue
        fields[key] = match.group(2).strip()
    return {
        "cth_type": cth_type,
        "fields": fields,
        "raw_body": text,
    }


def assess_cth_comment(body: str) -> dict[str, Any]:
    """Validate a CTH comment has required base fields and a known type."""
    parsed = parse_cth_comment(body)
    reasons: list[str] = []
    if not parsed:
        return {
            "valid": False,
            "block": True,
            "reasons": ["comment is not a Canonical Thread Handoff (CTH)"],
            "parsed": None,
        }

    cth_type = parsed.get("cth_type") or ""
    if cth_type not in CTH_TYPES:
        reasons.append(
            f"unknown CTH type '{cth_type}'; expected one of {sorted(CTH_TYPES)}"
        )

    fields = parsed.get("fields") or {}
    missing = [name for name in _REQUIRED_BASE_FIELDS if not (fields.get(name) or "").strip()]
    if missing:
        reasons.append(
            "CTH missing required fields: " + ", ".join(missing)
        )

    vague_prompt = (fields.get("ready-to-paste prompt") or "").strip().lower()
    if vague_prompt in {"", "none", "tbd", "n/a", "todo"}:
        reasons.append("ready-to-paste prompt must be concrete and paste-ready")

    return {
        "valid": not reasons,
        "block": bool(reasons),
        "reasons": reasons,
        "parsed": parsed,
        "cth_type": cth_type,
    }


def find_latest_cth(comments: list[dict[str, Any]]) -> dict[str, Any] | None:
    """Return the newest valid CTH comment from a Gitea comment list."""
    latest: dict[str, Any] | None = None
    latest_ts: datetime | None = None
    for comment in comments or []:
        body = comment.get("body") or ""
        parsed = parse_cth_comment(body)
        if not parsed:
            continue
        created = _parse_timestamp(comment.get("created_at"))
        if latest is None or (created and (latest_ts is None or created >= latest_ts)):
            latest = {
                "comment_id": comment.get("id"),
                "created_at": comment.get("created_at"),
                "author": (comment.get("user") or {}).get("login"),
                **parsed,
            }
            latest_ts = created
    return latest


def assess_cth_supersedes_non_cth(
    *,
    latest_cth: dict[str, Any] | None,
    narrative_claim: str,
) -> dict[str, Any]:
    """Fail closed when narrative relies on stale non-CTH state despite newer CTH."""
    if not latest_cth:
        return {"block": False, "reasons": []}
    text = (narrative_claim or "").strip()
    if not text:
        return {"block": False, "reasons": []}
    if parse_cth_comment(text):
        return {"block": False, "reasons": []}
    return {
        "block": True,
        "reasons": [
            "workflow narrative must treat the latest CTH comment as authoritative; "
            f"found newer CTH type '{latest_cth.get('cth_type')}' "
            f"(comment #{latest_cth.get('comment_id')})"
        ],
    }


def _parse_timestamp(value: str | None) -> datetime | None:
    if not value:
        return None
    text = value.strip()
    if text.endswith("Z"):
        text = text[:-1] + "+00:00"
    try:
        parsed = datetime.fromisoformat(text)
    except ValueError:
        return None
    if parsed.tzinfo is None:
        return parsed.replace(tzinfo=timezone.utc)
    return parsed.astimezone(timezone.utc)


EXAMPLE_SCENARIOS = (
    "pr_approved_ready_for_merger",
    "pr_request_changes_to_author",
    "duplicate_superseded_pr_closure",
    "blocked_dirty_root_worktree",
    "stale_head_requires_fresh_review",
    "issue_implementation_handoff",
)