fix(todos): fail closed on malformed closure witnesses in typed quota planning

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 迁移。

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

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};
 }

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

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};
+}