docs/llm-workflow-runbooks.md

@@ -939,6 +939,21 @@ When posting a Canonical Thread Handoff after a binding blocker:
 - [`credential-isolation.md`](credential-isolation.md) — credential handling.
 - [`release-workflows.md`](release-workflows.md) — release/merge workflow.
 - [`../README.md`](../README.md) — canonical config, thin launchers, the menu.
+- [`state-handoff-ledger.md`](state-handoff-ledger.md) — canonical state comments and next-action handoff (#494).
+
+## Canonical state handoff ledger (#494)
+
+Gitea comments and final reports must make continuation obvious without chat
+history. See [`state-handoff-ledger.md`](state-handoff-ledger.md) for:
+
+- discussion → issue → PR → review → merge → reconcile lifecycle
+- templates for discussion, issue, PR, and queue-controller state comments
+- final-report requirements: Current status, Next actor, Next action, Next prompt
+- queue-controller priority order and discussion ≥5-comment rule (urgent/trivial
+  exceptions)
+
+Helpers live in `state_handoff_ledger.py`; final-report enforcement is wired
+through `assess_final_report_validator`.
 
 ## PR-only queue cleanup mode (#390)
 

docs/state-handoff-ledger.md

@@ -0,0 +1,86 @@
+# Canonical state handoff ledger (#494)
+
+Gitea is the system of record for continuation. Every discussion, issue, and PR
+should answer: current state, last proof, who acts next, and the exact prompt for
+the next role.
+
+## Lifecycle
+
+1. Controller opens a discussion.
+2. Discussion accumulates substantive comments (default minimum: five).
+3. Controller posts a discussion summary when ready.
+4. Controller creates linked issues from the summary.
+5. Author locks issue, implements, opens PR.
+6. Reviewer reviews at current head.
+7. Merger merges after formal approval.
+8. Reconciler closes superseded or already-landed PRs.
+9. Issue/PR/discussion state comments make the next action obvious at every step.
+
+## Discussion rules
+
+- Do **not** convert a discussion into issues until it has at least **five
+  substantive comments**, unless the discussion state comment marks
+  `URGENCY: urgent` or `URGENCY: trivial`.
+- Substantive comment types: proposal, risk/concern, acceptance criteria,
+  implementation approach, dependency/sequence, summary.
+- Before issue creation, post a **discussion summary** with decision, issues to
+  create, non-goals, unresolved questions, and next prompt.
+- Created issues must link back to the discussion; the discussion must link
+  forward to created issues.
+
+## State comment templates
+
+Use `state_handoff_ledger.py` helpers or copy the canonical blocks:
+
+- `render_discussion_state_comment(...)`
+- `render_discussion_summary_comment(...)`
+- `render_issue_state_comment(...)`
+- `render_pr_state_comment(...)`
+- `render_queue_controller_report(...)`
+
+Post state comments as the **latest canonical update** on the object. Do not
+bury state inside PR bodies only.
+
+## Final report requirements
+
+Every final report must include a `Controller Handoff` section with:
+
+- **Current status** — live state after this session
+- **Next actor** — `author`, `reviewer`, `merger`, `reconciler`, or `controller`
+- **Next action** — one imperative step
+- **Next prompt** — ready-to-paste prompt for the next role
+
+`assess_final_report_next_action_handoff` and
+`assess_contradictory_state_handoff` enforce these fields and reject
+contradictory claims (for example, ready-to-merge without approval, issue done
+without PR proof, discussion complete without summary).
+
+## Queue controller selection (priority order)
+
+1. Merge clean approved PRs at current head.
+2. Review PRs needing review.
+3. Reconcile superseded or already-landed duplicates.
+4. Continue blocked-but-now-unblocked issues.
+5. Create issues from completed discussions.
+6. Start new author work only when higher-priority queue items are clear.
+
+For each candidate object, read the **latest canonical state comment** before
+choosing an action. Output the exact next-role prompt in the controller report.
+
+## Example workflow states
+
+| State | Next actor | Typical next action |
+|-------|------------|---------------------|
+| Discussion needs more comments | controller | Facilitate discussion until five substantive comments or urgent/trivial exception |
+| Issue ready for author | author | Lock issue and implement in `branches/` worktree |
+| PR needs review | reviewer | Review at pinned head in reviewer worktree |
+| PR approved at head | merger | Merge with merger profile after eligibility proof |
+| PR superseded | reconciler | Close duplicate/already-landed PR with proof |
+| Issue blocked | controller | Post issue state comment with blockers and next prompt |
+
+## Non-goals
+
+- Chat history is not the source of truth.
+- Do not weaken author/reviewer/merger/reconciler separation.
+- Do not replace Gitea issues, PRs, or canonical workflow files under
+  `skills/llm-project-workflow/`.

final_report_validator.py

@@ -24,6 +24,7 @@ from review_proofs import (
     assess_review_mutation_final_report,
     assess_validation_report,
 )
+from state_handoff_ledger import assess_state_handoff_ledger_report
 from validation_status_vocabulary import assess_validation_status_vocabulary
 
 FINAL_REPORT_TASK_KINDS = frozenset({
@@ -335,6 +336,38 @@ def _rule_shared_controller_handoff(
     )
 
 
+def _rule_shared_state_handoff_next_action(report_text: str) -> list[dict[str, str]]:
+    result = assess_state_handoff_ledger_report(report_text)
+    if result.get("complete"):
+        return []
+    findings: list[dict[str, str]] = []
+    next_action = result.get("next_action") or {}
+    contradictions = result.get("contradictions") or {}
+    if next_action.get("block"):
+        findings.extend(
+            _findings_from_reasons(
+                "shared.state_handoff_next_action",
+                next_action.get("reasons") or [],
+                field="Next action handoff",
+                severity="block",
+                safe_next_action=next_action.get("safe_next_action")
+                or "add next-action handoff fields to Controller Handoff",
+            )
+        )
+    if contradictions.get("block"):
+        findings.extend(
+            _findings_from_reasons(
+                "shared.state_handoff_contradiction",
+                contradictions.get("reasons") or [],
+                field="State handoff",
+                severity="block",
+                safe_next_action=contradictions.get("safe_next_action")
+                or "resolve contradictory state claims before submission",
+            )
+        )
+    return findings
+
+
 def _rule_shared_email_disclosure(report_text: str) -> list[dict[str, str]]:
     result = assess_email_disclosure(report_text)
     if result.get("proven"):
@@ -1282,6 +1315,7 @@ _SHARED_CLEANUP_PROOF_RULES = (
 _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = {
     "review_pr": [
         _rule_shared_controller_handoff,
+        _rule_shared_state_handoff_next_action,
         _rule_shared_email_disclosure,
         *_SHARED_ISSUE_LOCK_RULES,
         _rule_reviewer_legacy_workspace_mutations,
@@ -1310,6 +1344,7 @@ _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = {
     ],
     "reconcile_already_landed": [
         _rule_reconcile_controller_handoff,
+        _rule_shared_state_handoff_next_action,
         _rule_shared_email_disclosure,
         *_SHARED_ISSUE_LOCK_RULES,
         *_SHARED_CLEANUP_PROOF_RULES,
@@ -1325,12 +1360,14 @@ _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = {
     ],
     "author_issue": [
         _rule_shared_controller_handoff,
+        _rule_shared_state_handoff_next_action,
         _rule_shared_email_disclosure,
         *_SHARED_ISSUE_LOCK_RULES,
         _rule_reviewer_vague_mutations_none,
     ],
     "work_issue": [
         _rule_shared_controller_handoff,
+        _rule_shared_state_handoff_next_action,
         _rule_shared_email_disclosure,
         *_SHARED_ISSUE_LOCK_RULES,
         _rule_shared_issue_acceptance_gate,
@@ -1340,17 +1377,20 @@ _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = {
     ],
     "issue_filing": [
         _rule_shared_controller_handoff,
+        _rule_shared_state_handoff_next_action,
         _rule_shared_email_disclosure,
         *_SHARED_ISSUE_LOCK_RULES,
     ],
     "inventory": [
         _rule_shared_controller_handoff,
+        _rule_shared_state_handoff_next_action,
         _rule_shared_email_disclosure,
         *_SHARED_ISSUE_LOCK_RULES,
         _rule_reconcile_pagination_proof,
     ],
     "issue_selection": [
         _rule_shared_controller_handoff,
+        _rule_shared_state_handoff_next_action,
         _rule_shared_email_disclosure,
         *_SHARED_ISSUE_LOCK_RULES,
     ],

state_handoff_ledger.py

@@ -0,0 +1,374 @@
+"""Canonical state handoff ledger for discussions, issues, and PRs (#494)."""
+
+from __future__ import annotations
+
+import re
+from typing import Any
+
+HANDOFF_HEADING = "Controller Handoff"
+
+ISSUE_STATE_FIELDS = (
+    "STATE",
+    "NEXT_ACTOR",
+    "NEXT_ACTION",
+    "NEXT_PROMPT",
+    "STATUS",
+    "RELATED_PRS",
+    "BRANCH",
+    "HEAD_SHA",
+    "VALIDATION",
+    "BLOCKERS",
+    "DUPLICATES_OR_SUPERSEDES",
+    "LAST_UPDATED_BY",
+)
+
+PR_STATE_FIELDS = (
+    "STATE",
+    "NEXT_ACTOR",
+    "NEXT_ACTION",
+    "NEXT_PROMPT",
+    "ISSUE",
+    "BASE",
+    "HEAD",
+    "HEAD_SHA",
+    "REVIEW_STATUS",
+    "VALIDATION",
+    "BLOCKERS",
+    "SUPERSEDES",
+    "SUPERSEDED_BY",
+    "MERGE_READY",
+    "LAST_UPDATED_BY",
+)
+
+DISCUSSION_STATE_FIELDS = (
+    "STATE",
+    "NEXT_ACTOR",
+    "NEXT_ACTION",
+    "NEXT_PROMPT",
+    "COMMENT_COUNT",
+    "SUBSTANTIVE_COMMENTS",
+    "URGENCY",
+    "BLOCKERS",
+    "LAST_UPDATED_BY",
+)
+
+DISCUSSION_SUMMARY_FIELDS = (
+    "DECISION",
+    "ISSUES_TO_CREATE",
+    "NON_GOALS",
+    "UNRESOLVED_QUESTIONS",
+    "NEXT_PROMPT",
+    "LAST_UPDATED_BY",
+)
+
+QUEUE_CONTROLLER_FIELDS = (
+    "QUEUE_STATE",
+    "SELECTED_OBJECT",
+    "SELECTED_OBJECT_TYPE",
+    "NEXT_ACTOR",
+    "NEXT_ACTION",
+    "NEXT_PROMPT",
+    "EVIDENCE",
+    "LAST_UPDATED_BY",
+)
+
+FINAL_REPORT_NEXT_ACTION_FIELDS = (
+    ("Current status", ("current status",)),
+    ("Next actor", ("next actor", "next_actor")),
+    ("Next action", ("next action", "next_action")),
+    ("Next prompt", ("next prompt", "next_prompt")),
+)
+
+STATE_COMMENT_KINDS = frozenset({
+    "issue_state",
+    "pr_state",
+    "discussion_state",
+    "discussion_summary",
+    "queue_controller",
+})
+
+_FIELDS_BY_KIND = {
+    "issue_state": ISSUE_STATE_FIELDS,
+    "pr_state": PR_STATE_FIELDS,
+    "discussion_state": DISCUSSION_STATE_FIELDS,
+    "discussion_summary": DISCUSSION_SUMMARY_FIELDS,
+    "queue_controller": QUEUE_CONTROLLER_FIELDS,
+}
+
+_FIELD_LINE_RE = re.compile(
+    r"^([A-Z][A-Z0-9_]+)\s*:\s*(.*)$",
+    re.MULTILINE,
+)
+_HANDOFF_SECTION_RE = re.compile(r"^##\s*Controller Handoff\s*$", re.I | re.M)
+_READY_TO_MERGE_RE = re.compile(
+    r"\b(?:ready[_ ]to[_ ]merge|merge[_ ]ready\s*:\s*(?:yes|true))\b",
+    re.IGNORECASE,
+)
+_APPROVAL_PROOF_RE = re.compile(
+    r"\b(?:review decision\s*:\s*approve|approved at|approval at current head|"
+    r"formal approval|review_status\s*:\s*approved)\b",
+    re.IGNORECASE,
+)
+_EXTERNAL_NONE_RE = re.compile(
+    r"^\s*[-*]?\s*external[- ]state mutations\s*:\s*none\s*$",
+    re.IGNORECASE | re.MULTILINE,
+)
+_LOCK_MUTATION_RE = re.compile(
+    r"\b(?:gitea_lock_issue|issue-locks/|lock file edited)\b",
+    re.IGNORECASE,
+)
+_ISSUE_DONE_RE = re.compile(
+    r"\b(?:issue (?:done|closed)\b|current status\s*:\s*(?:done|closed)\b|"
+    r"current status\s*:\s*issue complete\b)",
+    re.IGNORECASE,
+)
+_PR_MERGE_PROOF_RE = re.compile(
+    r"\b(?:pr (?:merged|number)|merge result\s*:\s*merged|pr mutations\s*:\s*created)\b",
+    re.IGNORECASE,
+)
+_DISCUSSION_COMPLETE_RE = re.compile(
+    r"\b(?:discussion complete|discussion ready for issue creation)\b",
+    re.IGNORECASE,
+)
+_DISCUSSION_SUMMARY_PROOF_RE = re.compile(
+    r"\b(?:discussion summary|issues to create|issues created|linked issues)\b",
+    re.IGNORECASE,
+)
+
+MIN_DISCUSSION_COMMENTS_DEFAULT = 5
+
+
+def _render_fields(fields: tuple[str, ...], values: dict[str, Any]) -> str:
+    lines = ["```text"]
+    for name in fields:
+        raw = values.get(name, values.get(name.lower(), "none"))
+        text = str(raw).strip() if raw is not None else "none"
+        lines.append(f"{name}: {text or 'none'}")
+    lines.append("```")
+    return "\n".join(lines)
+
+
+def render_issue_state_comment(**values: Any) -> str:
+    """Render canonical issue state comment body."""
+    return _render_fields(ISSUE_STATE_FIELDS, values)
+
+
+def render_pr_state_comment(**values: Any) -> str:
+    """Render canonical PR state comment body."""
+    return _render_fields(PR_STATE_FIELDS, values)
+
+
+def render_discussion_state_comment(**values: Any) -> str:
+    """Render canonical discussion state comment body."""
+    return _render_fields(DISCUSSION_STATE_FIELDS, values)
+
+
+def render_discussion_summary_comment(**values: Any) -> str:
+    """Render discussion-to-issue conversion summary comment."""
+    return _render_fields(DISCUSSION_SUMMARY_FIELDS, values)
+
+
+def render_queue_controller_report(**values: Any) -> str:
+    """Render queue-controller selection report."""
+    return _render_fields(QUEUE_CONTROLLER_FIELDS, values)
+
+
+def parse_state_comment(text: str) -> dict[str, str]:
+    """Parse KEY: value lines from a canonical state comment."""
+    parsed: dict[str, str] = {}
+    for match in _FIELD_LINE_RE.finditer(text or ""):
+        parsed[match.group(1)] = match.group(2).strip()
+    return parsed
+
+
+def _handoff_field_map(report_text: str) -> dict[str, str]:
+    section_match = _HANDOFF_SECTION_RE.search(report_text or "")
+    if not section_match:
+        return {}
+    section = (report_text or "")[section_match.end():]
+    fields: dict[str, str] = {}
+    for line in section.splitlines():
+        stripped = line.strip().lstrip("-*").strip()
+        if ":" not in stripped:
+            continue
+        key, value = stripped.split(":", 1)
+        fields[key.strip().lower()] = value.strip()
+    return fields
+
+
+def assess_state_comment(text: str, *, kind: str) -> dict[str, Any]:
+    """Validate a canonical Gitea state comment for *kind*."""
+    normalized = (kind or "").strip().lower()
+    if normalized not in STATE_COMMENT_KINDS:
+        return {
+            "complete": False,
+            "block": True,
+            "missing_fields": [],
+            "reasons": [f"unknown state comment kind '{kind}'"],
+            "safe_next_action": "use a supported state comment kind",
+        }
+
+    required = _FIELDS_BY_KIND[normalized]
+    parsed = parse_state_comment(text)
+    missing = [name for name in required if name not in parsed or not parsed[name].strip()]
+    reasons = [f"state comment missing field: {name}" for name in missing]
+
+    if normalized == "discussion_state":
+        urgency = parsed.get("URGENCY", "").strip().lower()
+        count_raw = parsed.get("COMMENT_COUNT", "").strip()
+        try:
+            count = int(count_raw)
+        except (TypeError, ValueError):
+            count = -1
+        if (
+            urgency not in {"urgent", "trivial"}
+            and count >= 0
+            and count < MIN_DISCUSSION_COMMENTS_DEFAULT
+        ):
+            reasons.append(
+                f"discussion has {count} comments; need at least "
+                f"{MIN_DISCUSSION_COMMENTS_DEFAULT} substantive comments "
+                "or URGENCY: urgent|trivial"
+            )
+
+    block = bool(missing or reasons)
+    return {
+        "complete": not block,
+        "block": block,
+        "missing_fields": missing,
+        "parsed_fields": parsed,
+        "reasons": reasons,
+        "safe_next_action": (
+            "fill all required state comment fields before posting"
+            if block
+            else "proceed"
+        ),
+    }
+
+
+def assess_final_report_next_action_handoff(report_text: str) -> dict[str, Any]:
+    """Require current status plus next actor/action/prompt in Controller Handoff."""
+    fields = _handoff_field_map(report_text)
+    if not fields:
+        return {
+            "complete": False,
+            "block": True,
+            "missing_fields": [name for name, _ in FINAL_REPORT_NEXT_ACTION_FIELDS],
+            "reasons": [f"final report has no section titled exactly '{HANDOFF_HEADING}'"],
+            "safe_next_action": f"add a complete {HANDOFF_HEADING} section",
+        }
+
+    missing = []
+    for name, aliases in FINAL_REPORT_NEXT_ACTION_FIELDS:
+        value = ""
+        for alias in aliases:
+            if alias in fields:
+                value = fields[alias]
+                break
+        if not value or value.lower() in {"none", "n/a", "not verified in this session", ""}:
+            missing.append(name)
+
+    reasons = [f"handoff missing next-action field: {name}" for name in missing]
+    block = bool(missing)
+    return {
+        "complete": not block,
+        "block": block,
+        "missing_fields": missing,
+        "reasons": reasons,
+        "safe_next_action": (
+            "add Current status, Next actor, Next action, and Next prompt to Controller Handoff"
+            if block
+            else "proceed"
+        ),
+    }
+
+
+def assess_contradictory_state_handoff(report_text: str) -> dict[str, Any]:
+    """Fail closed on contradictory queue/state claims in final reports."""
+    text = report_text or ""
+    fields = _handoff_field_map(text)
+    reasons: list[str] = []
+
+    review_decision = fields.get("review decision", fields.get("review status", ""))
+    has_approval = (
+        review_decision.lower() in {"approve", "approved"}
+        or bool(_APPROVAL_PROOF_RE.search(text))
+    )
+    if _READY_TO_MERGE_RE.search(text) and not has_approval:
+        reasons.append(
+            "report claims ready to merge without approval-at-current-head proof"
+        )
+
+    lock_val = fields.get("external-state mutations", "")
+    mutation_lines = " ".join(
+        fields.get(key, "")
+        for key in (
+            "issue mutations",
+            "mcp/gitea mutations",
+            "external-state mutations",
+            "claim/lock state",
+        )
+    )
+    if (
+        (_EXTERNAL_NONE_RE.search(text) or lock_val.lower() == "none")
+        and _LOCK_MUTATION_RE.search(mutation_lines)
+    ):
+        reasons.append(
+            "report claims External-state mutations: none while lock "
+            "activity is documented in mutation fields"
+        )
+
+    if _ISSUE_DONE_RE.search(text) and not _PR_MERGE_PROOF_RE.search(text):
+        reasons.append(
+            "report claims issue done/complete without PR or merge proof"
+        )
+
+    if _DISCUSSION_COMPLETE_RE.search(text) and not _DISCUSSION_SUMMARY_PROOF_RE.search(text):
+        reasons.append(
+            "report claims discussion complete without summary or issue-creation proof"
+        )
+
+    review_status = fields.get("review status", fields.get("review decision", ""))
+    merge_ready = fields.get("merge ready", "")
+    if (
+        merge_ready.lower() in {"yes", "true", "ready"}
+        and review_status.lower() not in {"approve", "approved"}
+        and not _APPROVAL_PROOF_RE.search(text)
+    ):
+        reasons.append(
+            "MERGE_READY or merge-ready claim without approved review status"
+        )
+
+    block = bool(reasons)
+    return {
+        "complete": not block,
+        "block": block,
+        "reasons": reasons,
+        "safe_next_action": (
+            "correct contradictory state claims or add missing proof"
+            if block
+            else "proceed"
+        ),
+    }
+
+
+def assess_state_handoff_ledger_report(report_text: str) -> dict[str, Any]:
+    """Composite #494 assessment for final reports."""
+    next_action = assess_final_report_next_action_handoff(report_text)
+    contradictions = assess_contradictory_state_handoff(report_text)
+    reasons = list(next_action.get("reasons") or []) + list(contradictions.get("reasons") or [])
+    block = bool(next_action.get("block") or contradictions.get("block"))
+    return {
+        "complete": not block,
+        "block": block,
+        "next_action": next_action,
+        "contradictions": contradictions,
+        "reasons": reasons,
+        "safe_next_action": (
+            next_action.get("safe_next_action")
+            if next_action.get("block")
+            else contradictions.get("safe_next_action")
+            if contradictions.get("block")
+            else "proceed"
+        ),
+    }

tests/test_final_report_validator.py

@@ -35,6 +35,9 @@ def _review_handoff(**overrides):
         "- External-state mutations: none",
         "- Read-only diagnostics: issue and PR metadata inspected",
         "- Current status: PR open",
+        "- Next actor: author",
+        "- Next action: address review feedback",
+        "- Next prompt: Fix PR #203 per reviewer request_changes and push updated head",
         "- Blockers: none",
         "- Next: await author",
         "- Safety: no self-review; no self-merge",
@@ -92,6 +95,9 @@ def _reconcile_handoff(drop=(), **overrides):
         "- Read-only diagnostics: gitea_view_pr, gitea_view_issue",
         "- Reconciliation mutations: PR comment posted",
         "- Current status: reconciliation complete",
+        "- Next actor: reconciler",
+        "- Next action: close PR #99 after capability proof",
+        "- Next prompt: Reconcile already-landed PR #99 with prgs-reconciler profile",
         "- Blocker: missing gitea.pr.close capability",
         "- Safe next action: hand off to a profile with gitea.pr.close",
         "- No review/merge confirmation: confirmed",

tests/test_state_handoff_ledger.py

@@ -0,0 +1,243 @@
+"""Tests for canonical state handoff ledger (#494)."""
+import sys
+import unittest
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+from final_report_validator import assess_final_report_validator  # noqa: E402
+from state_handoff_ledger import (  # noqa: E402
+    ISSUE_STATE_FIELDS,
+    assess_contradictory_state_handoff,
+    assess_final_report_next_action_handoff,
+    assess_state_comment,
+    assess_state_handoff_ledger_report,
+    parse_state_comment,
+    render_issue_state_comment,
+    render_pr_state_comment,
+    render_queue_controller_report,
+)
+
+
+def _work_issue_handoff(**overrides):
+    lines = [
+        "## Controller Handoff",
+        "",
+        "- Task: work-issue",
+        "- Repo: Scaled-Tech-Consulting/Gitea-Tools",
+        "- Role: author",
+        "- Identity: jcwalker3 / prgs-author",
+        "- Selected issue: #494",
+        "- Current status: implementation complete; PR opened awaiting review",
+        "- Next actor: reviewer",
+        "- Next action: review PR at pinned head",
+        "- Next prompt: Review PR #500 for issue #494 at current head in reviewer worktree",
+        "- Safe next action: switch to reviewer session",
+        "- External-state mutations: none",
+    ]
+    text = "\n".join(lines)
+    for key, value in overrides.items():
+        needle = f"- {key}:"
+        if needle in text:
+            prefix = text.split(needle)[0]
+            rest = text.split(needle, 1)[1]
+            suffix = rest.split("\n", 1)[1] if "\n" in rest else ""
+            text = f"{prefix}{needle} {value}\n{suffix}".rstrip()
+        else:
+            text += f"\n- {key}: {value}"
+    return text
+
+
+class TestStateCommentTemplates(unittest.TestCase):
+    def test_render_issue_state_includes_all_fields(self):
+        body = render_issue_state_comment(
+            STATE="issue_ready_for_author",
+            NEXT_ACTOR="author",
+            NEXT_ACTION="lock and implement",
+            NEXT_PROMPT="Find issue #494 and open a PR",
+            LAST_UPDATED_BY="jcwalker3",
+        )
+        parsed = parse_state_comment(body)
+        for field in ISSUE_STATE_FIELDS:
+            self.assertIn(field, parsed)
+
+    def test_render_pr_state_parses(self):
+        body = render_pr_state_comment(
+            STATE="needs_review",
+            NEXT_ACTOR="reviewer",
+            NEXT_ACTION="review at head",
+            NEXT_PROMPT="Review PR #500",
+            ISSUE="#494",
+            HEAD_SHA="a" * 40,
+            LAST_UPDATED_BY="sysadmin",
+        )
+        parsed = parse_state_comment(body)
+        self.assertEqual(parsed["STATE"], "needs_review")
+        self.assertEqual(parsed["NEXT_ACTOR"], "reviewer")
+
+    def test_queue_controller_template(self):
+        body = render_queue_controller_report(
+            QUEUE_STATE="open_prs_need_review",
+            SELECTED_OBJECT="PR #500",
+            SELECTED_OBJECT_TYPE="pr",
+            NEXT_ACTOR="reviewer",
+            NEXT_ACTION="start review session",
+            NEXT_PROMPT="Review PR #500",
+            LAST_UPDATED_BY="controller",
+        )
+        result = assess_state_comment(body, kind="queue_controller")
+        self.assertTrue(result["complete"])
+
+
+class TestStateCommentValidation(unittest.TestCase):
+    def test_complete_issue_state_comment(self):
+        body = render_issue_state_comment(
+            STATE="in_progress",
+            NEXT_ACTOR="author",
+            NEXT_ACTION="push branch",
+            NEXT_PROMPT="Continue issue #494",
+            STATUS="open",
+            RELATED_PRS="none",
+            BRANCH="feat/issue-494-state-handoff-ledger",
+            HEAD_SHA="b" * 40,
+            VALIDATION="pytest passed",
+            BLOCKERS="none",
+            DUPLICATES_OR_SUPERSEDES="none",
+            LAST_UPDATED_BY="jcwalker3",
+        )
+        result = assess_state_comment(body, kind="issue_state")
+        self.assertTrue(result["complete"])
+        self.assertFalse(result["block"])
+
+    def test_missing_fields_blocked(self):
+        body = "STATE: open\nNEXT_ACTOR: author"
+        result = assess_state_comment(body, kind="issue_state")
+        self.assertFalse(result["complete"])
+        self.assertIn("NEXT_ACTION", result["missing_fields"])
+
+    def test_discussion_requires_five_comments_by_default(self):
+        body = render_issue_state_comment(
+            STATE="collecting_feedback",
+            NEXT_ACTOR="controller",
+            NEXT_ACTION="facilitate discussion",
+            NEXT_PROMPT="Add acceptance criteria",
+            STATUS="open",
+            RELATED_PRS="none",
+            BRANCH="none",
+            HEAD_SHA="none",
+            VALIDATION="none",
+            BLOCKERS="none",
+            DUPLICATES_OR_SUPERSEDES="none",
+            LAST_UPDATED_BY="controller",
+        )
+        # Re-map to discussion fields manually for comment-count test
+        from state_handoff_ledger import render_discussion_state_comment
+
+        disc = render_discussion_state_comment(
+            STATE="collecting_feedback",
+            NEXT_ACTOR="controller",
+            NEXT_ACTION="wait for comments",
+            NEXT_PROMPT="Continue discussion",
+            COMMENT_COUNT="2",
+            SUBSTANTIVE_COMMENTS="yes",
+            URGENCY="normal",
+            BLOCKERS="none",
+            LAST_UPDATED_BY="controller",
+        )
+        result = assess_state_comment(disc, kind="discussion_state")
+        self.assertFalse(result["complete"])
+        self.assertTrue(any("5" in reason for reason in result["reasons"]))
+
+    def test_discussion_urgent_exception(self):
+        from state_handoff_ledger import render_discussion_state_comment
+
+        disc = render_discussion_state_comment(
+            STATE="ready_for_summary",
+            NEXT_ACTOR="controller",
+            NEXT_ACTION="summarize",
+            NEXT_PROMPT="Post summary",
+            COMMENT_COUNT="2",
+            SUBSTANTIVE_COMMENTS="yes",
+            URGENCY="urgent",
+            BLOCKERS="none",
+            LAST_UPDATED_BY="controller",
+        )
+        result = assess_state_comment(disc, kind="discussion_state")
+        self.assertTrue(result["complete"])
+
+
+class TestFinalReportNextAction(unittest.TestCase):
+    def test_complete_next_action_handoff(self):
+        report = _work_issue_handoff()
+        result = assess_final_report_next_action_handoff(report)
+        self.assertTrue(result["complete"])
+        self.assertFalse(result["block"])
+
+    def test_missing_next_prompt_blocked(self):
+        report = _work_issue_handoff(**{"Next prompt": "none"})
+        result = assess_final_report_next_action_handoff(report)
+        self.assertFalse(result["complete"])
+        self.assertIn("Next prompt", result["missing_fields"])
+
+    def test_missing_handoff_section_blocked(self):
+        result = assess_final_report_next_action_handoff("no handoff here")
+        self.assertTrue(result["block"])
+
+
+class TestContradictoryState(unittest.TestCase):
+    def test_ready_to_merge_without_approval_blocked(self):
+        report = _work_issue_handoff(
+            **{
+                "Current status": "PR ready to merge",
+                "Review decision": "not attempted",
+            }
+        )
+        result = assess_contradictory_state_handoff(report)
+        self.assertTrue(result["block"])
+
+    def test_ready_to_merge_with_approval_allowed(self):
+        report = _work_issue_handoff(
+            **{
+                "Current status": "PR ready to merge",
+                "Review decision": "approve",
+            }
+        )
+        result = assess_contradictory_state_handoff(report)
+        self.assertFalse(result["block"])
+
+    def test_issue_done_without_pr_proof_blocked(self):
+        report = _work_issue_handoff(**{"Current status": "issue complete"})
+        result = assess_contradictory_state_handoff(report)
+        self.assertTrue(result["block"])
+
+    def test_external_none_with_lock_activity_blocked(self):
+        report = _work_issue_handoff(
+            **{
+                "External-state mutations": "none",
+            }
+        )
+        report += "\n- Issue mutations: gitea_lock_issue adopted branch"
+        result = assess_contradictory_state_handoff(report)
+        self.assertTrue(result["block"])
+
+    def test_discussion_complete_without_summary_blocked(self):
+        report = _work_issue_handoff(**{"Current status": "discussion complete"})
+        result = assess_contradictory_state_handoff(report)
+        self.assertTrue(result["block"])
+
+
+class TestFinalReportValidatorIntegration(unittest.TestCase):
+    def test_work_issue_validator_flags_missing_next_prompt(self):
+        report = _work_issue_handoff(**{"Next prompt": ""})
+        result = assess_final_report_validator(report, "work_issue")
+        rule_ids = [f["rule_id"] for f in result["findings"]]
+        self.assertIn("shared.state_handoff_next_action", rule_ids)
+
+    def test_work_issue_validator_accepts_complete_handoff(self):
+        report = _work_issue_handoff()
+        result = assess_state_handoff_ledger_report(report)
+        self.assertTrue(result["complete"])
+
+
+if __name__ == "__main__":
+    unittest.main()