canonical_thread_handoff.py

@@ -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",
+)

docs/canonical-thread-handoff.md

@@ -0,0 +1,142 @@
+# Canonical Thread Handoff (CTH)
+
+**CTH** = **Canonical Thread Handoff**
+
+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.
+
+CTH comments complement — but do not replace — formal Gitea review state.
+A CTH may summarize an APPROVE or REQUEST_CHANGES decision, yet merge gates
+still require the live Gitea review verdict.
+
+## CTH comment types
+
+- `CTH: State Handoff`
+- `CTH: Controller Decision`
+- `CTH: Author Handoff`
+- `CTH: Reviewer Handoff`
+- `CTH: Merger Handoff`
+- `CTH: Supersession Notice`
+- `CTH: Blocker`
+
+## Required base template
+
+```md
+## CTH: <Type>
+
+Status:
+Next owner:
+Current blocker:
+Decision:
+Proof:
+Next action:
+Ready-to-paste prompt:
+```
+
+Extended fields may map to canonical issue/PR state templates from #495 when
+that layer is available. Until then, keep the base fields complete.
+
+## Discovery and posting rules
+
+Before acting in an issue or PR thread:
+
+1. **Find the latest CTH comment** before doing work.
+2. Treat the **latest valid CTH** as the current handoff state.
+3. **Post a new CTH** when you finish, block, skip, supersede, request
+   changes, approve, or hand off.
+4. Do **not** rely on stale non-CTH comments when a newer CTH exists.
+5. Casual discussion comments do not need to be CTH comments.
+
+## Examples
+
+### PR approved and ready for merger
+
+```md
+## CTH: Reviewer Handoff
+
+Status: approved_at_current_head
+Next owner: merger
+Current blocker: none
+Decision: APPROVE recorded at head abc123...
+Proof: gitea_submit_pr_review performed; visible verdict APPROVE
+Next action: eligible merger merges PR #N with pinned expected_head_sha
+Ready-to-paste prompt: Merge PR #N for issue #M if live head still abc123... and merge gates pass.
+```
+
+### PR request-changes back to author
+
+```md
+## CTH: Reviewer Handoff
+
+Status: request_changes_at_current_head
+Next owner: author
+Current blocker: unresolved findings in validation report
+Decision: REQUEST_CHANGES at head def456...
+Proof: gitea_submit_pr_review performed; blocking review visible
+Next action: author fixes findings and pushes; reviewer re-validates fresh head
+Ready-to-paste prompt: Fix PR #N review findings, push to feat/issue-M-..., post Author Handoff CTH.
+```
+
+### Duplicate / superseded PR closure
+
+```md
+## CTH: Supersession Notice
+
+Status: superseded
+Next owner: controller
+Current blocker: duplicate branch/PR work
+Decision: close PR #N; continue on PR #M
+Proof: duplicate gate linked open PR #M for issue #K
+Next action: controller closes superseded PR and records canonical state
+Ready-to-paste prompt: Close superseded PR #N; confirm PR #M remains canonical for issue #K.
+```
+
+### Blocked workflow due to dirty root/worktree
+
+```md
+## CTH: Blocker
+
+Status: blocked_preflight
+Next owner: operator
+Current blocker: dirty control checkout / branches worktree mismatch
+Decision: stop before mutation
+Proof: verify_preflight_purity failed; workspace diagnostics attached
+Next action: repair worktree or relaunch MCP from branches/ worktree
+Ready-to-paste prompt: Repair dirty workspace, relaunch MCP from branches/review-prN, retry review.
+```
+
+### Stale head requiring fresh review
+
+```md
+## CTH: Reviewer Handoff
+
+Status: stale_head
+Next owner: reviewer
+Current blocker: live head differs from pinned review head
+Decision: prior approval not valid for current head
+Proof: expected_head_sha mismatch at merge gate
+Next action: re-run validation and post fresh review decision
+Ready-to-paste prompt: Re-validate PR #N at live head, dry-run review gates, submit fresh verdict.
+```
+
+### Issue implementation handoff
+
+```md
+## CTH: Author Handoff
+
+Status: implementation_complete_pending_review
+Next owner: reviewer
+Current blocker: none
+Decision: PR #N ready for review at head fedcba...
+Proof: tests passed in branches/issue-M-...; PR opened with Closes #M
+Next action: reviewer acquires lease and validates PR #N
+Ready-to-paste prompt: Review PR #N for issue #M; pin head fedcba... before mutations.
+```
+
+## Related
+
+- [`llm-workflow-runbooks.md`](llm-workflow-runbooks.md)
+- [`../skills/llm-project-workflow/templates/canonical-thread-handoff.md`](../skills/llm-project-workflow/templates/canonical-thread-handoff.md)
+- #495 — canonical next-action comment fields
+- #496 — fail-closed validation for workflow-changing comments

docs/llm-workflow-runbooks.md

@@ -786,6 +786,27 @@ Never imply full-suite success unless the full-suite command itself passed
 (`full_suite_passed: true`). A report that hides a failed or skipped check
 is worse than a failing report.
 
+## Canonical Thread Handoff (CTH)
+
+**CTH** = **Canonical Thread Handoff** — the authoritative workflow handoff
+comment in a Gitea issue or PR thread. See
+[`canonical-thread-handoff.md`](canonical-thread-handoff.md) for types,
+templates, and examples.
+
+Before acting in an issue or PR thread:
+
+1. **Find the latest CTH comment** before doing work.
+2. Treat the **latest valid CTH** as the current handoff state.
+3. **Post a new CTH** when you finish, block, skip, supersede, request
+   changes, approve, or hand off.
+4. Do **not** rely on stale non-CTH comments when a newer CTH exists.
+
+A CTH summarizes workflow state for the next session. Formal Gitea review
+verdicts remain authoritative for merge gates — a CTH is not merge approval
+by itself.
+
+Template: [`../skills/llm-project-workflow/templates/canonical-thread-handoff.md`](../skills/llm-project-workflow/templates/canonical-thread-handoff.md)
+
 ## Controller Handoff (required, every task)
 
 Every task — implementation, review, merge, triage, documentation,

skills/llm-project-workflow/templates/canonical-thread-handoff.md

@@ -0,0 +1,36 @@
+# Template: Canonical Thread Handoff (CTH)
+
+Copy, fill the fields, and post as an issue or PR comment.
+
+```text
+## CTH: <Type>
+
+Status: <current workflow state>
+Next owner: <author|reviewer|merger|controller|operator>
+Current blocker: <none or exact blocker>
+Decision: <what was decided this session>
+Proof: <commands, SHAs, gate outputs, or explicit pending proof>
+Next action: <one concrete step for the next actor>
+Ready-to-paste prompt: <full prompt the next session should paste>
+```
+
+Allowed `<Type>` values:
+
+- State Handoff
+- Controller Decision
+- Author Handoff
+- Reviewer Handoff
+- Merger Handoff
+- Supersession Notice
+- Blocker
+
+Rules:
+
+- Find the latest CTH comment in the thread before starting work.
+- Treat the latest valid CTH as the current source of truth.
+- Post a new CTH when you finish, block, skip, supersede, approve, request
+  changes, or hand off.
+- A CTH summarizes workflow state; formal Gitea review verdicts remain
+  authoritative for merge gates.
+
+Full protocol: `docs/canonical-thread-handoff.md`

skills/llm-project-workflow/templates/merge-pr.md

@@ -10,6 +10,9 @@ Load the canonical workflow first:
 Final report schema: `schemas/review-merge-final-report.md`.
 
 Rules (llm-project-workflow):
+- Find the latest CTH comment on the PR/issue thread before starting work.
+  Post a new CTH: Merger Handoff (or CTH: Blocker) at session end.
+  Template: skills/llm-project-workflow/templates/canonical-thread-handoff.md
 - Repository targeting (#530): pass explicit `remote=`, `org=`, and `repo=` on
   every gitea-tools call (e.g. `remote=prgs org=Scaled-Tech-Consulting
   repo=Gitea-Tools`). A bare `remote=prgs` can resolve to the wrong default repo

skills/llm-project-workflow/templates/review-pr.md

@@ -35,6 +35,9 @@ Load the canonical workflow first:
 Final report schema: `schemas/review-merge-final-report.md`.
 
 Rules (llm-project-workflow):
+- Find the latest CTH comment on the PR/issue thread before starting work.
+  Post a new CTH: Reviewer Handoff (or CTH: Blocker) at session end.
+  Template: skills/llm-project-workflow/templates/canonical-thread-handoff.md
 - Repository targeting (#530): pass explicit `remote=`, `org=`, and `repo=` on
   every gitea-tools call (e.g. `remote=prgs org=Scaled-Tech-Consulting
   repo=Gitea-Tools`). A bare `remote=prgs` can resolve to the wrong default repo

skills/llm-project-workflow/templates/start-issue.md

@@ -10,6 +10,9 @@ Final report schema: skills/llm-project-workflow/schemas/work-issue-final-report
 Router: skills/llm-project-workflow/SKILL.md (task mode: work-issue)
 
 Rules (llm-project-workflow):
+- Find the latest CTH comment on the issue/PR thread before starting work.
+  Post a new CTH: Author Handoff (or CTH: Blocker) at session end.
+  Template: skills/llm-project-workflow/templates/canonical-thread-handoff.md
 - No repo changes without a tracking issue. If none exists, create one first;
   if it can't be created, stop.
 - Work only in an isolated branch worktree under branches/. The main checkout

skills/llm-project-workflow/workflows/review-merge-pr.md

@@ -26,6 +26,12 @@ workflow rules exactly.
 
 ## 0. Load the canonical workflow first
 
+Before starting PR work, **find the latest CTH comment** on the PR or linked
+issue thread. Treat that CTH as the current handoff state. Post a new
+**CTH: Reviewer Handoff**, **CTH: Merger Handoff**, **CTH: Blocker**, or
+**CTH: Supersession Notice** when you finish, block, skip, supersede,
+approve, request changes, or hand off. See `docs/canonical-thread-handoff.md`.
+
 Before starting PR work, check whether the project provides a canonical PR review/merge workflow through a project skill, runbook, or MCP helper.
 
 If available, load it first and report:

skills/llm-project-workflow/workflows/work-issue.md

@@ -26,6 +26,12 @@ This is an author/coder workflow. It is not a reviewer workflow.
 
 ## 0. Load the canonical workflow first
 
+Before starting issue work, **find the latest CTH comment** on the target
+issue or linked PR thread. Treat that CTH as the current handoff state unless
+a newer authoritative gate supersedes it. Post a new **CTH: Author Handoff**
+(or `CTH: Blocker`) when you finish, block, or hand off.
+See `docs/canonical-thread-handoff.md`.
+
 Before starting issue work, check whether the project provides a canonical work-on-issue workflow through a project skill, runbook, or MCP helper.
 
 If available, load it first and report:

tests/test_canonical_thread_handoff.py

@@ -0,0 +1,139 @@
+"""Tests for Canonical Thread Handoff (CTH) protocol (#505)."""
+
+from __future__ import annotations
+
+import sys
+import unittest
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+import canonical_thread_handoff as cth  # noqa: E402
+
+REPO_ROOT = Path(__file__).resolve().parent.parent
+RUNBOOK = REPO_ROOT / "docs" / "llm-workflow-runbooks.md"
+PROTOCOL_DOC = REPO_ROOT / "docs" / "canonical-thread-handoff.md"
+TEMPLATE = REPO_ROOT / "skills" / "llm-project-workflow/templates" / "canonical-thread-handoff.md"
+REVIEW_WORKFLOW = REPO_ROOT / "skills" / "llm-project-workflow/workflows" / "review-merge-pr.md"
+WORK_WORKFLOW = REPO_ROOT / "skills" / "llm-project-workflow/workflows" / "work-issue.md"
+REVIEW_TEMPLATE = REPO_ROOT / "skills" / "llm-project-workflow/templates" / "review-pr.md"
+START_TEMPLATE = REPO_ROOT / "skills" / "llm-project-workflow/templates" / "start-issue.md"
+MERGE_TEMPLATE = REPO_ROOT / "skills" / "llm-project-workflow/templates" / "merge-pr.md"
+
+
+class TestCthModule(unittest.TestCase):
+    def test_defines_cth_term_and_types(self):
+        self.assertEqual(cth.CTH_ABBREV, "CTH")
+        self.assertIn("State Handoff", cth.CTH_TYPES)
+        self.assertIn("Reviewer Handoff", cth.CTH_TYPES)
+        self.assertIn("Blocker", cth.CTH_TYPES)
+
+    def test_format_and_parse_round_trip(self):
+        body = cth.format_cth_body(
+            cth_type="Author Handoff",
+            status="implementation_complete",
+            next_owner="reviewer",
+            current_blocker="none",
+            decision="PR opened",
+            proof="pytest passed",
+            next_action="review PR #9",
+            ready_to_paste_prompt="Review PR #9 for issue #5",
+        )
+        parsed = cth.parse_cth_comment(body)
+        self.assertIsNotNone(parsed)
+        self.assertEqual(parsed["cth_type"], "Author Handoff")
+        self.assertEqual(parsed["fields"]["status"], "implementation_complete")
+        self.assertEqual(parsed["fields"]["next owner"], "reviewer")
+
+    def test_valid_cth_passes_assessment(self):
+        body = cth.format_cth_body(
+            cth_type="Reviewer Handoff",
+            status="approved",
+            next_owner="merger",
+            current_blocker="none",
+            decision="APPROVE",
+            proof="review id 42",
+            next_action="merge if gates pass",
+            ready_to_paste_prompt="Merge PR #12 with pinned head abcdef0123456789abcdef0123456789abcdef01",
+        )
+        result = cth.assess_cth_comment(body)
+        self.assertFalse(result["block"])
+        self.assertTrue(result["valid"])
+
+    def test_missing_fields_fail_closed(self):
+        body = "## CTH: Blocker\n\nStatus: blocked\n"
+        result = cth.assess_cth_comment(body)
+        self.assertTrue(result["block"])
+        self.assertIn("missing required fields", result["reasons"][0])
+
+    def test_find_latest_cth_prefers_newer_comment(self):
+        older = cth.format_cth_body(
+            cth_type="State Handoff",
+            status="old",
+            next_owner="author",
+            current_blocker="none",
+            decision="old",
+            proof="old",
+            next_action="old",
+            ready_to_paste_prompt="old prompt text here",
+        )
+        newer = cth.format_cth_body(
+            cth_type="Reviewer Handoff",
+            status="new",
+            next_owner="merger",
+            current_blocker="none",
+            decision="approve",
+            proof="new",
+            next_action="merge",
+            ready_to_paste_prompt="merge PR #3 now with pinned head",
+        )
+        comments = [
+            {"id": 1, "created_at": "2026-07-01T10:00:00Z", "body": older},
+            {"id": 2, "created_at": "2026-07-02T10:00:00Z", "body": newer},
+        ]
+        latest = cth.find_latest_cth(comments)
+        self.assertEqual(latest["comment_id"], 2)
+        self.assertEqual(latest["cth_type"], "Reviewer Handoff")
+
+    def test_cth_does_not_replace_formal_reviews_documented_in_protocol(self):
+        text = PROTOCOL_DOC.read_text(encoding="utf-8")
+        self.assertIn("do not replace", text.lower())
+        self.assertIn("formal Gitea review", text)
+
+    def test_examples_cover_required_scenarios(self):
+        text = PROTOCOL_DOC.read_text(encoding="utf-8")
+        expected_phrases = (
+            "PR approved and ready for merger",
+            "request-changes back to author",
+            "superseded PR closure",
+            "dirty root/worktree",
+            "Stale head requiring fresh review",
+            "Issue implementation handoff",
+        )
+        for phrase in expected_phrases:
+            self.assertIn(phrase, text)
+
+
+class TestCthDocumentationChecks(unittest.TestCase):
+    def test_runbook_references_cth_discovery_and_posting(self):
+        text = RUNBOOK.read_text(encoding="utf-8")
+        self.assertIn("Canonical Thread Handoff", text)
+        self.assertIn("Find the latest CTH comment", text)
+        self.assertIn("Post a new CTH", text)
+        self.assertIn("canonical-thread-handoff.md", text)
+
+    def test_workflows_reference_cth_rules(self):
+        for path in (REVIEW_WORKFLOW, WORK_WORKFLOW):
+            text = path.read_text(encoding="utf-8")
+            self.assertIn("latest CTH comment", text)
+            self.assertIn("canonical-thread-handoff.md", text)
+
+    def test_prompt_templates_reference_cth_rules(self):
+        for path in (REVIEW_TEMPLATE, START_TEMPLATE, MERGE_TEMPLATE, TEMPLATE):
+            text = path.read_text(encoding="utf-8")
+            self.assertIn("CTH", text)
+            self.assertIn("canonical-thread-handoff", text)
+
+
+if __name__ == "__main__":
+    unittest.main()