diff --git a/docs/examples/two-comment-workflow-examples.md b/docs/examples/two-comment-workflow-examples.md
new file mode 100644
index 0000000..84e1d58
--- /dev/null
+++ b/docs/examples/two-comment-workflow-examples.md
@@ -0,0 +1,43 @@
+# Two-comment workflow examples (#507)
+
+Paired `[CONTROLLER HANDOFF]` + `[THREAD STATE LEDGER]` comments for Gitea threads.
+See `thread_state_ledger_examples.py` for machine-checked fixtures.
+
+## Approved review posted
+
+**Handoff** (detailed): identity, worktree, validation commands, mutation ledger with
+`gitea_submit_pr_review → APPROVED review posted to Gitea`.
+
+**Ledger** (concise):
+
+```markdown
+[THREAD STATE LEDGER] PR #487 — APPROVED review posted to Gitea
+
+What is true now:
+- PR state: open
+- Server-side decision state: APPROVED review posted to Gitea
+- Local verdict/state: APPROVE verdict prepared locally
+
+What is blocked:
+- Blocker classification: no blocker
+
+Who/what acts next:
+- Next actor: merger
+- Required action: merge on explicit operator command
+- Do not do: re-post APPROVE
+```
+
+## Approve validated locally but blocked before posting
+
+Ledger must show `no server-side state changed` under server-side decision state and
+`APPROVE verdict prepared locally` under local verdict/state.
+
+## Environment / tooling blocker
+
+Ledger blocker classification: `environment/tooling blocker`. Mutation ledger:
+`none — no server-side state changed`.
+
+## Stale head blocker
+
+Ledger: `approval_at_current_head is false`; classification `stale head`.
+Do not do: merge with stale approval.
\ No newline at end of file
diff --git a/docs/llm-workflow-runbooks.md b/docs/llm-workflow-runbooks.md
index d67769d..c6ee768 100644
--- a/docs/llm-workflow-runbooks.md
+++ b/docs/llm-workflow-runbooks.md
@@ -760,6 +760,46 @@ touched release state names the exact tag/commit and why. Design debates
 belong in **discussion/RFC issues** (e.g. #100 `profiles.json v2`) — comment
 on the issue, create no branches/PRs, and end the comment with this handoff.
 
+## Two-comment workflow reporting (#507)
+
+After meaningful controller/workflow work, post **two separate Gitea comments**
+(not one combined blob):
+
+1. **`[CONTROLLER HANDOFF]`** — detailed operational continuation for the
+   next LLM/controller (proof-heavy; may be long).
+2. **`[THREAD STATE LEDGER]`** — short canonical truth readable in ~30 seconds.
+
+The ledger must answer: what is true now, what changed, what is blocked,
+who/what acts next — and must **separate**:
+
+- local verdict/state
+- server-side Gitea state
+- attempted-but-blocked mutations
+- completed mutations
+
+Use precise state phrases (`APPROVED review posted to Gitea`,
+`APPROVE verdict prepared locally`, `merge performed`, `merge not performed`,
+`no server-side state changed`, `lease attempt blocked`) instead of ambiguous
+standalone words (`approved`, `merged`, `ready`, `blocked`, `done`).
+
+The ledger must include a **blocker classification** from:
+`code blocker`, `test blocker`, `merge conflict`, `stale head`,
+`permission/capability blocker`, `environment/tooling blocker`,
+`process/rule blocker`, `queue/lease blocker`,
+`duplicate/canonicalization blocker`, `no blocker`.
+
+Templates and worked examples:
+[`examples/two-comment-workflow-examples.md`](examples/two-comment-workflow-examples.md).
+
+Validation: `thread_state_ledger_validator.py` checks tagged comments at post
+time (`gitea_create_issue_comment`) and tagged final reports via
+`assess_final_report_validator`. Legacy `## Controller Handoff` final reports
+remain valid during transition; the tagged pair is required for new workflow
+comments.
+
+Related (do not duplicate): #494/#495 lifecycle state, #501 mutation-ledger
+consistency, #505 CTH umbrella, #496 workflow comment gate when merged.
+
 ## Fail-closed behavior
 
 Before any mutating action the workflow verifies identity, active profile,
diff --git a/final_report_validator.py b/final_report_validator.py
index 352a3ff..a9cc1fc 100644
--- a/final_report_validator.py
+++ b/final_report_validator.py
@@ -12,6 +12,7 @@ import re
 from typing import Any, Callable
 
 import issue_lock_provenance
+import thread_state_ledger_validator
 from review_proofs import (
     HANDOFF_HEADING,
     assess_controller_handoff,
@@ -1042,16 +1043,26 @@ def _rule_reviewer_review_mutation(
     )
 
 
+def _rule_shared_two_comment_workflow(report_text: str) -> list[dict[str, str]]:
+    """#507: tagged Controller Handoff must pair with Thread State Ledger."""
+    return thread_state_ledger_validator.findings_for_final_report(report_text)
+
+
 _SHARED_ISSUE_LOCK_RULES = (
     _rule_shared_issue_lock_external_state,
     _rule_shared_manual_lock_pr_override,
     _rule_shared_author_reviewer_same_run,
 )
 
+_SHARED_TWO_COMMENT_RULES = (
+    _rule_shared_two_comment_workflow,
+)
+
 _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = {
     "review_pr": [
         _rule_shared_controller_handoff,
         _rule_shared_email_disclosure,
+        *_SHARED_TWO_COMMENT_RULES,
         *_SHARED_ISSUE_LOCK_RULES,
         _rule_reviewer_legacy_workspace_mutations,
         _rule_reviewer_vague_mutations_none,
@@ -1076,6 +1087,7 @@ _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = {
     "reconcile_already_landed": [
         _rule_reconcile_controller_handoff,
         _rule_shared_email_disclosure,
+        *_SHARED_TWO_COMMENT_RULES,
         *_SHARED_ISSUE_LOCK_RULES,
         _rule_reconcile_stale_author_fields,
         _rule_reconcile_eligible_reviewed,
@@ -1088,12 +1100,14 @@ _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = {
     "author_issue": [
         _rule_shared_controller_handoff,
         _rule_shared_email_disclosure,
+        *_SHARED_TWO_COMMENT_RULES,
         *_SHARED_ISSUE_LOCK_RULES,
         _rule_reviewer_vague_mutations_none,
     ],
     "work_issue": [
         _rule_shared_controller_handoff,
         _rule_shared_email_disclosure,
+        *_SHARED_TWO_COMMENT_RULES,
         *_SHARED_ISSUE_LOCK_RULES,
         _rule_reviewer_vague_mutations_none,
         _rule_conflict_fix_push_proof,
@@ -1101,17 +1115,20 @@ _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = {
     "issue_filing": [
         _rule_shared_controller_handoff,
         _rule_shared_email_disclosure,
+        *_SHARED_TWO_COMMENT_RULES,
         *_SHARED_ISSUE_LOCK_RULES,
     ],
     "inventory": [
         _rule_shared_controller_handoff,
         _rule_shared_email_disclosure,
+        *_SHARED_TWO_COMMENT_RULES,
         *_SHARED_ISSUE_LOCK_RULES,
         _rule_reconcile_pagination_proof,
     ],
     "issue_selection": [
         _rule_shared_controller_handoff,
         _rule_shared_email_disclosure,
+        *_SHARED_TWO_COMMENT_RULES,
         *_SHARED_ISSUE_LOCK_RULES,
     ],
 }
diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py
index 160e4de..2a08151 100644
--- a/gitea_mcp_server.py
+++ b/gitea_mcp_server.py
@@ -548,6 +548,7 @@ import reconciliation_workflow  # noqa: E402
 import review_merge_state_machine  # noqa: E402
 import pr_work_lease  # noqa: E402
 import native_mcp_preference  # noqa: E402
+import thread_state_ledger_validator  # noqa: E402
 
 
 # Keyed issue-lock storage (#443): per remote/org/repo/issue files under
@@ -5217,6 +5218,23 @@ def gitea_list_issue_comments(
     return {"success": True, "issue_number": issue_number, "comments": out}
 
 
+def _two_comment_workflow_comment_gate(body: str) -> list[str]:
+    """Fail closed on invalid tagged workflow comments (#507)."""
+    text = body or ""
+    reasons: list[str] = []
+    if thread_state_ledger_validator.has_controller_handoff_marker(text):
+        result = thread_state_ledger_validator.assess_controller_handoff_comment(
+            text
+        )
+        reasons.extend(result.get("reasons") or [])
+    if thread_state_ledger_validator.has_thread_state_ledger_marker(text):
+        result = thread_state_ledger_validator.assess_thread_state_ledger_comment(
+            text
+        )
+        reasons.extend(result.get("reasons") or [])
+    return reasons
+
+
 @mcp.tool()
 def gitea_create_issue_comment(
     issue_number: int,
@@ -5259,6 +5277,8 @@ def gitea_create_issue_comment(
     reasons = list(gate_reasons)
     if not (body or "").strip():
         reasons.append("comment body must be a non-empty string")
+    if not reasons:
+        reasons.extend(_two_comment_workflow_comment_gate(body))
     if reasons:
         blocked = {"success": False, "performed": False,
                    "issue_number": issue_number, "reasons": reasons}
diff --git a/tests/test_final_report_validator.py b/tests/test_final_report_validator.py
index 88c5b23..45942af 100644
--- a/tests/test_final_report_validator.py
+++ b/tests/test_final_report_validator.py
@@ -108,6 +108,28 @@ def _reconcile_handoff(drop=(), **overrides):
     return text
 
 
+class TestTwoCommentWorkflowIntegration(unittest.TestCase):
+    def test_tagged_pair_produces_no_two_comment_findings(self):
+        from thread_state_ledger_validator import findings_for_final_report  # noqa: E402
+        from tests.test_thread_state_ledger_validator import (  # noqa: E402
+            _minimal_handoff,
+            _minimal_ledger,
+        )
+
+        report = _minimal_handoff() + "\n\n" + _minimal_ledger()
+        self.assertEqual(findings_for_final_report(report), [])
+
+    def test_tagged_handoff_without_ledger_blocks_via_validator(self):
+        from thread_state_ledger_validator import findings_for_final_report  # noqa: E402
+        from tests.test_thread_state_ledger_validator import _minimal_handoff  # noqa: E402
+
+        findings = findings_for_final_report(_minimal_handoff())
+        self.assertTrue(findings)
+        self.assertTrue(
+            any(f["severity"] == "block" for f in findings)
+        )
+
+
 class TestValidatorFinding(unittest.TestCase):
     def test_finding_shape(self):
         finding = validator_finding(
diff --git a/tests/test_mcp_server.py b/tests/test_mcp_server.py
index 651b253..67b3974 100644
--- a/tests/test_mcp_server.py
+++ b/tests/test_mcp_server.py
@@ -3078,6 +3078,20 @@ class TestIssueCommentTools(unittest.TestCase):
         self.assertFalse(result["success"])
         mock_api.assert_not_called()
 
+    @patch("mcp_server.api_request")
+    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
+    def test_create_tagged_handoff_missing_mutation_ledger_blocked(
+        self, _auth, mock_api
+    ):
+        body = "[CONTROLLER HANDOFF] PR #1 — test\n\nBlockers:\n- none"
+        with patch.dict(os.environ, self.AUTHOR_ENV, clear=True):
+            result = gitea_create_issue_comment(
+                issue_number=9, body=body, remote="prgs")
+        self.assertFalse(result["success"])
+        self.assertFalse(result["performed"])
+        self.assertTrue(result["reasons"])
+        mock_api.assert_not_called()
+
     @patch("mcp_server.api_request")
     @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
     def test_create_empty_body_blocked(self, _auth, mock_api):
diff --git a/tests/test_thread_state_ledger_validator.py b/tests/test_thread_state_ledger_validator.py
new file mode 100644
index 0000000..8b5b0f2
--- /dev/null
+++ b/tests/test_thread_state_ledger_validator.py
@@ -0,0 +1,187 @@
+"""Tests for two-comment workflow validation (#507)."""
+import sys
+import unittest
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+from thread_state_ledger_validator import (  # noqa: E402
+    BLOCKER_CLASSIFICATIONS,
+    assess_controller_handoff_comment,
+    assess_thread_state_ledger_comment,
+    assess_two_comment_workflow,
+    assess_two_comment_pair,
+)
+
+
+def _minimal_handoff(**overrides):
+    lines = [
+        "[CONTROLLER HANDOFF] PR #203 / Issue #182 — review complete",
+        "",
+        "Identity/profile:",
+        "- Active profile: prgs-reviewer",
+        "- Authenticated identity: sysadmin",
+        "",
+        "Server-side mutation ledger:",
+        "- gitea_submit_pr_review → APPROVED review posted to Gitea",
+        "",
+        "Blockers:",
+        "- none",
+    ]
+    text = "\n".join(lines)
+    for key, value in overrides.items():
+        if f"{key}:" in text:
+            text = text.replace(f"{key}:", f"{key}: {value}", 1)
+    return text
+
+
+def _minimal_ledger(**overrides):
+    lines = [
+        "[THREAD STATE LEDGER] PR #203 — APPROVED review posted to Gitea",
+        "",
+        "What is true now:",
+        "- PR state: open",
+        "- Current head SHA: 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9",
+        "- Server-side decision state: APPROVED review posted to Gitea",
+        "- Local verdict/state: APPROVE verdict prepared locally",
+        "- Latest known validation: tests passed locally",
+        "",
+        "What changed:",
+        "- APPROVED review posted to Gitea at pinned head",
+        "",
+        "What is blocked:",
+        "- Blocker classification: no blocker",
+        "",
+        "Who/what acts next:",
+        "- Next actor: merger",
+        "- Required action: merge on explicit operator command",
+        "- Do not do: re-post APPROVE",
+        "- Resume from: PR #203 review feedback",
+    ]
+    text = "\n".join(lines)
+    for key, value in overrides.items():
+        text = text.replace(f"- {key}:", f"- {key}: {value}")
+    return text
+
+
+class TestTwoCommentWorkflow(unittest.TestCase):
+    def test_valid_combined_report_passes(self):
+        report = _minimal_handoff() + "\n\n" + _minimal_ledger()
+        result = assess_two_comment_workflow(report)
+        self.assertTrue(result["proven"])
+        self.assertFalse(result["block"])
+
+    def test_missing_ledger_after_tagged_handoff_blocks(self):
+        result = assess_two_comment_workflow(_minimal_handoff())
+        self.assertTrue(result["block"])
+        self.assertTrue(
+            any("THREAD STATE LEDGER" in r for r in result["reasons"])
+        )
+
+    def test_legacy_controller_handoff_section_not_required_ledger(self):
+        legacy = "## Controller Handoff\n\n- Task: work issue\n- Next: continue\n"
+        result = assess_two_comment_workflow(legacy)
+        self.assertFalse(result["block"])
+
+    def test_ambiguous_approved_blocks(self):
+        handoff = _minimal_handoff().replace(
+            "APPROVED review posted to Gitea",
+            "approved",
+        )
+        report = handoff + "\n\n" + _minimal_ledger()
+        result = assess_two_comment_workflow(report)
+        self.assertTrue(result["block"])
+        self.assertTrue(any("approved" in r.lower() for r in result["reasons"]))
+
+    def test_ambiguous_merged_blocks(self):
+        handoff = _minimal_handoff() + "\n\n- Narrative: merged successfully"
+        ledger = _minimal_ledger()
+        report = handoff + "\n\n" + ledger
+        result = assess_two_comment_workflow(report)
+        self.assertTrue(result["block"])
+        self.assertTrue(any("merge" in r.lower() for r in result["reasons"]))
+
+    def test_blocked_without_gate_blocks(self):
+        handoff = _minimal_handoff().replace("- none", "- blocked")
+        ledger = _minimal_ledger().replace(
+            "no blocker", "blocked but no gate named"
+        )
+        report = handoff + "\n\n" + ledger
+        result = assess_two_comment_workflow(report)
+        self.assertTrue(result["block"])
+
+    def test_mutation_ledger_contradiction_blocks(self):
+        handoff = _minimal_handoff().replace(
+            "APPROVED review posted to Gitea",
+            "none — no server-side state changed",
+        )
+        handoff += "\n- Narrative: APPROVED review posted to Gitea"
+        report = handoff + "\n\n" + _minimal_ledger()
+        result = assess_two_comment_workflow(report)
+        self.assertTrue(result["block"])
+        self.assertTrue(any("contradict" in r.lower() for r in result["reasons"]))
+
+    def test_local_verdict_as_server_side_blocks(self):
+        ledger = _minimal_ledger(
+            **{
+                "Server-side decision state": "APPROVE verdict prepared locally",
+                "Local verdict/state": "none",
+            }
+        )
+        report = _minimal_handoff() + "\n\n" + ledger
+        result = assess_two_comment_workflow(report)
+        self.assertTrue(result["block"])
+
+
+class TestHandoffComment(unittest.TestCase):
+    def test_handoff_comment_valid(self):
+        result = assess_controller_handoff_comment(_minimal_handoff())
+        self.assertTrue(result["proven"])
+
+    def test_handoff_without_marker_skips(self):
+        result = assess_controller_handoff_comment("casual comment")
+        self.assertTrue(result["proven"])
+        self.assertFalse(result["performed"])
+
+
+class TestLedgerComment(unittest.TestCase):
+    def test_ledger_requires_blocker_classification(self):
+        ledger = _minimal_ledger().replace(
+            "Blocker classification: no blocker",
+            "something unclear",
+        )
+        result = assess_thread_state_ledger_comment(ledger)
+        self.assertTrue(result["block"])
+
+    def test_all_blocker_classifications_recognized(self):
+        self.assertIn("no blocker", BLOCKER_CLASSIFICATIONS)
+        self.assertIn("environment/tooling blocker", BLOCKER_CLASSIFICATIONS)
+
+
+class TestPairedComments(unittest.TestCase):
+    def test_pair_passes(self):
+        result = assess_two_comment_pair(_minimal_handoff(), _minimal_ledger())
+        self.assertTrue(result["proven"])
+
+    def test_pair_missing_ledger_blocks(self):
+        result = assess_two_comment_pair(_minimal_handoff(), "")
+        self.assertTrue(result["block"])
+
+
+class TestExamples(unittest.TestCase):
+    """Smoke-test bundled examples from docs."""
+
+    def test_examples_module_loads(self):
+        from thread_state_ledger_examples import EXAMPLES  # noqa: E402
+
+        self.assertGreaterEqual(len(EXAMPLES), 8)
+        for name, handoff, ledger in EXAMPLES:
+            result = assess_two_comment_pair(handoff, ledger)
+            self.assertTrue(
+                result["proven"],
+                f"example '{name}' failed: {result['reasons']}",
+            )
+
+
+if __name__ == "__main__":
+    unittest.main()
\ No newline at end of file
diff --git a/thread_state_ledger_examples.py b/thread_state_ledger_examples.py
new file mode 100644
index 0000000..9c17442
--- /dev/null
+++ b/thread_state_ledger_examples.py
@@ -0,0 +1,341 @@
+"""Worked examples for the two-comment workflow (#507)."""
+
+from __future__ import annotations
+
+HEAD_SHA = "08202f7eaa6d09b2eca8f2960126994a4b22646b"
+
+EXAMPLES: list[tuple[str, str, str]] = []
+
+
+def _example(name: str, handoff: str, ledger: str) -> tuple[str, str, str]:
+    return (name, handoff.strip(), ledger.strip())
+
+
+EXAMPLES.append(
+    _example(
+        "approved_review_posted",
+        f"""
+[CONTROLLER HANDOFF] PR #487 / Issue #485 — review approved
+
+Server-side mutation ledger:
+- gitea_submit_pr_review → APPROVED review posted to Gitea (comment id 6566)
+
+Blockers:
+- none
+""",
+        f"""
+[THREAD STATE LEDGER] PR #487 — APPROVED review posted to Gitea
+
+What is true now:
+- PR state: open
+- Current head SHA: {HEAD_SHA}
+- Server-side decision state: APPROVED review posted to Gitea
+- Local verdict/state: APPROVE verdict prepared locally
+- Latest known validation: tests passed locally
+
+What changed:
+- APPROVED review posted to Gitea at pinned head
+
+What is blocked:
+- Blocker classification: no blocker
+
+Who/what acts next:
+- Next actor: merger
+- Required action: merge on explicit operator command
+- Do not do: re-post APPROVE
+- Resume from: PR #487 review feedback
+""",
+    )
+)
+
+EXAMPLES.append(
+    _example(
+        "approve_blocked_before_posting",
+        f"""
+[CONTROLLER HANDOFF] PR #487 / Issue #485 — approve blocked
+
+Server-side mutation ledger:
+- none — no server-side state changed
+
+Blockers:
+- process/rule blocker: MCP process root still control checkout
+""",
+        f"""
+[THREAD STATE LEDGER] PR #487 — APPROVE verdict prepared locally; post blocked
+
+What is true now:
+- PR state: open
+- Current head SHA: {HEAD_SHA}
+- Server-side decision state: no server-side state changed
+- Local verdict/state: APPROVE verdict prepared locally
+- Latest known validation: tests passed locally
+
+What changed:
+- Review validation completed locally; APPROVE not posted
+
+What is blocked:
+- Blocker classification: process/rule blocker
+
+Who/what acts next:
+- Next actor: operator
+- Required action: reconnect reviewer MCP to branches worktree
+- Do not do: retry APPROVE from root-bound session
+- Resume from: lease + mark_final_review_decision + submit
+""",
+    )
+)
+
+EXAMPLES.append(
+    _example(
+        "request_changes_posted",
+        f"""
+[CONTROLLER HANDOFF] PR #200 — request changes
+
+Server-side mutation ledger:
+- gitea_submit_pr_review → REQUEST_CHANGES posted to Gitea
+
+Blockers:
+- none
+""",
+        """
+[THREAD STATE LEDGER] PR #200 — REQUEST_CHANGES posted to Gitea
+
+What is true now:
+- PR state: open
+- Server-side decision state: REQUEST_CHANGES posted to Gitea
+- Local verdict/state: REQUEST_CHANGES prepared locally
+- Latest known validation: tests passed locally
+
+What changed:
+- REQUEST_CHANGES posted to Gitea
+
+What is blocked:
+- Blocker classification: no blocker
+
+Who/what acts next:
+- Next actor: author
+- Required action: address review feedback and push fixes
+- Do not do: merge
+- Resume from: PR review comments
+""",
+    )
+)
+
+EXAMPLES.append(
+    _example(
+        "merge_completed",
+        f"""
+[CONTROLLER HANDOFF] PR #487 — merge performed
+
+Server-side mutation ledger:
+- gitea_merge_pr → merge performed
+
+Blockers:
+- none
+""",
+        f"""
+[THREAD STATE LEDGER] PR #487 — merge performed
+
+What is true now:
+- PR state: merged
+- Current head SHA: {HEAD_SHA}
+- Server-side decision state: merge performed
+- Local verdict/state: merge not attempted locally before MCP merge
+- Latest known validation: full suite not rerun
+
+What changed:
+- merge performed via gitea_merge_pr
+
+What is blocked:
+- Blocker classification: no blocker
+
+Who/what acts next:
+- Next actor: reconciler
+- Required action: verify linked issue closure policy
+- Do not do: re-merge
+- Resume from: post-merge cleanup checklist
+""",
+    )
+)
+
+EXAMPLES.append(
+    _example(
+        "merge_blocked",
+        f"""
+[CONTROLLER HANDOFF] PR #487 — merge not performed
+
+Server-side mutation ledger:
+- gitea_merge_pr attempted → merge not performed (stale approval gate)
+
+Blockers:
+- stale head blocker: approval not at current head
+""",
+        f"""
+[THREAD STATE LEDGER] PR #487 — merge not performed
+
+What is true now:
+- PR state: open
+- Current head SHA: {HEAD_SHA}
+- Server-side decision state: merge not performed
+- Local verdict/state: merge not performed
+- Latest known validation: tests passed locally
+
+What changed:
+- merge attempt blocked by stale-head gate
+
+What is blocked:
+- Blocker classification: stale head
+
+Who/what acts next:
+- Next actor: reviewer
+- Required action: re-review at current head
+- Do not do: merge until approval_at_current_head is true
+- Resume from: gitea_get_pr_review_feedback
+""",
+    )
+)
+
+EXAMPLES.append(
+    _example(
+        "reconciler_closure",
+        f"""
+[CONTROLLER HANDOFF] PR #99 — reconciler closure
+
+Server-side mutation ledger:
+- gitea_reconcile_already_landed_pr → PR closed on Gitea
+
+Blockers:
+- none
+""",
+        """
+[THREAD STATE LEDGER] PR #99 — reconciler closure complete
+
+What is true now:
+- PR state: closed
+- Server-side decision state: server-side state changed
+- Local verdict/state: no server-side state changed locally before MCP close
+- Latest known validation: ancestor proof passed
+
+What changed:
+- superseded PR closed by reconciler
+
+What is blocked:
+- Blocker classification: no blocker
+
+Who/what acts next:
+- Next actor: controller
+- Required action: verify queue has no duplicate open PR
+- Do not do: reopen without canonical PR proof
+- Resume from: queue inventory
+""",
+    )
+)
+
+EXAMPLES.append(
+    _example(
+        "environment_tooling_blocker",
+        f"""
+[CONTROLLER HANDOFF] PR #487 — tooling blocker
+
+Server-side mutation ledger:
+- none — no server-side state changed
+
+Blockers:
+- environment/tooling blocker: gitea-reviewer MCP process root mismatch
+""",
+        f"""
+[THREAD STATE LEDGER] PR #487 — lease attempt blocked
+
+What is true now:
+- PR state: open
+- Current head SHA: {HEAD_SHA}
+- Server-side decision state: no server-side state changed
+- Local verdict/state: APPROVE verdict prepared locally
+- Latest known validation: tests passed locally
+
+What changed:
+- lease attempt blocked by workspace guard
+
+What is blocked:
+- Blocker classification: environment/tooling blocker
+
+Who/what acts next:
+- Next actor: operator
+- Required action: set GITEA_AUTHOR_WORKTREE and /mcp reconnect
+- Do not do: acquire lease from control checkout
+- Resume from: gitea_get_runtime_context
+""",
+    )
+)
+
+EXAMPLES.append(
+    _example(
+        "stale_head_blocker",
+        f"""
+[CONTROLLER HANDOFF] PR #300 — stale head
+
+Server-side mutation ledger:
+- none — no server-side state changed
+
+Blockers:
+- stale head blocker: live head differs from pinned reviewed head
+""",
+        """
+[THREAD STATE LEDGER] PR #300 — stale approval
+
+What is true now:
+- PR state: open
+- Server-side decision state: no server-side state changed
+- Local verdict/state: approval_at_current_head is false (stale head)
+- Latest known validation: full suite not rerun
+
+What changed:
+- author pushed after approval
+
+What is blocked:
+- Blocker classification: stale head
+
+Who/what acts next:
+- Next actor: reviewer
+- Required action: fresh review at current head
+- Do not do: merge with stale approval
+- Resume from: gitea_get_pr_review_feedback
+""",
+    )
+)
+
+EXAMPLES.append(
+    _example(
+        "duplicate_canonicalization_blocker",
+        f"""
+[CONTROLLER HANDOFF] Issue filing — duplicate found
+
+Server-side mutation ledger:
+- gitea_create_issue_comment on #505 with missing acceptance criteria
+
+Blockers:
+- duplicate/canonicalization blocker: #505 tracks CTH umbrella
+""",
+        """
+[THREAD STATE LEDGER] Issue #507 — filed as focused split from #505
+
+What is true now:
+- Issue state: open
+- Server-side decision state: server-side state changed
+- Local verdict/state: no server-side state changed before issue create
+- Latest known validation: duplicate search completed
+
+What changed:
+- new issue #507 created after duplicate assessment
+
+What is blocked:
+- Blocker classification: no blocker
+
+Who/what acts next:
+- Next actor: author
+- Required action: implement #507 two-comment validator
+- Do not do: recreate duplicate CTH issue
+- Resume from: issue #507 body
+""",
+    )
+)
\ No newline at end of file
diff --git a/thread_state_ledger_validator.py b/thread_state_ledger_validator.py
new file mode 100644
index 0000000..5dde2b1
--- /dev/null
+++ b/thread_state_ledger_validator.py
@@ -0,0 +1,399 @@
+"""Two-comment workflow validation for Controller Handoff + Thread State Ledger (#507).
+
+Validates the paired reporting pattern without replacing the CTH umbrella (#505)
+or the broader lifecycle ledger (#494/#495).
+"""
+
+from __future__ import annotations
+
+import re
+from typing import Any
+
+CONTROLLER_HANDOFF_TAG = "[CONTROLLER HANDOFF]"
+THREAD_STATE_LEDGER_TAG = "[THREAD STATE LEDGER]"
+
+BLOCKER_CLASSIFICATIONS = frozenset({
+    "code blocker",
+    "test blocker",
+    "merge conflict",
+    "stale head",
+    "permission/capability blocker",
+    "environment/tooling blocker",
+    "process/rule blocker",
+    "queue/lease blocker",
+    "duplicate/canonicalization blocker",
+    "no blocker",
+})
+
+_PRECISE_APPROVE_PHRASES = (
+    "approve verdict prepared locally",
+    "approved review posted to gitea",
+    "request_changes posted to gitea",
+)
+
+_PRECISE_MERGE_PHRASES = (
+    "merge performed",
+    "merge not performed",
+)
+
+_PRECISE_SERVER_PHRASES = (
+    "server-side state changed",
+    "no server-side state changed",
+)
+
+_AMBIGUOUS_STANDALONE_TERMS = (
+    "approved",
+    "ready",
+    "queued",
+    "blocked",
+    "done",
+    "merged",
+    "submitted",
+    "validated",
+)
+
+_HANDOFF_TAG_RE = re.compile(r"\[CONTROLLER\s+HANDOFF\]", re.IGNORECASE)
+_LEDGER_TAG_RE = re.compile(r"\[THREAD\s+STATE\s+LEDGER\]", re.IGNORECASE)
+_LEGACY_HANDOFF_RE = re.compile(r"^##\s*Controller Handoff\s*$", re.I | re.M)
+
+_MUTATION_LEDGER_RE = re.compile(
+    r"server-side mutation ledger\s*:",
+    re.IGNORECASE,
+)
+_BLOCKER_CLASS_RE = re.compile(
+    r"blocker\s+classification\s*:\s*(.+)",
+    re.IGNORECASE,
+)
+
+_LEDGER_REQUIRED_SECTIONS = (
+    "what is true now",
+    "what changed",
+    "what is blocked",
+    "who/what acts next",
+)
+
+_LEDGER_REQUIRED_FIELDS = (
+    "server-side decision state",
+    "local verdict/state",
+    "next actor",
+    "required action",
+)
+
+
+def _lower(text: str) -> str:
+    return (text or "").lower()
+
+
+def has_controller_handoff_marker(text: str) -> bool:
+    return bool(_HANDOFF_TAG_RE.search(text or ""))
+
+
+def has_thread_state_ledger_marker(text: str) -> bool:
+    return bool(_LEDGER_TAG_RE.search(text or ""))
+
+
+def _has_tagged_handoff(text: str) -> bool:
+    return has_controller_handoff_marker(text)
+
+
+def _has_tagged_ledger(text: str) -> bool:
+    return has_thread_state_ledger_marker(text)
+
+
+def _has_legacy_handoff(text: str) -> bool:
+    return bool(_LEGACY_HANDOFF_RE.search(text or ""))
+
+
+def _section_after_tag(text: str, tag_pattern: re.Pattern[str]) -> str | None:
+    text = text or ""
+    match = tag_pattern.search(text)
+    if not match:
+        return None
+    start = match.start()
+    other_tags = (
+        _HANDOFF_TAG_RE,
+        _LEDGER_TAG_RE,
+    )
+    end = len(text)
+    for other in other_tags:
+        if other.pattern == tag_pattern.pattern:
+            continue
+        later = other.search(text, match.end())
+        if later:
+            end = min(end, later.start())
+    return text[start:end]
+
+
+def _extract_mutation_ledger(text: str) -> str:
+    lines = (text or "").splitlines()
+    capture = False
+    chunks: list[str] = []
+    for line in lines:
+        stripped = line.strip().lower()
+        if "server-side mutation ledger" in stripped:
+            capture = True
+            continue
+        if capture:
+            if stripped.startswith("local-only") or stripped.startswith("blockers:"):
+                break
+            chunks.append(line)
+    return "\n".join(chunks)
+
+
+def _ambiguous_term_violations(text: str, *, scoped: bool) -> list[str]:
+    if not scoped:
+        return []
+    reasons: list[str] = []
+    lower = _lower(text)
+    for term in _AMBIGUOUS_STANDALONE_TERMS:
+        if not re.search(rf"\b{re.escape(term)}\b", lower):
+            continue
+        if term == "approved":
+            if any(p in lower for p in _PRECISE_APPROVE_PHRASES):
+                continue
+            if "review decision:" in lower and "approve" in lower:
+                continue
+            if "approval_at_current_head" in lower:
+                continue
+        elif term == "merged":
+            if any(p in lower for p in _PRECISE_MERGE_PHRASES):
+                continue
+            if "merge result:" in lower:
+                continue
+        elif term == "blocked":
+            if any(c in lower for c in BLOCKER_CLASSIFICATIONS):
+                continue
+            if "blocker classification:" in lower:
+                continue
+        reasons.append(
+            f"ambiguous standalone term '{term}' without precise state language"
+        )
+    return reasons
+
+
+def _mutation_contradiction_reasons(handoff_text: str) -> list[str]:
+    reasons: list[str] = []
+    lower = _lower(handoff_text)
+    ledger = _lower(_extract_mutation_ledger(handoff_text))
+    no_server = (
+        "no server-side state changed" in ledger
+        or "none — no server-side" in ledger
+        or re.search(r"\bnone\b", ledger)
+    )
+    claims_posted = "approved review posted to gitea" in lower
+    claims_merge = "merge performed" in lower or (
+        re.search(r"\bmerged\b", lower) and "merge not performed" not in lower
+    )
+    if claims_posted and no_server:
+        reasons.append(
+            "narrative claims APPROVED review posted but mutation ledger "
+            "shows no server-side state changed"
+        )
+    if claims_merge and no_server:
+        reasons.append(
+            "narrative claims merge but mutation ledger lacks merge proof"
+        )
+    if (
+        "approved review posted to gitea" in lower
+        and "no server-side state changed" in ledger
+    ):
+        reasons.append(
+            "mutation ledger contradicts APPROVED review posted claim"
+        )
+    return reasons
+
+
+def _blocked_without_gate_reasons(handoff_text: str, ledger_text: str) -> list[str]:
+    reasons: list[str] = []
+    combined = _lower(handoff_text) + "\n" + _lower(ledger_text)
+    if not re.search(r"\bblocked\b", combined):
+        return reasons
+    has_classification = bool(_BLOCKER_CLASS_RE.search(ledger_text or ""))
+    has_gate = any(
+        marker in combined
+        for marker in (
+            "exact gate",
+            "failing gate",
+            "blocker classification:",
+            "process/rule blocker",
+            "environment/tooling blocker",
+            "permission/capability blocker",
+        )
+    )
+    if not has_classification and not has_gate:
+        reasons.append(
+            "handoff or ledger says blocked but does not identify exact "
+            "failing gate or blocker classification"
+        )
+    return reasons
+
+
+def _local_server_separation_reasons(ledger_text: str) -> list[str]:
+    reasons: list[str] = []
+    lower = _lower(ledger_text)
+    server_line = ""
+    local_line = ""
+    for line in (ledger_text or "").splitlines():
+        stripped = line.strip().lower()
+        if stripped.startswith("- server-side decision state:"):
+            server_line = stripped
+        if stripped.startswith("- local verdict/state:"):
+            local_line = stripped
+    if not server_line or not local_line:
+        return reasons
+    if "approve verdict prepared locally" in server_line:
+        reasons.append(
+            "local-only verdict incorrectly listed under server-side "
+            "decision state"
+        )
+    if "approved review posted to gitea" in local_line:
+        reasons.append(
+            "server-side decision incorrectly listed under local verdict/state"
+        )
+    return reasons
+
+
+def _ledger_field_reasons(ledger_text: str) -> list[str]:
+    reasons: list[str] = []
+    lower = _lower(ledger_text)
+    for section in _LEDGER_REQUIRED_SECTIONS:
+        if section not in lower:
+            reasons.append(f"thread state ledger missing section '{section}'")
+    for field in _LEDGER_REQUIRED_FIELDS:
+        if field not in lower:
+            reasons.append(f"thread state ledger missing field '{field}'")
+    match = _BLOCKER_CLASS_RE.search(ledger_text or "")
+    if not match:
+        reasons.append("thread state ledger missing blocker classification")
+    else:
+        value = match.group(1).strip().lower().rstrip(".")
+        if value not in BLOCKER_CLASSIFICATIONS:
+            reasons.append(
+                f"blocker classification '{value}' is not a recognized value"
+            )
+    if "do not do:" not in lower:
+        reasons.append(
+            "thread state ledger missing explicit 'Do not do' guidance"
+        )
+    return reasons
+
+
+def _assessment(
+    reasons: list[str],
+    *,
+    performed: bool = True,
+) -> dict[str, Any]:
+    return {
+        "proven": not reasons,
+        "block": bool(reasons),
+        "reasons": reasons,
+        "performed": performed,
+    }
+
+
+def assess_controller_handoff_comment(body: str) -> dict[str, Any]:
+    """Validate a standalone ``[CONTROLLER HANDOFF]`` comment body."""
+    text = body or ""
+    if not _has_tagged_handoff(text):
+        return _assessment([], performed=False)
+    reasons: list[str] = []
+    if not _MUTATION_LEDGER_RE.search(text):
+        reasons.append(
+            "controller handoff missing 'Server-side mutation ledger' section"
+        )
+    reasons.extend(_ambiguous_term_violations(text, scoped=True))
+    reasons.extend(_mutation_contradiction_reasons(text))
+    return _assessment(reasons)
+
+
+def assess_thread_state_ledger_comment(body: str) -> dict[str, Any]:
+    """Validate a standalone ``[THREAD STATE LEDGER]`` comment body."""
+    text = body or ""
+    if not _has_tagged_ledger(text):
+        return _assessment([], performed=False)
+    reasons = _ledger_field_reasons(text)
+    reasons.extend(_ambiguous_term_violations(text, scoped=True))
+    reasons.extend(_local_server_separation_reasons(text))
+    return _assessment(reasons)
+
+
+def assess_two_comment_pair(
+    handoff_body: str,
+    ledger_body: str,
+) -> dict[str, Any]:
+    """Validate paired Gitea comments (handoff then ledger)."""
+    reasons: list[str] = []
+    handoff = handoff_body or ""
+    ledger = ledger_body or ""
+    if _has_tagged_handoff(handoff) and not _has_tagged_ledger(ledger):
+        reasons.append(
+            "controller handoff posted without follow-up THREAD STATE LEDGER"
+        )
+    handoff_result = assess_controller_handoff_comment(handoff)
+    ledger_result = assess_thread_state_ledger_comment(ledger)
+    reasons.extend(handoff_result.get("reasons") or [])
+    reasons.extend(ledger_result.get("reasons") or [])
+    reasons.extend(_blocked_without_gate_reasons(handoff, ledger))
+    if handoff_result.get("performed") or ledger_result.get("performed"):
+        performed = True
+    else:
+        performed = False
+    return _assessment(reasons, performed=performed)
+
+
+def assess_two_comment_workflow(report_text: str) -> dict[str, Any]:
+    """Validate combined final-report text containing both comment types."""
+    text = report_text or ""
+    if not (_has_tagged_handoff(text) or _has_tagged_ledger(text)):
+        if _has_legacy_handoff(text):
+            return _assessment([], performed=False)
+        return _assessment([], performed=False)
+
+    reasons: list[str] = []
+    if _has_tagged_handoff(text) and not _has_tagged_ledger(text):
+        reasons.append(
+            "tagged controller handoff present without THREAD STATE LEDGER"
+        )
+
+    handoff_section = _section_after_tag(text, _HANDOFF_TAG_RE) or text
+    ledger_section = _section_after_tag(text, _LEDGER_TAG_RE) or ""
+
+    reasons.extend(
+        (assess_controller_handoff_comment(handoff_section).get("reasons") or [])
+    )
+    if ledger_section:
+        reasons.extend(
+            (assess_thread_state_ledger_comment(ledger_section).get("reasons") or [])
+        )
+    reasons.extend(_blocked_without_gate_reasons(handoff_section, ledger_section))
+    reasons.extend(_mutation_contradiction_reasons(handoff_section))
+    reasons.extend(_local_server_separation_reasons(ledger_section))
+
+    return _assessment(reasons)
+
+
+def findings_for_final_report(report_text: str) -> list[dict[str, str]]:
+    """Return final_report_validator findings for the two-comment workflow."""
+    from final_report_validator import validator_finding
+
+    if not (_has_tagged_handoff(report_text) or _has_tagged_ledger(report_text)):
+        return []
+
+    result = assess_two_comment_workflow(report_text)
+    if not result.get("reasons"):
+        return []
+
+    findings: list[dict[str, str]] = []
+    for reason in result.get("reasons") or []:
+        severity = "block" if result.get("block") else "downgrade"
+        findings.append(
+            validator_finding(
+                "shared.two_comment_workflow",
+                severity,
+                "Thread State Ledger",
+                reason,
+                "add a [THREAD STATE LEDGER] after [CONTROLLER HANDOFF] "
+                "with precise server-side vs local state language",
+            )
+        )
+    return findings
\ No newline at end of file