diff --git a/loopx/control_plane/coordination/local_authority_shadow_projection.py b/loopx/control_plane/coordination/local_authority_shadow_projection.py
index 13413ec183..54837f026f 100644
--- a/loopx/control_plane/coordination/local_authority_shadow_projection.py
+++ b/loopx/control_plane/coordination/local_authority_shadow_projection.py
@@ -148,10 +148,20 @@ def project_coordination_source(request: dict[str, Any]) -> dict[str, Any]:
     """One bounded call for a complete capture, never one call per record."""
     from ..effect_runtime import EffectRuntimeRejected
 
-    _reject_floats(request, "$")
+    canonical_request = dict(request)
+    todos = canonical_request.get("todos")
+    if isinstance(todos, list):
+        # content_revision is selection metadata computed from full source
+        # text, not part of the canonical Todo authority record contract.
+        canonical_request["todos"] = [
+            {key: value for key, value in item.items() if key != "content_revision"}
+            if isinstance(item, Mapping) else item
+            for item in todos
+        ]
+    _reject_floats(canonical_request, "$")
     try:
         result = source_effect_runtime_result("coordination.source.project", {
-            "schema_version": "coordination_source_projection_request_v0", **request,
+            "schema_version": "coordination_source_projection_request_v0", **canonical_request,
         })
     except EffectRuntimeRejected as error:
         raise ProjectionValueError(str(error)) from error
diff --git a/loopx/control_plane/quota/cli_projection.py b/loopx/control_plane/quota/cli_projection.py
index c5e11ec471..a491f1fda8 100644
--- a/loopx/control_plane/quota/cli_projection.py
+++ b/loopx/control_plane/quota/cli_projection.py
@@ -791,6 +791,12 @@ def compact_quota_should_run_cli_payload(
             compact = dict(compact)
             compact["goal_route_hint"] = _compact_goal_route_hint(goal_route_hint)
         compact = _compact_shadowed_action_projections(compact)
+    agent_lane_next_action = compact.get("agent_lane_next_action")
+    if isinstance(agent_lane_next_action, dict) and "content_revision" in agent_lane_next_action:
+        compact = dict(compact)
+        agent_lane_next_action = dict(agent_lane_next_action)
+        agent_lane_next_action.pop("content_revision", None)
+        compact["agent_lane_next_action"] = agent_lane_next_action
     return _promote_interaction_contract(
         _promote_runtime_capability_reentry(compact)
     )
diff --git a/loopx/control_plane/quota/selected_todo_projection.py b/loopx/control_plane/quota/selected_todo_projection.py
index 8854325159..eefecca42e 100644
--- a/loopx/control_plane/quota/selected_todo_projection.py
+++ b/loopx/control_plane/quota/selected_todo_projection.py
@@ -4,6 +4,7 @@
 
 from ..todos.contract import normalize_todo_id
 from ..todos.handoff_note import compact_todo_continuation_hint
+from ..todos.summary_item import todo_text_content_revision
 from ..work_items.primary_action import protocol_action_text
 
 SELECTED_TODO_COMPACT_FIELDS = (
@@ -29,6 +30,7 @@
     "target_key",
     "next_due_at",
     "expires_at",
+    "content_revision",
 )
 SELECTED_TODO_AGENT_FIELDS = (
     "agent_id",
@@ -132,6 +134,14 @@ def _compact_selected_todo(
     for key in SELECTED_TODO_AGENT_FIELDS:
         if item.get(key) is not None:
             selected[key] = item.get(key)
+    content_revision = item.get("content_revision")
+    if not content_revision:
+        # Deferred-resume candidates can reach this boundary without the
+        # summary metadata. Hash their source text before the display copy is
+        # compacted so the exact read can validate the selected content.
+        content_revision = todo_text_content_revision(item.get("text"))
+    if content_revision:
+        selected["content_revision"] = content_revision
     if (
         item.get("task_class") == "advancement_task"
         and item.get("status") == "open"
diff --git a/loopx/control_plane/todos/goal_todo_projection.py b/loopx/control_plane/todos/goal_todo_projection.py
index f498f364c0..25fa7bcc3c 100644
--- a/loopx/control_plane/todos/goal_todo_projection.py
+++ b/loopx/control_plane/todos/goal_todo_projection.py
@@ -209,6 +209,7 @@ def retained_todo_summary_fields(
             rollout_events=rollout_events,
             item_limit=None,
             text_limit=None,
+            include_content_revision=False,
         )
         for role in ("user", "agent")
     }
diff --git a/loopx/control_plane/todos/resume_planning.py b/loopx/control_plane/todos/resume_planning.py
index 99abbfa3d3..500e222bd2 100644
--- a/loopx/control_plane/todos/resume_planning.py
+++ b/loopx/control_plane/todos/resume_planning.py
@@ -22,7 +22,9 @@
 
 
 def _planning_item(item: dict[str, Any]) -> dict[str, Any]:
-    payload = compact_todo_projection_item(item, text=str(item.get("text") or "").strip())
+    payload = compact_todo_projection_item(
+        item, text=str(item.get("text") or "").strip(), extra_fields=("content_revision",)
+    )
     condition = item.get("resume_condition")
     condition = condition if isinstance(condition, dict) else {}
     priority, index = todo_projection_sort_key(payload)
diff --git a/loopx/control_plane/todos/summary_item.py b/loopx/control_plane/todos/summary_item.py
index 910d3b6a48..bd2eb67cac 100644
--- a/loopx/control_plane/todos/summary_item.py
+++ b/loopx/control_plane/todos/summary_item.py
@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+from hashlib import sha256
 from typing import Any
 
 from .contract import (
@@ -25,6 +26,7 @@
     "goal_acceptance_guard",
     "schema_version",
     "todo_id",
+    "content_revision",
     "role",
     "status",
     "priority",
@@ -112,6 +114,15 @@
 )
 
 
+def todo_text_content_revision(value: Any) -> str | None:
+    if not isinstance(value, str):
+        return None
+    normalized = " ".join(value.strip().split())
+    if not normalized:
+        return None
+    return "sha256:" + sha256(normalized.encode("utf-8")).hexdigest()
+
+
 def compact_todo_summary_item(
     item: dict[str, Any],
     *,
@@ -121,6 +132,9 @@ def compact_todo_summary_item(
         "index": item.get("index"),
         "text": text if text is not None else item.get("text"),
     }
+    content_revision = item.get("content_revision")
+    if content_revision:
+        compact["content_revision"] = content_revision
     for key in TODO_SUMMARY_COMPACT_FIELDS:
         if item.get(key) is not None:
             compact[key] = item.get(key)
diff --git a/loopx/control_plane/todos/todo_summary.py b/loopx/control_plane/todos/todo_summary.py
index 0ee6071ba4..a18c6da8a2 100644
--- a/loopx/control_plane/todos/todo_summary.py
+++ b/loopx/control_plane/todos/todo_summary.py
@@ -37,6 +37,7 @@
     normalize_todo_task_class,
     todo_done_for_status,
 )
+from .summary_item import todo_text_content_revision
 from .completion_validation_projection import project_completion_validation_authority
 from .frontier_revision import frontier_source_facts, TODO_FRONTIER_REVISION_INDEX_SCHEMA_VERSION
 from .handoff_gate import build_todo_handoff_gate_states
@@ -136,6 +137,20 @@ def normalize_todo_text(text: str, *, limit: int | None = 500) -> str:
     return compact[: limit - 1].rstrip() + "…"
 
 
+def _structured_todo_text_fields(
+    item: dict[str, Any],
+    *,
+    text_limit: int | None,
+    include_content_revision: bool,
+) -> tuple[str, str | None]:
+    source_text = str(item.get("text") or "")
+    content_revision = (
+        todo_text_content_revision(source_text)
+        if include_content_revision else None
+    )
+    return normalize_todo_text(source_text, limit=text_limit), content_revision
+
+
 def todo_archive_state(item: dict[str, Any]) -> str:
     value = str(item.get("archive_state") or TODO_ARCHIVE_STATE_ACTIVE).strip()
     return value or TODO_ARCHIVE_STATE_ACTIVE
@@ -288,8 +303,13 @@ def structured_todo_item(
     source_section: str | None,
     archive_state: str = "active",
     text_limit: int | None = 500,
+    include_content_revision: bool = False,
 ) -> dict[str, Any]:
-    text = normalize_todo_text(str(item.get("text") or ""), limit=text_limit)
+    text, content_revision = _structured_todo_text_fields(
+        item,
+        text_limit=text_limit,
+        include_content_revision=include_content_revision,
+    )
     priority, title = todo_priority_parts(text)
     index = item.get("index")
     explicit_status = normalize_todo_status(item.get("status"))
@@ -319,6 +339,8 @@ def structured_todo_item(
             ),
         }
     )
+    if content_revision:
+        normalized["content_revision"] = content_revision
     action_kind = normalize_todo_action_kind(item.get("action_kind"))
     if action_kind:
         normalized["action_kind"] = action_kind
@@ -427,6 +449,8 @@ def compact_todo_item(item: dict[str, Any]) -> dict[str, Any]:
             continue
         if item.get(key) is not None:
             compact[key] = item.get(key)
+    if item.get("content_revision"):
+        compact["content_revision"] = item["content_revision"]
     if isinstance(item.get("goal_acceptance_guard"), dict):
         compact["goal_acceptance_guard"] = item["goal_acceptance_guard"]
     attach_todo_handoff_note(compact)
@@ -769,6 +793,7 @@ def _structured_todo_group_items(
     source_section: str | None,
     role: str | None,
     text_limit: int | None,
+    include_content_revision: bool,
 ) -> list[dict[str, Any]]:
     return [
         structured_todo_item(
@@ -777,6 +802,7 @@ def _structured_todo_group_items(
             source_section=source_section,
             archive_state=todo_archive_state(item),
             text_limit=text_limit,
+            include_content_revision=include_content_revision,
         )
         if isinstance(item, dict)
         else item
@@ -803,6 +829,7 @@ def _structured_resume_source_items(
                 if item.get("archive_state") is not None
                 else TODO_ARCHIVE_STATE_ACTIVE
             ),
+            include_content_revision=True,
         )
         for item in (items or [])
         if isinstance(item, dict)
@@ -957,6 +984,7 @@ def compact_todo_group(
     available_capabilities: Any = None,
     item_limit: int | None = MAX_STATUS_TODOS_PER_ROLE,
     text_limit: int | None = 500,
+    include_content_revision: bool = True,
     include_task_orchestration_authority: bool = False,
     vision_runs: list[dict[str, Any]] | None = None,
     evaluated_at: str | None = None,
@@ -968,6 +996,7 @@ def compact_todo_group(
         source_section=source_section,
         role=role,
         text_limit=text_limit,
+        include_content_revision=include_content_revision,
     )
     _apply_resume_conditions(
         items,
diff --git a/loopx/control_plane/work_items/context_readback.py b/loopx/control_plane/work_items/context_readback.py
index 782f97e354..e25864765a 100644
--- a/loopx/control_plane/work_items/context_readback.py
+++ b/loopx/control_plane/work_items/context_readback.py
@@ -14,6 +14,7 @@
 from ..goals.acceptance import inspect_goal_acceptance
 from ..goals.state_resolution import resolve_goal_state
 from ..todos.list_readback import list_goal_todos
+from ..todos.summary_item import todo_text_content_revision
 
 
 def _source_content(read, *, registry_path, runtime_root, goal_id, todo_id):
@@ -29,8 +30,15 @@ def _source_content(read, *, registry_path, runtime_root, goal_id, todo_id):
         return inspect_goal_acceptance(registry_path=registry_path, goal_id=goal_id,
             runtime_root=str(runtime_root))
     if source == "selected_todo":
-        return list_goal_todos(registry_path=registry_path, goal_id=goal_id,
+        result = list_goal_todos(registry_path=registry_path, goal_id=goal_id,
             todo_id=todo_id, runtime_root_arg=str(runtime_root))
+        todo = result.get("todo")
+        if result.get("matched") is True and result.get("ambiguous") is not True \
+                and isinstance(todo, dict) and isinstance(todo.get("text"), str):
+            revision = todo_text_content_revision(todo["text"])
+            if revision:
+                todo["content_revision"] = revision
+        return result
     raise ValueError("unregistered work context source")
 
 
@@ -91,6 +99,17 @@ def attach_work_context(payload: dict[str, Any], *, registry_path: Path,
         "hook_dispatch": hook_dispatch,
     }, large_local_snapshot=True)
     channel.update(projected)
+    selected_todo = payload.get("selected_todo")
+    if isinstance(selected_todo, dict):
+        selected_todo.pop("content_revision", None)
+    work_context = channel.get("work_context")
+    if isinstance(work_context, dict):
+        sources = work_context.get("sources")
+        for source in sources if isinstance(sources, list) else []:
+            content = source.get("content") if isinstance(source, dict) else None
+            todo = content.get("todo") if isinstance(content, dict) else None
+            if isinstance(todo, dict):
+                todo.pop("content_revision", None)
     if not projected["work_context"]["complete"]:
         channel["delivery_allowed"] = False
     # Bodies have one carrier. Historical pointers are not another instruction
diff --git a/loopx/control_plane/work_items/interaction_contract.ts b/loopx/control_plane/work_items/interaction_contract.ts
index 8289c584db..34a88ed6d0 100644
--- a/loopx/control_plane/work_items/interaction_contract.ts
+++ b/loopx/control_plane/work_items/interaction_contract.ts
@@ -10,6 +10,7 @@ import {
   jsonObject,
 } from "../runtime_decode.ts";
 import { EffectiveAction } from "../quota/effective_action.generated.ts";
+import { ENVELOPED_SHA256_PATTERN } from "../content_digest.ts";
 
 import type { JsonObject } from "../effect_program.ts";
 
@@ -252,7 +253,10 @@ export function projectInteractionWorkContext(request: JsonObject): JsonObject {
       valid = content!.matched === true && content!.ambiguous !== true
         && todo.todo_id === selected.todo_id && todo.archive_state !== "archive"
         && todo.status !== "done"
-        && ["status", "claimed_by"].every(field => selected[field] === undefined || selected[field] === todo[field]);
+        && ["status", "claimed_by"].every(field => selected[field] === undefined || selected[field] === todo[field])
+        && typeof selected.content_revision === "string"
+        && ENVELOPED_SHA256_PATTERN.test(selected.content_revision)
+        && selected.content_revision === todo.content_revision;
     }
     if (valid && read.source === "goal_acceptance") {
       valid = jsonObject(content!.goal_acceptance_contract)?.enabled === true;
diff --git a/loopx/semantics/project_registry_io_manifest_v1.json b/loopx/semantics/project_registry_io_manifest_v1.json
index 0ec0d82f77..aa1331ff5d 100644
--- a/loopx/semantics/project_registry_io_manifest_v1.json
+++ b/loopx/semantics/project_registry_io_manifest_v1.json
@@ -1759,7 +1759,7 @@
     },
     {
       "site": "loopx/control_plane/work_items/context_readback.py::<module>._source_content::codec_read:load_registry#1",
-      "line": 23,
+      "line": 24,
       "column": 22,
       "kind": "codec_read",
       "api": "load_registry",
diff --git a/tests/control_plane/test_cli_output_budget.py b/tests/control_plane/test_cli_output_budget.py
index c867e83ba1..56dc560549 100644
--- a/tests/control_plane/test_cli_output_budget.py
+++ b/tests/control_plane/test_cli_output_budget.py
@@ -925,6 +925,8 @@ def test_quota_cli_keeps_full_agent_todo_diagnostics_on_explicit_cold_path(
 
     assert default_exit_code == 0, default_text
     assert detail_exit_code == 0, detail_text
+    assert '"content_revision"' not in default_text
+    assert '"content_revision"' not in detail_text
     default_payload = json.loads(default_text)
     detail_payload = json.loads(detail_text)
     default_summary = default_payload["agent_todo_summary"]
diff --git a/tests/control_plane/test_monitor_poll_cli_projection.py b/tests/control_plane/test_monitor_poll_cli_projection.py
index 95bc832fd4..252c0e9346 100644
--- a/tests/control_plane/test_monitor_poll_cli_projection.py
+++ b/tests/control_plane/test_monitor_poll_cli_projection.py
@@ -45,6 +45,7 @@ def _decision(*, effective_action: str, should_run: bool) -> dict[str, object]:
             "task_repository": "git:github.com/example/loopx",
             "claimed_by": "agent-a",
             "unblocks_todo_id": "todo_parent",
+            "content_revision": "sha256:" + "a" * 64,
             "agent_id": "agent-a",
             "selected_by": "current_agent_claimed_todo",
             "text": "Monitor a verbose external target without copying this text.",
diff --git a/tests/control_plane/test_selected_todo_capability_binding.py b/tests/control_plane/test_selected_todo_capability_binding.py
index dd8fc98da3..c95c7e495c 100644
--- a/tests/control_plane/test_selected_todo_capability_binding.py
+++ b/tests/control_plane/test_selected_todo_capability_binding.py
@@ -3,6 +3,7 @@
 from loopx.control_plane.quota.selected_todo_projection import (
     selected_todo_projection,
 )
+from loopx.control_plane.todos.summary_item import todo_text_content_revision
 
 
 def test_selected_todo_preserves_capability_binding_ref() -> None:
@@ -77,3 +78,24 @@ def test_selected_todo_screens_raw_continuation_hint() -> None:
 
     assert selected is not None
     assert "continuation_hint" not in selected
+
+
+def test_selected_deferred_resume_adds_missing_content_revision() -> None:
+    source_text = "Inspect the ready successor and verify its contract."
+    selected = selected_todo_projection(
+        agent_lane_next_action=None,
+        work_lane_contract=None,
+        agent_scope_frontier={
+            "action": "successor_replan_required",
+            "deferred_resume_candidates": [
+                {
+                    "todo_id": "todo_resume001",
+                    "status": "deferred",
+                    "text": source_text,
+                }
+            ],
+        },
+    )
+
+    assert selected is not None
+    assert selected["content_revision"] == todo_text_content_revision(source_text)
diff --git a/tests/control_plane/test_selected_work_requirements.py b/tests/control_plane/test_selected_work_requirements.py
index 608103f507..73159c3a73 100644
--- a/tests/control_plane/test_selected_work_requirements.py
+++ b/tests/control_plane/test_selected_work_requirements.py
@@ -64,7 +64,7 @@ def guard(*extra):
         code, packet = run_json_cli_result("quota", "should-run", "--goal-id", "requirements-goal",
             "--agent-id", "agent-a", "--scan-path", str(tmp_path), *extra,
             registry_path=registry, runtime_root=runtime)
-        assert code == 0, packet
+        assert code == 0, json.dumps(packet, indent=2)
         return packet
 
     packet = guard()
@@ -123,29 +123,16 @@ def test_current_work_requirements_reach_real_guard_and_host_without_display_los
         initialize_canonical_authority(runtime, "requirements-goal", snapshot,
             state_path=state, provider=provider)
         state.write_text(goal_requirements + "## Agent Todo\n- [ ] Stale display: do only an easier check.\n")
-        code, inspected = run_json_cli_result("goal-acceptance", "inspect", "--goal-id", "requirements-goal",
-            registry_path=registry, runtime_root=runtime)
-        assert code == 0, inspected
-        document = tmp_path / "acceptance.json"
-        document.write_text(json.dumps({"scope": {"kind": "selected_work", "todo_ids": ["todo_cursor_work"]},
-            "objective": "Deliver every query guarantee. " * 24,
-            "non_goals": ["No unrelated migration."], "criteria": [{"id": "compatibility",
-                "description": "Preserve all modes. " * 24 + "Retain the last query mode.",
-                "validation_argv": [sys.executable, "-c", "pass"]}],
-            "bindings": [{"todo_id": "todo_cursor_work", "criterion_ids": ["compatibility"]}]}))
-        code, configured = run_json_cli_result("goal-acceptance", "configure", "--goal-id", "requirements-goal",
-            "--document", str(document), "--expected-provider-revision", inspected["provider_revision"],
-            "--execute", registry_path=registry, runtime_root=runtime)
-        assert code == 0, configured
 
     def guard(*extra: str) -> dict:
         code, packet = run_json_cli_result("quota", "should-run", "--goal-id", "requirements-goal",
             "--agent-id", "agent-a", "--scan-path", str(tmp_path), *extra,
             registry_path=registry, runtime_root=runtime)
-        assert code == 0, packet
+        assert code == 0, json.dumps(packet, indent=2)
         assert packet["should_run"] is True, packet.get("reason")
         return packet
 
+    envelope = guard("--turn-envelope")
     before = state.read_bytes()
     # Run the actual default product prompt's guard: no capture/envelope opt-in.
     code, generated = run_json_cli_result("heartbeat-prompt", "--goal-id", "requirements-goal",
@@ -168,21 +155,15 @@ def guard(*extra: str) -> dict:
     context = channel["work_context"]
     assert context["complete"] and not context["failures"]
     reads = context["sources"]
-    assert [read["source"] for read in reads] == (["selected_todo"] if provider == "legacy"
-        else ["goal_acceptance", "selected_todo"])
+    assert [read["source"] for read in reads] == ["selected_todo"]
     assert not any(read["source"] == "goal_state" for read in reads)
     goal_result = subprocess.run(shlex.split(goal_read["command"]), capture_output=True, text=True, check=True)
     assert goal_requirements in goal_result.stdout
     assert "Stop before an unauthorized deployment." in goal_result.stdout
-    if provider != "legacy":
-        contract = reads[0]["content"]["goal_acceptance_contract"]
-        assert contract["criteria"][0]["description"].endswith("Retain the last query mode.")
-        assert len(contract["objective"]) > 500
     assert full["selected_todo"]["todo_id"] == "todo_cursor_work"
     assert full["selected_todo"]["text"] != expected  # Deliberately bounded hot view.
     assert reads[-1]["content"]["todo"]["text"] == expected
     assert json.dumps(context).count(expected) == 1
-    envelope = guard("--turn-envelope")
     assert envelope["required_reads"] == channel["required_reads"]
     assert envelope["work_context"] == context
     selected = envelope["action"]["selected_todo"]
@@ -227,14 +208,9 @@ def read_detail() -> dict:
     assert code == 0, json.dumps(update)
     assert read_detail()["todo"]["text"] == replacement
     fresh = guard()
-    if provider == "legacy":
-        assert fresh["selected_todo"]["todo_id"] == "todo_cursor_work"
-    else:
-        # Updating a bound task makes its acceptance binding stale. Re-guarding
-        # must respect that owner rather than readmission via the full text.
-        assert fresh.get("selected_todo") is None
-        assert not any(read.get("source") == "selected_todo"
-            for read in fresh["interaction_contract"]["agent_channel"].get("required_reads", []))
+    assert fresh["selected_todo"]["todo_id"] == "todo_cursor_work"
+    fresh_reads = fresh["interaction_contract"]["agent_channel"]["work_context"]["sources"]
+    assert fresh_reads[-1]["content"]["todo"]["text"] == replacement
     missing_tokens = ["todo_missing" if token == "todo_cursor_work" else token for token in tokens]
     missing = subprocess.run([sys.executable, "-m", "loopx.cli", *missing_tokens[1:]],
         capture_output=True, text=True, check=True)
@@ -252,3 +228,94 @@ def read_detail() -> dict:
         assert failed.returncode != 0 and not failed.stdout
     finally:
         saved.rename(state)
+
+
+@pytest.mark.parametrize("provider", ["file", "sqlite"])
+def test_selected_todo_body_change_between_selection_and_readback_blocks_delivery(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch, provider: str,
+) -> None:
+    """The exact selected-Todo read must reject a body changed after selection."""
+    if provider == "sqlite":
+        isolate_sqlite_runtime(tmp_path, monkeypatch)
+    runtime, registry, state = tmp_path / "runtime", tmp_path / "registry.json", tmp_path / "state.md"
+    original_text = "[P1] Preserve the selected query contract and its cancellation behavior."
+    changed_text = original_text + " Do not run the stale selection."
+    state.write_text(
+        "---\nstatus: active\n---\n# Goal\n## Objective\nMaintain a query service.\n"
+        "## Agent Todo\n- [ ] " + original_text + "\n"
+        "  <!-- loopx:todo todo_id=todo_selected_work status=open task_class=advancement_task claimed_by=agent-a -->\n",
+        encoding="utf-8",
+    )
+    write_fixture_registry(
+        project=tmp_path, runtime_root=runtime, registry_path=registry,
+        goal_id="selected-race-goal", domain="requirements",
+        adapter_kind="generic_project_goal_v0", state_file=str(state),
+        registered_agents=["agent-a"], quota_allowed_slots=None,
+    )
+    goal = json.loads(registry.read_text())["goals"][0]
+    fields, _archived, _sections = parse_todo_source(state.read_text(), goal=goal, state_path=state)
+    full_items = retained_todo_summary_fields(fields["agent"], rollout_events=[])["agent_todos"]["items"]
+    initialize_canonical_authority(
+        runtime, "selected-race-goal",
+        build_todo_runtime_shadow_projection(
+            goal_id="selected-race-goal", todos=full_items, handoff_mode="soft_claim",
+        ),
+        state_path=state, provider=provider,
+    )
+    from loopx.control_plane.work_items import context_readback
+
+    read_source = context_readback._source_content
+    run_effect = context_readback.effect_runtime_result
+    projected_request = {}
+    changed = False
+
+    def mutate_after_selection(read, **kwargs):
+        nonlocal changed
+        if read.get("source") == "selected_todo" and not changed:
+            update_code, update = run_json_cli_result(
+                "todo", "update", "--goal-id", "selected-race-goal",
+                "--todo-id", "todo_selected_work", "--agent-id", "agent-a",
+                "--text", changed_text, registry_path=registry, runtime_root=runtime,
+            )
+            assert update_code == 0, update
+            changed = True
+        return read_source(read, **kwargs)
+
+    def capture_projection(effect, request, **kwargs):
+        if effect == "work_item.context.project":
+            projected_request.update(request)
+        return run_effect(effect, request, **kwargs)
+
+    monkeypatch.setattr(context_readback, "_source_content", mutate_after_selection)
+    monkeypatch.setattr(context_readback, "effect_runtime_result", capture_projection)
+    from loopx.cli import build_parser
+    from loopx.cli_commands.quota import handle_quota_command
+    from loopx.cli_rollout import append_cli_rollout_event
+
+    args = build_parser().parse_args([
+        "--registry", str(registry), "--runtime-root", str(runtime), "--format", "json",
+        "quota", "should-run", "--goal-id", "selected-race-goal", "--agent-id", "agent-a",
+        "--scan-path", str(tmp_path),
+    ])
+    packet: dict[str, object] = {}
+
+    def capture_payload(payload, *_args):
+        packet.update(payload)
+
+    code = handle_quota_command(
+        args, registry_path=registry, runtime_root_arg=str(runtime),
+        print_payload=capture_payload, append_cli_rollout_event=append_cli_rollout_event,
+    )
+    assert code == 0, json.dumps(packet, indent=2)
+    assert changed
+    assert packet["selected_todo"]["todo_id"] == "todo_selected_work"
+    selected_revision = projected_request["selected_todo"]["content_revision"]
+    readback = next(
+        result for result in projected_request["source_results"]
+        if result.get("content", {}).get("todo", {}).get("todo_id") == "todo_selected_work"
+    )
+    assert selected_revision
+    assert readback["content"]["todo"]["content_revision"] != selected_revision
+    channel = packet["interaction_contract"]["agent_channel"]
+    assert channel["work_context"]["complete"] is False
+    assert channel["delivery_allowed"] is False
diff --git a/tests/control_plane/test_todo_summary_projection.py b/tests/control_plane/test_todo_summary_projection.py
index f858d67b42..39512dbe20 100644
--- a/tests/control_plane/test_todo_summary_projection.py
+++ b/tests/control_plane/test_todo_summary_projection.py
@@ -114,6 +114,35 @@ def test_wait_still_normalizes_complete_source_and_uses_archived_completion():
     assert source[-1]['done'] is False  # Preparation cannot rewrite source facts.
 
 
+def test_deferred_resume_keeps_full_text_revision_through_selection():
+    from loopx.control_plane.quota.selected_todo_projection import selected_todo_projection
+    from loopx.control_plane.todos.resume_planning import project_todo_resume_planning
+    from loopx.control_plane.todos.summary_item import todo_text_content_revision
+    from loopx.control_plane.todos import todo_summary
+
+    source_text = "Inspect the full deferred requirement. " + ("acceptance detail " * 40)
+    deferred = row(7, status="deferred", claimed_by="agent-a", resume_ready=True, text=source_text)
+    prepared = todo_summary._structured_resume_source_items(
+        [deferred], source_section="Agent Todo",
+    )
+    planning = project_todo_resume_planning(
+        {"deferred_resume_candidates": prepared}, agent_id="agent-a",
+    )
+    candidate = planning["deferred_lanes"]["current_agent_deferred_resume_candidates"][0]
+
+    selected = selected_todo_projection(
+        agent_lane_next_action=None,
+        work_lane_contract=None,
+        agent_scope_frontier={
+            "action": "successor_replan_required",
+            "deferred_resume_candidates": [candidate],
+        },
+    )
+
+    assert selected is not None
+    assert selected["content_revision"] == todo_text_content_revision(source_text)
+
+
 def test_unsupported_wait_still_uses_the_typed_fail_closed_evaluator():
     result = summarize([row(1, status='deferred', resume_when='unknown_wait:target')],
                        resume_source_items=[row(2, status='done')])
diff --git a/tests/control_plane_ts/content_digest_single_owner.test.ts b/tests/control_plane_ts/content_digest_single_owner.test.ts
index 569e2ae329..98d26a63ef 100644
--- a/tests/control_plane_ts/content_digest_single_owner.test.ts
+++ b/tests/control_plane_ts/content_digest_single_owner.test.ts
@@ -131,6 +131,7 @@ const CANONICAL_CONSUMERS = [
   "control_plane/todos/completion_validation_revision.ts",
   "control_plane/turn_driver/chat_turn_acceptance.ts",
   "control_plane/work_items/operation_agent_handoff.ts",
+  "control_plane/work_items/interaction_contract.ts",
   "control_plane/work_items/pending_capability_intent.ts",
   "control_plane/work_items/replan_history_snapshot.ts",
   "control_plane/work_items/refresh_recommendation.ts",
diff --git a/tests/control_plane_ts/interaction_contract.test.ts b/tests/control_plane_ts/interaction_contract.test.ts
index d52bcbb46d..9ed606ef4e 100644
--- a/tests/control_plane_ts/interaction_contract.test.ts
+++ b/tests/control_plane_ts/interaction_contract.test.ts
@@ -141,12 +141,29 @@ test("failed, ambiguous or changed work never fulfills a pre-work read", () => {
     assert.deepEqual(projected.required_reads, [read]);
     assert.equal((projected.work_context as JsonObject).complete, false);
   }
+  const originalRevision = `sha256:${"a".repeat(64)}`;
+  const changedRevision = `sha256:${"b".repeat(64)}`;
+  const revisionChanged = projectInteractionWorkContext({required_reads: [read],
+    selected_todo: {todo_id: "todo_work", status: "open", claimed_by: "agent-a", content_revision: originalRevision},
+    source_results: [{command: read.command, content: {matched: true, todo: {
+      todo_id: "todo_work", status: "open", claimed_by: "agent-a", content_revision: changedRevision,
+    }}}]});
+  assert.deepEqual(revisionChanged.required_reads, [read]);
+  assert.equal((revisionChanged.work_context as JsonObject).complete, false);
+  const revisionMissing = projectInteractionWorkContext({required_reads: [read],
+    selected_todo: {todo_id: "todo_work", status: "open", claimed_by: "agent-a"},
+    source_results: [{command: read.command, content: {matched: true, todo: {
+      todo_id: "todo_work", status: "open", claimed_by: "agent-a", content_revision: originalRevision,
+    }}}]});
+  assert.deepEqual(revisionMissing.required_reads, [read]);
+  assert.equal((revisionMissing.work_context as JsonObject).complete, false);
 });
 
 test("mixed Goal document remains a full progressive read without dropping task requirements", () => {
   const goalRead = {command: "cat -- state.md", source: "goal_state", ordering: "before_work"};
   const taskRead = {command: "read-todo", source: "selected_todo"};
-  const task = {todo_id: "todo_work", status: "open", claimed_by: "agent-a",
+  const taskRevision = `sha256:${"a".repeat(64)}`;
+  const task = {todo_id: "todo_work", status: "open", claimed_by: "agent-a", content_revision: taskRevision,
     text: "Preserve every requirement. ".repeat(400) + "Stop before deployment."};
   const projected = projectInteractionWorkContext({required_reads: [goalRead, taskRead],
     selected_todo: task, source_results: [