feat: add Canonical Thread Handoff protocol (Closes #505)
Define CTH comment types, base template, parser/validator helpers, protocol documentation with examples, and workflow/runbook/prompt updates so agents discover and post authoritative thread handoffs before acting. Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
This commit is contained in:
Claude Opus 4.8
@@ -0,0 +1,213 @@
|
||||
"""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",
|
||||
)
|
||||
Reference in New Issue
Block a user