diff --git a/docs/reference/todo-work-counts.md b/docs/reference/todo-work-counts.md
index 6d0386a631..a170b2bc61 100644
--- a/docs/reference/todo-work-counts.md
+++ b/docs/reference/todo-work-counts.md
@@ -138,6 +138,20 @@ A source already marked partial cannot regain `source_proof` or
 rows. Query scope and source completeness are independent conditions. These
 proofs remain read-only observations, not permission to settle a Goal.
 
+Quota validates retained closure witnesses through the existing
+`todo.quota_planning.project` batch, alongside selection and resume planning.
+The TS succession owner now validates the source and terminal proofs as well as
+producing them; Python transports compact source facts and materializes fields.
+The internal request is versioned to v2; v0/v1 readers retain their wire behavior.
+No extra RPC, provider setting or persisted schema is added.
+
+**Closure read correction:** boolean, null, string, fractional and negative
+counts cannot certify terminal closure. Equal malformed Monitor counts are
+invalid evidence, not proof that the two counts agree. Invalid evidence clears
+the derived no-followup intent instead of suppressing quota work. Genuine empty
+sources, complete proofs with bounded displays and watch-only Monitor closure
+remain supported. This does not itself complete or settle a Goal.
+
 This changes status, Todo-list and quota summary readback for both legacy and
 promoted Goals without a flag. Existing frontend and Lark views consume these
 Core projections; no new setting or frontend asset is required. No provider,
@@ -156,5 +170,12 @@ Python 继续负责旧数据解码、公开字段筛选、隐私处理与文本
 最近完成列表。后继缺口警告仍按最后更新时间排序，未知时间靠后，不丢弃警告。
 已有 partial 来源不会因为再次筛选命中所有可见行，就重新获得整个来源的收尾证明。
 
+quota 在原有 `todo.quota_planning.project` 批次中同时验证来源／收尾证明，不再由
+Python 独立判断。TS succession owner 负责生成和验证；Python 只传紧凑事实并还原展示。
+内部请求升级为 v2，保留 v0/v1 兼容。布尔、null、字符串、小数和负数不能充当计数，
+两个非法 Monitor 计数相同也不构成证明；非法证明不再产生 no-followup 停止意图。
+合法空来源、有界展示保留的完整来源证明及 watch-only Monitor 收尾保持原行为。
+没有新增 RPC、设置或持久化格式，这条读取规则本身不能完成或结算 Goal。
+
 覆盖 legacy 与 canonical 的 status、Todo 查询和 quota 摘要；展示预算保持原值。
 没有新增设置、权限或 writer，不改变 provider 默认值，也不宣称完成整 Goal 迁移。
diff --git a/loopx/control_plane/todos/quota_selection.py b/loopx/control_plane/todos/quota_selection.py
index 0525596cd8..dcb06b8a1d 100644
--- a/loopx/control_plane/todos/quota_selection.py
+++ b/loopx/control_plane/todos/quota_selection.py
@@ -22,6 +22,47 @@
 from .user_gate import is_user_gate_todo_item
 
 
+def _closure_source_facts(value: dict[str, Any]) -> dict[str, Any]:
+    # JSON erases Python's int/float distinction. Retain the strict integer
+    # evidence type before transport; the typed owner decides validity.
+    def integer(raw: Any) -> int | None:
+        return raw if type(raw) is int else None
+
+    def proof(key: str, fields: tuple[str, ...], counts: dict[str, int | None]) -> Any:
+        raw = value.get(key)
+        if not isinstance(raw, dict):
+            return None
+        return {**{name: raw.get(name) for name in fields},
+                **{name: integer(raw.get(name, default)) for name, default in counts.items()}}
+
+    def rows(key: str, *, absent_empty: bool = False) -> Any:
+        raw = value.get(key)
+        if raw is None and absent_empty:
+            raw = []
+        if not isinstance(raw, list):
+            return None
+        return [{"status": item.get("status"), "done": item.get("done"),
+                 "watch_only": todo_item_is_watch_only_monitor(item),
+                 "route_continuation_replan_required": item.get("route_continuation_replan_required")}
+                if isinstance(item, dict) else None for item in raw]
+
+    return {
+        **{key: value.get(key) for key in ("schema_version", "source_section")},
+        **{key: integer(value.get(key)) for key in ("total_count", "open_count", "done_count", "deferred_count")},
+        "convergence_open_count": integer(value.get("convergence_open_count", value.get("open_count"))),
+        **{key: integer(value.get(key, 0)) for key in ("completed_without_successor_count", "route_continuation_replan_count")},
+        "source_proof": proof("source_proof", ("schema_version", "role", "derived"), {"item_count": None}),
+        "terminal_closure_proof": proof("terminal_closure_proof",
+            ("schema_version", "role", "source_section", "all_todos_done", "all_convergent_todos_done", "derived"),
+            {"item_count": None, "monitor_open_count": None, "watch_only_monitor_count": 0,
+             "successor_gap_count": None, "route_replan_count": None, "no_followup_count": None}),
+        "closure_intent": proof("closure_intent", ("schema_version", "kind", "derived"), {"count": None}),
+        "items": rows("items"), "monitor_open_items": rows("monitor_open_items", absent_empty=True),
+        "deferred_item_count": len(value["deferred_items"]) if isinstance(value.get("deferred_items"), list) else None,
+        "deferred_resume_count": len(value["deferred_resume_candidates"]) if isinstance(value.get("deferred_resume_candidates"), list) else None,
+    }
+
+
 def project_quota_planning(
     value: dict[str, Any], *, all_open_items: list[dict[str, Any]],
     source_open_count: Any, agent_identity: dict[str, Any] | None,
@@ -62,7 +103,8 @@ def active(key: str) -> list[dict[str, Any]]:
 
     try:
         result = effect_runtime_result("todo.quota_planning.project", {
-            "schema_version": "todo_quota_planning_request_v1",
+            "schema_version": "todo_quota_planning_request_v2",
+            "source_contract": _closure_source_facts(value),
             "resume": build_todo_resume_planning_request(value, agent_id=agent, item_limit=8,
                 available_capabilities=(available_capabilities or []) if resolve_capacity else None),
             "selection": {
@@ -83,4 +125,10 @@ def active(key: str) -> list[dict[str, Any]]:
         raise ValueError(str(exc)) from None
     if not isinstance(result, dict) or result.get("schema_version") != "todo_quota_planning_v0":
         raise RuntimeError("TypeScript Todo quota planning shape mismatch")
+    if not isinstance(result.get("source_completeness"), dict) or "closure_intent" not in result:
+        raise RuntimeError("TypeScript Todo quota planning source contract missing")
+    if isinstance(result["closure_intent"], dict):
+        result["closure_intent"] = {**value["closure_intent"], **result["closure_intent"]}
+    elif result["closure_intent"] is not None:
+        raise RuntimeError("TypeScript Todo quota planning closure intent shape mismatch")
     return result
diff --git a/loopx/control_plane/todos/quota_selection.ts b/loopx/control_plane/todos/quota_selection.ts
index 5ccf5bd01b..9ec6f3d438 100644
--- a/loopx/control_plane/todos/quota_selection.ts
+++ b/loopx/control_plane/todos/quota_selection.ts
@@ -7,6 +7,7 @@ import { projectTodoResumePlanning } from "./resume_planning.ts";
 import { gateAddressesAgent, actionAddressesAgent, claimAllowsAgent } from "./agent_scope.ts";
 import { missingRequiredCapabilities } from "../agents/capability_gate.ts";
 import {claimedAdvancementCountFromIndex} from "./frontier_revision.ts";
+import {validateTodoClosureSource} from "./succession.ts";
 
 interface Row {
   payload: JsonObject; display: JsonObject; claim: string | null;
@@ -188,13 +189,15 @@ export function projectQuotaSelection(value: unknown): JsonObject {
   }, claim_visibility: claimVisibility};
 }
 
-/** One quota read boundary composes scope/claim selection and existing resume rules. */
+/** One quota read boundary composes scope/claim, resume and source closure rules. */
 export function projectTodoQuotaPlanning(value: unknown): JsonObject {
   const request = requireJsonObject(value, "quota planning");
-  if (!["todo_quota_planning_request_v0", "todo_quota_planning_request_v1"].includes(String(request.schema_version))) throw new EffectRuntimeRequestError("quota planning schema mismatch");
-  if (request.schema_version === "todo_quota_planning_request_v1") {
+  if (!["todo_quota_planning_request_v0", "todo_quota_planning_request_v1", "todo_quota_planning_request_v2"].includes(String(request.schema_version))) throw new EffectRuntimeRequestError("quota planning schema mismatch");
+  if (request.schema_version !== "todo_quota_planning_request_v0") {
     requireStringArray(requireJsonObject(request.selection, "selection").available, "available");
   }
+  const closure = request.schema_version === "todo_quota_planning_request_v2"
+    ? validateTodoClosureSource(request.source_contract) : {};
   return {schema_version: "todo_quota_planning_v0", resume_planning: projectTodoResumePlanning(request.resume),
-    ...projectQuotaSelection(request.selection)};
+    ...projectQuotaSelection(request.selection), ...closure};
 }
diff --git a/loopx/control_plane/todos/quota_summary.py b/loopx/control_plane/todos/quota_summary.py
index bdf94ffd5d..46c8aef265 100644
--- a/loopx/control_plane/todos/quota_summary.py
+++ b/loopx/control_plane/todos/quota_summary.py
@@ -13,7 +13,6 @@
 from .frontier_deadline import todo_summary_frontier_deadline
 from .handoff_gate import build_todo_handoff_gate_lanes
 from .todo_semantics import (
-    todo_item_is_watch_only_monitor,
     todo_item_task_class,
     todo_presentation_sort_key,
     todo_summary_monitor_schedule_gap_items,
@@ -194,146 +193,6 @@ class _QuotaTodoLanes:
     open_count: Any
 
 
-def _strict_non_negative_int(value: Any) -> int | None:
-    if type(value) is not int or value < 0:
-        return None
-    return value
-
-
-def _terminal_closure_proof_is_valid(
-    value: dict[str, Any],
-    *,
-    counts: dict[str, int | None],
-    source_proof: dict[str, Any],
-) -> bool:
-    proof = value.get("terminal_closure_proof")
-    items = value.get("items")
-    total_count = counts["total_count"]
-    displayed_items_cover_source = bool(
-        isinstance(total_count, int)
-        and (
-            (total_count == 0 and items == [])
-            or (
-                total_count > 0
-                and isinstance(items, list)
-                and 0 < len(items) <= total_count
-            )
-        )
-    )
-    return bool(
-        value.get("schema_version") == "todo_summary_v0"
-        and isinstance(items, list)
-        and displayed_items_cover_source
-        and all(
-            isinstance(item, dict)
-            and (
-                (item.get("status") == "done" and item.get("done") is True)
-                or (
-                    todo_item_is_watch_only_monitor(item)
-                )
-            )
-            and item.get("route_continuation_replan_required") is not True
-            for item in items
-        )
-        and all(
-            isinstance(item, dict)
-            and todo_item_is_watch_only_monitor(item)
-            for item in (value.get("monitor_open_items") or [])
-        )
-        and value.get("deferred_items") == []
-        and value.get("deferred_resume_candidates") == []
-        and _strict_non_negative_int(
-            value.get("convergence_open_count", value.get("open_count"))
-        ) == 0
-        and _strict_non_negative_int(value.get("completed_without_successor_count", 0)) == 0
-        and _strict_non_negative_int(value.get("route_continuation_replan_count", 0)) == 0
-        and isinstance(proof, dict)
-        and proof.get("schema_version") == "todo_terminal_closure_proof_v0"
-        and proof.get("role") == source_proof.get("role")
-        and proof.get("source_section") == value.get("source_section")
-        and proof.get("item_count") == total_count
-        and (
-            proof.get("all_todos_done") is True
-            or proof.get("all_convergent_todos_done") is True
-        )
-        and _strict_non_negative_int(proof.get("monitor_open_count"))
-        == _strict_non_negative_int(proof.get("watch_only_monitor_count", 0))
-        and _strict_non_negative_int(proof.get("successor_gap_count")) == 0
-        and _strict_non_negative_int(proof.get("route_replan_count")) == 0
-        and _strict_non_negative_int(proof.get("no_followup_count")) is not None
-        and proof.get("derived") is True
-    )
-
-
-def validate_todo_source_contract(
-    value: dict[str, Any],
-) -> tuple[dict[str, Any], dict[str, Any] | None]:
-    proof = value.get("source_proof")
-    counts = {
-        key: _strict_non_negative_int(value.get(key))
-        for key in ("total_count", "open_count", "done_count", "deferred_count")
-    }
-    total_count = counts["total_count"]
-    open_count = counts["open_count"]
-    done_count = counts["done_count"]
-    deferred_count = counts["deferred_count"]
-    valid_counts = (
-        total_count is not None
-        and open_count is not None
-        and done_count is not None
-        and deferred_count is not None
-        and total_count == open_count + done_count + deferred_count
-    )
-    valid_proof = bool(
-        isinstance(proof, dict)
-        and proof.get("schema_version") == "todo_source_proof_v0"
-        and proof.get("role") in {"user", "agent"}
-        and proof.get("derived") is True
-        and bool(str(value.get("source_section") or "").strip())
-        and type(proof.get("item_count")) is int
-        and proof.get("item_count") == total_count
-    )
-    valid_terminal_closure = bool(
-        valid_counts
-        and valid_proof
-        and isinstance(proof, dict)
-        and _terminal_closure_proof_is_valid(
-            value,
-            counts=counts,
-            source_proof=proof,
-        )
-    )
-    completeness = {
-        "schema_version": "todo_source_completeness_v0",
-        "status": "valid" if valid_terminal_closure else "invalid",
-        "source": "structured_todo_projection",
-        "role": proof.get("role") if isinstance(proof, dict) else None,
-        "terminal_closure": "valid" if valid_terminal_closure else "invalid",
-    }
-
-    intent = value.get("closure_intent")
-    terminal_proof = value.get("terminal_closure_proof")
-    intent_count = intent.get("count") if isinstance(intent, dict) else None
-    valid_intent = bool(
-        valid_terminal_closure
-        and isinstance(intent, dict)
-        and intent.get("schema_version") == "todo_closure_intent_v0"
-        and intent.get("kind") == "no_followup"
-        and intent.get("derived") is True
-        and type(intent_count) is int
-        and done_count is not None
-        and 0 < intent_count <= done_count
-        and isinstance(terminal_proof, dict)
-        and intent_count == terminal_proof.get("no_followup_count")
-    )
-    closure_intent = (
-        {**intent, "source": "todo_no_followup"}
-        if valid_intent and isinstance(intent, dict)
-        else None
-    )
-    return completeness, closure_intent
-
-
 def summarize_user_todos_for_quota(
     value: Any,
     *,
@@ -344,7 +203,6 @@ def summarize_user_todos_for_quota(
 ) -> dict[str, Any] | None:
     if not isinstance(value, dict):
         return None
-    source_completeness, closure_intent = validate_todo_source_contract(value)
     all_open_items = sorted(
         todo_summary_source_items(value),
         key=todo_presentation_sort_key,
@@ -399,7 +257,7 @@ def summarize_user_todos_for_quota(
         "work_counts": planning["work_counts"],
         "done_count": value.get("done_count"),
         "deferred_count": value.get("deferred_count"),
-        "source_completeness": source_completeness,
+        "source_completeness": planning["source_completeness"],
         "first_open_items": lanes.display_open_items[:3],
         "first_executable_items": lanes.executable_items[:3],
         "gate_open_items": gate_items[:3],
@@ -449,8 +307,8 @@ def summarize_user_todos_for_quota(
         summary["current_agent_blocker_items"] = current_agent_blocker_items[
             :QUOTA_PAYLOAD_DIAGNOSTIC_LANE_LIMIT
         ]
-    if closure_intent:
-        summary["closure_intent"] = closure_intent
+    if planning["closure_intent"]:
+        summary["closure_intent"] = planning["closure_intent"]
     monitor_writeback = todo_summary_monitor_writeback_contract(value)
     if monitor_writeback:
         summary["monitor_writeback"] = monitor_writeback
diff --git a/loopx/control_plane/todos/succession.ts b/loopx/control_plane/todos/succession.ts
index 97d6a98c0c..d282c08759 100644
--- a/loopx/control_plane/todos/succession.ts
+++ b/loopx/control_plane/todos/succession.ts
@@ -4,7 +4,7 @@ import wire from "./succession_wire_v1.json" with {type: "json"};
 import {canonicalAuthoritySha256} from "../coordination/authority_store_codec.ts";
 import {EffectRuntimeRequestError} from "../effect_runtime_errors.ts";
 import type {JsonObject} from "../effect_program.ts";
-import {requireBoolean, requireJsonObject, requireStringLiteral} from "../runtime_decode.ts";
+import {jsonObject, requireBoolean, requireJsonObject, requireStringLiteral} from "../runtime_decode.ts";
 
 export const HANDOFF_STATES = ["blocking", "cleared_without_successor", "cleared_with_successor",
   "cleared_no_followup", "superseded", "deferred"] as const;
@@ -209,3 +209,45 @@ export function projectTodoClosure(value: unknown): JsonObject {
   if (noFollowup) result.closure_intent = {schema_version: "todo_closure_intent_v0", kind: "no_followup", derived: true, count: noFollowup};
   return result;
 }
+
+/** Validate retained source witnesses at the quota read boundary. Malformed
+ * evidence is an invalid observation, never an exception or closure authority.
+ * A bounded display may retain a full-source proof but cannot manufacture it. */
+export function validateTodoClosureSource(value: unknown): JsonObject {
+  const source = requireJsonObject(value, "Todo closure source");
+  const proof = jsonObject(source.source_proof), terminal = jsonObject(source.terminal_closure_proof);
+  const count = (value: unknown): value is number => typeof value === "number" && Number.isSafeInteger(value) && value >= 0;
+  const total = source.total_count, open = source.open_count, done = source.done_count, deferred = source.deferred_count;
+  const countsValid = count(total) && count(open) && count(done) && count(deferred) && total === open + done + deferred;
+  const sourceValid = proof !== null && proof.schema_version === "todo_source_proof_v0" &&
+    (proof.role === "user" || proof.role === "agent") && proof.derived === true &&
+    typeof source.source_section === "string" && source.source_section.trim() !== "" &&
+    count(proof.item_count) && proof.item_count === total;
+  const rows = source.items, monitors = source.monitor_open_items;
+  const covered = count(total) && Array.isArray(rows) &&
+    (total === 0 ? rows.length === 0 : rows.length > 0 && rows.length <= total);
+  const rowsValid = Array.isArray(rows) && rows.every(value => {
+    const row = jsonObject(value);
+    return row !== null && ((row.status === "done" && row.done === true) || row.watch_only === true) &&
+      row.route_continuation_replan_required !== true;
+  });
+  const monitorsValid = Array.isArray(monitors) && monitors.every(value => jsonObject(value)?.watch_only === true);
+  const terminalValid = countsValid && sourceValid && source.schema_version === "todo_summary_v0" &&
+    covered && rowsValid && monitorsValid && source.deferred_item_count === 0 && source.deferred_resume_count === 0 &&
+    source.convergence_open_count === 0 && source.completed_without_successor_count === 0 && source.route_continuation_replan_count === 0 &&
+    terminal !== null && terminal.schema_version === "todo_terminal_closure_proof_v0" && terminal.role === proof!.role &&
+    terminal.source_section === source.source_section && count(terminal.item_count) && terminal.item_count === total &&
+    (terminal.all_todos_done === true || terminal.all_convergent_todos_done === true) &&
+    count(terminal.monitor_open_count) && count(terminal.watch_only_monitor_count) &&
+    terminal.monitor_open_count === terminal.watch_only_monitor_count &&
+    terminal.successor_gap_count === 0 && terminal.route_replan_count === 0 &&
+    count(terminal.no_followup_count) && terminal.derived === true;
+  const intent = jsonObject(source.closure_intent);
+  const intentValid = terminalValid && intent !== null && intent.schema_version === "todo_closure_intent_v0" &&
+    intent.kind === "no_followup" && intent.derived === true && count(intent.count) && count(done) &&
+    intent.count > 0 && intent.count <= done && intent.count === terminal!.no_followup_count;
+  return {source_completeness: {schema_version: "todo_source_completeness_v0",
+    status: terminalValid ? "valid" : "invalid", source: "structured_todo_projection",
+    role: proof?.role ?? null, terminal_closure: terminalValid ? "valid" : "invalid"},
+    closure_intent: intentValid ? {...intent, source: "todo_no_followup"} : null};
+}
diff --git a/tests/control_plane/test_todo_closure_source_contract.py b/tests/control_plane/test_todo_closure_source_contract.py
new file mode 100644
index 0000000000..2345945c04
--- /dev/null
+++ b/tests/control_plane/test_todo_closure_source_contract.py
@@ -0,0 +1,64 @@
+from __future__ import annotations
+
+from copy import deepcopy
+
+import pytest
+
+from loopx.control_plane.todos.active_state_todo_parser import parse_active_state_todos
+from loopx.control_plane.todos.quota_summary import summarize_user_todos_for_quota
+from loopx.control_plane.testing.quota_fixtures import quota_status_payload
+from loopx.quota import build_quota_should_run
+
+
+SOURCE = """## User Todo / Owner Review Reading Queue
+
+## Agent Todo
+
+- [x] [P1] Complete bounded work with independent acceptance.
+  <!-- loopx:todo todo_id=todo_closed status=done task_class=advancement_task no_followup=true -->
+"""
+
+
+@pytest.mark.parametrize("patch", [
+    {"item_count": True},
+    {"monitor_open_count": "bad", "watch_only_monitor_count": "bad"},
+    {"monitor_open_count": None, "watch_only_monitor_count": None},
+    {"monitor_open_count": -1, "watch_only_monitor_count": -1},
+    {"monitor_open_count": False, "watch_only_monitor_count": False},
+    {"item_count": -1},
+    {"item_count": 1.5},
+    {"item_count": "1"},
+    {"role": "user"},
+    {"monitor_open_count": 1},
+    {"successor_gap_count": 1},
+    {"route_replan_count": 1},
+    {"derived": "true"},
+    {"no_followup_count": True},
+])
+def test_malformed_terminal_proof_cannot_stop_quota(patch: dict[str, object]) -> None:
+    parsed = parse_active_state_todos(SOURCE)
+    source = deepcopy(parsed["agent_todos"])
+    source["terminal_closure_proof"].update(patch)
+    summary = summarize_user_todos_for_quota(source)
+    assert summary["source_completeness"]["status"] == "invalid"
+    assert "closure_intent" not in summary
+    status = quota_status_payload(goal_id="goal-proof-test", status="active",
+        recommended_action="Inspect exact closure evidence.",
+        user_todos=parsed["user_todos"], agent_todos=source)
+    decision = build_quota_should_run(status, goal_id="goal-proof-test")
+    assert decision["effective_action"] != "terminal_no_followup"
+    assert "terminal_state" not in decision["goal_frontier_projection"]
+
+
+def test_valid_complete_source_still_closes_without_changing_its_proof() -> None:
+    parsed = parse_active_state_todos(SOURCE)
+    before = deepcopy(parsed)
+    summary = summarize_user_todos_for_quota(parsed["agent_todos"])
+    assert summary["source_completeness"]["status"] == "valid"
+    assert summary["closure_intent"]["count"] == 1
+    status = quota_status_payload(goal_id="goal-proof-test", status="active",
+        recommended_action="Observe completed work.", **parsed)
+    decision = build_quota_should_run(status, goal_id="goal-proof-test")
+    assert decision["effective_action"] == "terminal_no_followup"
+    assert decision["should_run"] is False
+    assert parsed == before
diff --git a/tests/control_plane_ts/quota_selection.test.ts b/tests/control_plane_ts/quota_selection.test.ts
index 06404d6641..fc86b3dea3 100644
--- a/tests/control_plane_ts/quota_selection.test.ts
+++ b/tests/control_plane_ts/quota_selection.test.ts
@@ -1,7 +1,7 @@
 import assert from "node:assert/strict";
 import test from "node:test";
 import type { JsonObject } from "../../loopx/control_plane/effect_program.ts";
-import { projectQuotaSelection } from "../../loopx/control_plane/todos/quota_selection.ts";
+import { projectQuotaSelection, projectTodoQuotaPlanning } from "../../loopx/control_plane/todos/quota_selection.ts";
 import { productionScaleCoordinationFixture } from "./production_scale_coordination_fixture.ts";
 import {projectAdvancementFrontier} from "../../loopx/control_plane/todos/frontier_revision.ts";
 
@@ -132,3 +132,21 @@ test("malformed facts are rejected, not coerced into scope or execution authorit
   }
   assert.throws(() => projectQuotaSelection(request([], {visibility_limit: -1})));
 });
+
+test("quota v2 validates closure in the existing batch while retaining v0/v1 wire behavior", () => {
+  const resume = {schema_version: "todo_resume_planning_request_v0", sources: {
+    items: [], backlog_items: [], first_open_items: [], deferred_items: [], deferred_resume_candidates: [],
+    resume_blocked_items: [], monitor_open_items: [], current_agent_claimed_monitor_items: [], claimed_monitor_open_items: []}, agent_id: null,
+    item_limit: 8, has_deferred_count: false, has_visible_deferred_count: false, deferred_count: null, available_capabilities: null};
+  const selection = request([], {available: []});
+  const v0 = projectTodoQuotaPlanning({schema_version: "todo_quota_planning_request_v0", resume, selection});
+  const v1 = projectTodoQuotaPlanning({schema_version: "todo_quota_planning_request_v1", resume, selection});
+  assert.deepEqual(v0, v1);
+  assert.equal(v0.source_completeness, undefined);
+  assert.throws(() => projectTodoQuotaPlanning({schema_version: "todo_quota_planning_request_v2", resume, selection}), /closure source/);
+  const v2 = projectTodoQuotaPlanning({schema_version: "todo_quota_planning_request_v2", resume, selection, source_contract: {}});
+  const {source_completeness, closure_intent, ...unchanged} = v2;
+  assert.deepEqual(unchanged, v1);
+  assert.equal((source_completeness as JsonObject).status, "invalid");
+  assert.equal(closure_intent, null);
+});
diff --git a/tests/control_plane_ts/todo_succession.test.ts b/tests/control_plane_ts/todo_succession.test.ts
index d8f483fc09..acb8570d41 100644
--- a/tests/control_plane_ts/todo_succession.test.ts
+++ b/tests/control_plane_ts/todo_succession.test.ts
@@ -1,6 +1,6 @@
 import assert from "node:assert/strict";
 import test from "node:test";
-import {evaluateTodoSuccession, projectTodoSuccession, projectTodoClosure, SUCCESSION_FACT_COLUMNS, SUCCESSION_EVALUATION_COLUMNS} from "../../loopx/control_plane/todos/succession.ts";
+import {evaluateTodoSuccession, projectTodoSuccession, projectTodoClosure, validateTodoClosureSource, SUCCESSION_FACT_COLUMNS, SUCCESSION_EVALUATION_COLUMNS} from "../../loopx/control_plane/todos/succession.ts";
 
 const row = (todo_id: string, overrides = {}) => ({todo_id, status: "done", active: true,
   advancement: true, no_followup: false, successors: [], superseded_by: null,
@@ -77,6 +77,62 @@ test("terminal proofs require full selection and absence of every unresolved obl
   assert.equal((watch.terminal_closure_proof as Record<string, unknown>).all_todos_done, false);
 });
 
+function closureSource() {
+  return {schema_version: "todo_summary_v0", source_section: "Agent Todo",
+    total_count: 1, open_count: 0, done_count: 1, deferred_count: 0,
+    convergence_open_count: 0, completed_without_successor_count: 0, route_continuation_replan_count: 0,
+    items: [{status: "done", done: true, watch_only: false}], monitor_open_items: [],
+    deferred_item_count: 0, deferred_resume_count: 0,
+    source_proof: {schema_version: "todo_source_proof_v0", role: "agent", derived: true, item_count: 1},
+    terminal_closure_proof: {schema_version: "todo_terminal_closure_proof_v0", role: "agent", derived: true,
+      source_section: "Agent Todo", item_count: 1, all_todos_done: true, monitor_open_count: 0,
+      watch_only_monitor_count: 0, successor_gap_count: 0, route_replan_count: 0, no_followup_count: 1},
+    closure_intent: {schema_version: "todo_closure_intent_v0", kind: "no_followup", derived: true, count: 1}};
+}
+test("retained closure proofs reject malformed integer witnesses without false terminal intent", () => {
+  for (const bad of [true, false, null, "0", "bad", -1, 0.5, Number.MAX_SAFE_INTEGER + 1]) {
+    for (const patch of [{item_count: bad}, {monitor_open_count: bad, watch_only_monitor_count: bad},
+      {successor_gap_count: bad}, {route_replan_count: bad}, {no_followup_count: bad}]) {
+      const input = closureSource();
+      const result = validateTodoClosureSource({...input, terminal_closure_proof: {...input.terminal_closure_proof, ...patch}});
+      assert.equal((result.source_completeness as Record<string, unknown>).status, "invalid");
+      assert.equal(result.closure_intent, null);
+    }
+  }
+  const input = closureSource(), before = structuredClone(input);
+  assert.equal((validateTodoClosureSource(input).closure_intent as Record<string, unknown>).count, 1);
+  assert.deepEqual(input, before);
+});
+test("closure source validation retains empty, bounded full-source and watch-only observations", () => {
+  const input = closureSource();
+  const cases = [{...input, total_count: 0, done_count: 0, items: [],
+    source_proof: {...input.source_proof, item_count: 0},
+    terminal_closure_proof: {...input.terminal_closure_proof, item_count: 0, no_followup_count: 0}},
+  {...input, total_count: 13, done_count: 13,
+    source_proof: {...input.source_proof, item_count: 13},
+    terminal_closure_proof: {...input.terminal_closure_proof, item_count: 13}},
+  {...input, total_count: 2, open_count: 1,
+    items: [...input.items, {status: "open", done: false, watch_only: true}],
+    monitor_open_items: [{watch_only: true}], source_proof: {...input.source_proof, item_count: 2},
+    terminal_closure_proof: {...input.terminal_closure_proof, item_count: 2, all_todos_done: false,
+      all_convergent_todos_done: true, monitor_open_count: 1, watch_only_monitor_count: 1}}];
+  for (const source of cases) assert.equal((validateTodoClosureSource(source).source_completeness as Record<string, unknown>).status, "valid");
+  assert.equal(validateTodoClosureSource(cases[0]).closure_intent, null);
+});
+test("open, partial, deferred and replan source evidence cannot reuse a terminal proof", () => {
+  const input = closureSource();
+  for (const patch of [{source_proof: null}, {total_count: true}, {done_count: "1"}, {open_count: 1},
+    {source_section: ""}, {items: []}, {items: [null]}, {items: [{status: "open", done: false}]},
+    {items: [{status: "done", done: false}]}, {items: [{status: "done", done: true, route_continuation_replan_required: true}]},
+    {monitor_open_items: [{watch_only: false}]}, {monitor_open_items: null}, {deferred_item_count: null},
+    {deferred_item_count: 1}, {deferred_resume_count: 1}, {convergence_open_count: true},
+    {completed_without_successor_count: 1}, {route_continuation_replan_count: 1}]) {
+    const result = validateTodoClosureSource({...input, ...patch});
+    assert.equal((result.source_completeness as Record<string, unknown>).status, "invalid");
+    assert.equal(result.closure_intent, null);
+  }
+});
+
 test("interned field sets are lossless and reject invalid references", () => {
   const rows = Array.from({length: 4000}, (_, index) => row(`todo_history_${index}`, {
     active: false, context_fields: index % 2 ? ["claimed_by", "completed_at"] : [],