fix(quota): bound mixed diagnostic summaries without losing unfamiliar records

docs/quota-allocation.md

@@ -923,6 +923,16 @@ read-only quota guard fail. Task graph uses the same canonical Todo rows, while
 extending them with terminal predecessor/evidence rows and remaining an
 agent-neutral read model with separate gate/evidence context.
 
+Known diagnostic Todo rows use the existing bounded summary view. An unfamiliar
+record (including a structured recovery condition) remains complete at its
+original relative position; it does not expand every known sibling. Candidate
+collections and selection obligations stay complete. Omission counts describe
+only removed rows, and explicit detail views retain the original records.
+
+已知诊断 Todo 按原有摘要规则显示。包含结构化恢复条件或未知字段的记录仍按原相对
+顺序完整保留，不再使同组其他已知记录全部展开。候选集合与选择义务保持完整；省略
+计数只统计真正移除的记录，显式详情视图保留原始记录。
+
 Blocked candidates retain typed `resolution_bindings` even when another todo is
 runnable. Each binding names the capability, its resolution owner, and the
 exact `blocked_todo_ids`. The interaction contract turns an owner-held binding

loopx/control_plane/quota/diagnostic_projection.ts

@@ -74,14 +74,12 @@ const DIAGNOSTIC_LANES = new Set([
 // The read codec adds these display fields to canonical records. Retain them
 // on visible items; they do not make a known diagnostic list a future schema.
 const ITEM_FIELDS = new Set<string>([...TODO_CANONICAL_READ_RECORD_FIELDS, "index", "content_revision"]);
-function knownItems(items: unknown[]): boolean {
+function knownItem(value: unknown): boolean {
   const scalar = (value: unknown) => value === null || ["string", "number", "boolean"].includes(typeof value);
-  return items.every(value => {
-    const item = jsonObject(value);
-    return item !== null && (item.schema_version === undefined || item.schema_version === "todo_item_v0") &&
-      Object.entries(item).every(([key, field]) => ITEM_FIELDS.has(key) &&
-        (scalar(field) || (Array.isArray(field) && field.every(scalar))));
-  });
+  const item = jsonObject(value);
+  return item !== null && (item.schema_version === undefined || item.schema_version === "todo_item_v0") &&
+    Object.entries(item).every(([key, field]) => ITEM_FIELDS.has(key) &&
+      (scalar(field) || (Array.isArray(field) && field.every(scalar))));
 }
 
 function compactSummary(summary: JsonObject, role: "agent" | "user", detail: string, captured: boolean): JsonObject {
@@ -92,11 +90,14 @@ function compactSummary(summary: JsonObject, role: "agent" | "user", detail: str
   const owned = previous?.schema_version === schema;
   const omitted: JsonObject = {...(owned ? jsonObject(previous?.omitted_lanes) : {})};
   for (const [key, value] of Object.entries(summary)) {
-    if (!DIAGNOSTIC_LANES.has(key) || !Array.isArray(value) || !knownItems(value)) continue;
+    if (!DIAGNOSTIC_LANES.has(key) || !Array.isArray(value)) continue;
     const limit = limits[key] ?? 0;
-    if (limit) compact[key] = value.slice(0, limit); // Complete items, including unknown fields.
+    // Preserve each unfamiliar record verbatim, wherever it occurs. One such
+    // record must not disable the existing projection for every known sibling.
+    const retained = value.filter((item, i) => i < limit || !knownItem(item));
+    if (limit || retained.length) compact[key] = retained;
     else delete compact[key];
-    if (value.length > limit) omitted[key] = value.length - limit;
+    if (value.length > retained.length) omitted[key] = value.length - retained.length;
   }
   for (const [parentKey, fields] of [
     ["claim_scope", ["other_agent_claimed_items", "blocked_claimed_items"]],
@@ -107,13 +108,15 @@ function compactSummary(summary: JsonObject, role: "agent" | "user", detail: str
     const projected = {...parent};
     for (const key of fields) {
       const value = parent[key];
-      if (!Array.isArray(value) || !knownItems(value)) continue;
+      if (!Array.isArray(value)) continue;
       if (parentKey === "todo_succession_warning" && !Object.hasOwn(parent, "todo_ids")) {
         const ids = value.map(item => jsonObject(item)?.todo_id).filter(id => typeof id === "string" && id).slice(0, 3);
         if (ids.length) projected.todo_ids = ids;
       }
-      delete projected[key];
-      if (value.length) omitted[`${parentKey}.${key}`] = value.length;
+      const retained = value.filter(item => !knownItem(item));
+      if (retained.length) projected[key] = retained;
+      else delete projected[key];
+      if (value.length > retained.length) omitted[`${parentKey}.${key}`] = value.length - retained.length;
     }
     compact[parentKey] = projected;
   }
@@ -122,6 +125,27 @@ function compactSummary(summary: JsonObject, role: "agent" | "user", detail: str
     retained_item_lanes: Object.keys(limits).filter(key => key !== "current_agent_blocker_items" ||
       (Array.isArray(summary[key]) && (summary[key] as unknown[]).length > 0)).sort(),
     omitted_lanes: omitted};
+  // A removed lane has zero visible rows. Fold its known upstream truncation
+  // into one total instead of describing an intermediate list no longer shown.
+  // Retained lanes still need the source's text/row truncation evidence. Unknown
+  // schemas, fields or inconsistent counts remain verbatim for diagnosis.
+  if (!owned && previous?.schema_version === "quota_todo_summary_payload_compaction_v0") {
+    const lanes = jsonObject(previous.compacted_lanes);
+    if (lanes) {
+      const remaining = {...lanes};
+      for (const [key, value] of Object.entries(lanes)) {
+        const counts = jsonObject(value), rows = summary[key];
+        if (Object.hasOwn(compact, key) || !Array.isArray(rows) || !counts ||
+            Object.keys(counts).some(field => field !== "shown" && field !== "total") ||
+            !Number.isSafeInteger(counts.shown) || !Number.isSafeInteger(counts.total) ||
+            counts.shown !== rows.length || (counts.total as number) < rows.length ||
+            omitted[key] !== rows.length) continue;
+        omitted[key] = counts.total;
+        delete remaining[key];
+      }
+      metadata.source_projection = {...previous, compacted_lanes: remaining};
+    }
+  }
   delete metadata.full_detail_cold_path;
   delete metadata.full_detail_capture_ref;
   metadata[captured ? "full_detail_capture_ref" : "full_detail_cold_path"] = detail;

tests/control_plane/test_context_delivery_cli.py

@@ -171,6 +171,44 @@ def test_page_output_budget_rejects_guard_use_without_admission(tmp_path, monkey
     assert _spend_run_count(runtime) == 0
 
 
+@pytest.mark.parametrize("provider", ["file", "sqlite"])
+@pytest.mark.parametrize("profile", ["codex_app", "generic_cli"])
+def test_resumed_task_keeps_its_condition_without_expanding_diagnostic_siblings(tmp_path, monkeypatch, provider, profile):
+    isolate_sqlite_runtime(tmp_path, monkeypatch)
+    project, runtime, registry = _write_fixture(tmp_path, required_capability="shell")
+    state = project / f".codex/goals/{GOAL_ID}/ACTIVE_GOAL_STATE.md"
+    template = read_fixture_todos(state)[0]
+    todos = [{**deepcopy(template), "todo_id": TODO_ID if i == 0 else f"todo_work_{i}",
+        "text": f"Work item {i}. " + "Keep the complete requirement. " * 20 + "Final condition: stop before publication.",
+        "title": f"Work item {i}", "index": i} for i in range(8)]
+    for item in todos:
+        item["title"] = item["text"]
+    todos[1]["resume_when"] = "todo_done:todo_dependency"
+    todos.append({**deepcopy(template), "todo_id": "todo_dependency", "status": "done", "done": True,
+        "text": "Dependency accepted.", "index": 8, "no_followup": True})
+    initialize_canonical_authority(runtime, GOAL_ID,
+        build_todo_runtime_shadow_projection(goal_id=GOAL_ID, todos=todos, handoff_mode="soft_claim"),
+        state_path=state, provider=provider)
+    host = ["--codex-app"] if profile == "codex_app" else ["--runtime-profile", profile]
+    before = state.read_bytes()
+    code, guard = _run_cli(registry, runtime, "quota", "should-run", *host,
+        "--goal-id", GOAL_ID, "--agent-id", AGENT_ID, "--todo-id", TODO_ID,
+        "--turn-instance-id", "mixed-diagnostics", cwd=project)
+    assert code == 0 and guard["should_run"], guard
+    summary = guard["agent_todo_summary"]
+    assert [item["todo_id"] for item in summary["backlog_items"]] == ["todo_work_1"]
+    condition = summary["backlog_items"][0]["resume_condition"]
+    assert condition["satisfied"] and condition["target_todo_id"] == "todo_dependency"
+    assert len(summary["first_executable_items"]) == 3
+    candidates = guard["capability_gate"]["runnable_candidates"]
+    assert {row["todo_id"] for row in candidates} == {row["todo_id"] for row in todos[:8]}
+    assert all(row["title"].endswith("Final condition: stop before publication.") for row in candidates)
+    assert state.read_bytes() == before and _spend_run_count(runtime) == 0
+    code, detailed = _run_cli(registry, runtime, "todo", "list", "--goal-id", GOAL_ID,
+        "--todo-id", "todo_work_7", cwd=project)
+    assert code == 0 and detailed["todo"]["text"] == todos[7]["text"]
+
+
 @pytest.mark.parametrize("provider", ["file", "sqlite"])
 @pytest.mark.parametrize("profile", ["codex_app", "generic_cli"])
 def test_cli_capture_exact_delta_and_same_capture_recovery(tmp_path: Path, monkeypatch, provider, profile):

tests/control_plane/test_quota_cli_projection.py

@@ -17,6 +17,44 @@
 )
 
 
+@pytest.mark.parametrize("variant", ["known", "future-schema", "future-field", "mismatch", "negative", "unsafe"])
+def test_hidden_diagnostic_lane_counts_all_source_rows_once(variant):
+    counts = {"shown": 2, "total": 8}
+    source = {"schema_version": "quota_todo_summary_payload_compaction_v0",
+        "item_text_limit": 180,
+        "compacted_lanes": {"backlog_items": counts,
+            "first_executable_items": {"shown": 2, "total": 9}},
+        "future_fact": {"value": 2**60 + 1}}
+    if variant == "future-schema":
+        source["schema_version"] = "future_v1"
+    if variant == "future-field":
+        counts["pending"] = 3
+    if variant == "mismatch":
+        counts["shown"] = 1
+    if variant == "negative":
+        counts["total"] = -1
+    if variant == "unsafe":
+        counts["total"] = 2**60 + 1
+    summary = {"schema_version": "todo_summary_v0", "payload_compaction": source,
+        "backlog_items": _items(2, prefix="cold"),
+        "first_executable_items": _items(2, prefix="work")}
+    payload = {"agent_todo_summary": summary}
+    original = deepcopy(payload)
+    result = compact_quota_should_run_cli_payload(payload)
+    actual = result["agent_todo_summary"]
+    metadata = actual["payload_compaction"]
+    assert "backlog_items" not in actual
+    assert actual["first_executable_items"] == summary["first_executable_items"]
+    assert metadata["omitted_lanes"]["backlog_items"] == (8 if variant == "known" else 2)
+    expected = deepcopy(source)
+    if variant == "known":
+        del expected["compacted_lanes"]["backlog_items"]
+    assert metadata["source_projection"] == expected
+    assert compact_quota_should_run_cli_payload(result) == result
+    assert compact_quota_should_run_cli_payload(payload, include_todo_summary_detail=True) == payload
+    assert payload == original
+
+
 @pytest.mark.parametrize("variant", ["text", "title", "different", "short", "incomplete", "unverified",
     "unmatched", "ambiguous", "partial", "failed", "wrong-id", "wrong-selection", "duplicate", "existing-ref",
     "archived", "done", "boolean-id"])
@@ -1007,6 +1045,44 @@ def test_shared_diagnostics_preserve_unknown_lists_nested_fields_and_complete_it
     assert detailed == payload
 
 
+@pytest.mark.parametrize("role", ["agent", "user"])
+@pytest.mark.parametrize("position", [1, 4, 7])
+@pytest.mark.parametrize("exception", ["resume", "future", "schema", "scalar"])
+def test_unfamiliar_diagnostic_row_does_not_expand_known_siblings(role, position, exception):
+    lane = "first_executable_items" if role == "agent" else "first_open_items"
+    rows = _items(8, prefix="known")
+    exceptional = {**rows[position], "resume_condition": {"schema_version": "todo_resume_condition_v0",
+        "resume_when": "todo_done:dependency", "satisfied": False, "target_todo_id": "dependency"}}
+    if exception == "future":
+        exceptional = {**rows[position], "future_control": {"stop": "Preserve the final constraint.", "epoch": 2**60 + 1}}
+    elif exception == "schema":
+        exceptional = {**rows[position], "schema_version": "future_todo_v1"}
+    elif exception == "scalar":
+        exceptional = None
+    rows[position] = exceptional
+    summary = {"schema_version": "todo_summary_v0", lane: rows, "backlog_items": rows,
+        "claim_scope": {"blocked_claimed_items": rows}, "todo_succession_warning": {"items": rows}}
+    payload = {f"{role}_todo_summary": summary, "capability_gate": {"runnable_candidates": rows},
+        "action_portfolio": {"suggested_actions": rows}, "planning_horizon": {"work_items": rows}}
+    original = deepcopy(payload)
+    projected = compact_quota_should_run_cli_payload(payload)
+    actual = projected[f"{role}_todo_summary"]
+    expected = rows[:3] + ([exceptional] if position >= 3 else [])
+    assert actual[lane] == expected
+    assert actual["backlog_items"] == [exceptional]
+    assert actual["claim_scope"]["blocked_claimed_items"] == [exceptional]
+    assert actual["todo_succession_warning"]["items"] == [exceptional]
+    omitted = actual["payload_compaction"]["omitted_lanes"]
+    assert omitted[lane] == len(rows) - len(expected)
+    assert omitted["backlog_items"] == omitted["claim_scope.blocked_claimed_items"] == 7
+    for key in ("capability_gate", "action_portfolio", "planning_horizon"):
+        assert projected[key] == original[key]
+    assert compact_quota_should_run_cli_payload(projected) == projected
+    assert compact_quota_should_run_cli_payload(payload, include_todo_summary_detail=True,
+        include_user_todo_summary_detail=True) == original
+    assert payload == original
+
+
 def test_readonly_plan_projects_all_rows_in_one_typed_call(monkeypatch):
     from loopx.control_plane.quota.cli_projection import compact_quota_plan_cli_payload
     from loopx.control_plane.quota import turn_envelope