fix(quota): fence selected todo content freshness

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

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)
     )

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"

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")
     }

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)

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)

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,

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

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;

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",

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"]

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.",

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)