fix(quota): bound CLI todo diagnostics

loopx/cli_commands/quota.py

@@ -16,6 +16,9 @@
 from ..rollout_event_log import load_rollout_events, rollout_event_log_path
 from ..upgrade import resolve_codex_app_automation_rrule
 from ..control_plane.quota.monitor_poll import find_quota_monitor_poll_turn
+from ..control_plane.quota.cli_projection import (
+    compact_quota_should_run_cli_payload,
+)
 from ..control_plane.quota.turn_envelope import build_turn_envelope
 from ..control_plane.quota.live_decision import build_live_quota_should_run_decision
 from ..control_plane.quota.scheduler_ack import (
@@ -210,6 +213,14 @@ def register_quota_command(subparsers: argparse._SubParsersAction) -> None:
             "and Claude loop runtimes in `quota should-run` JSON."
         ),
     )
+    quota_parser.add_argument(
+        "--include-todo-summary-detail",
+        action="store_true",
+        help=(
+            "Include cold-path agent todo diagnostic lanes in `quota should-run`. "
+            "The default keeps counts and decision-relevant todo items only."
+        ),
+    )
     quota_parser.add_argument(
         "--codex-app-current-rrule",
         help=(
@@ -374,6 +385,13 @@ def handle_quota_command(
     try:
         if bool(getattr(args, "turn_envelope", False)) and args.quota_command != "should-run":
             raise ValueError("--turn-envelope is only valid with `quota should-run`")
+        if (
+            bool(getattr(args, "include_todo_summary_detail", False))
+            and args.quota_command != "should-run"
+        ):
+            raise ValueError(
+                "--include-todo-summary-detail is only valid with `quota should-run`"
+            )
         raw_heartbeat_turn_id = getattr(args, "turn_instance_id", None)
         heartbeat_turn_id = normalize_turn_instance_id(raw_heartbeat_turn_id)
         if heartbeat_turn_id and args.quota_command != "should-run":
@@ -868,6 +886,11 @@ def handle_quota_command(
             payload,
             scheduler_execution_context=scheduler_context,
         )
+    elif (
+        args.quota_command == "should-run"
+        and not bool(getattr(args, "include_todo_summary_detail", False))
+    ):
+        payload = compact_quota_should_run_cli_payload(payload)
     renderer = (
         render_turn_envelope_markdown
         if bool(getattr(args, "turn_envelope", False))

loopx/control_plane/quota/cli_projection.py

@@ -0,0 +1,137 @@
+from __future__ import annotations
+
+from typing import Any
+
+
+QUOTA_CLI_TODO_SUMMARY_COMPACTION_SCHEMA_VERSION = (
+    "quota_cli_todo_summary_compaction_v0"
+)
+QUOTA_CLI_TODO_SUMMARY_DETAIL_COMMAND = (
+    "quota should-run --include-todo-summary-detail"
+)
+_RETAINED_AGENT_ITEM_LANES = {
+    "first_executable_items": 3,
+    "unclaimed_priority_open_items": 3,
+    "monitor_due_items": 1,
+    "monitor_capability_blocked_due_items": 2,
+    "monitor_schedule_gap_items": 1,
+}
+_RETAINED_AGENT_ITEM_FIELDS = (
+    "schema_version",
+    "todo_id",
+    "index",
+    "text",
+    "status",
+    "priority",
+    "task_class",
+    "action_kind",
+    "claimed_by",
+    "bound_agent",
+    "goal_bound",
+    "blocks_agent",
+    "global_gate",
+    "task_repository",
+    "required_capabilities",
+    "required_write_scopes",
+    "excluded_agents",
+    "unblocks_todo_id",
+    "continuation_policy",
+    "resume_when",
+    "target_key",
+    "cadence",
+    "next_due_at",
+    "expires_at",
+    "last_checked_at",
+    "consecutive_no_change",
+    "material_change",
+    "result_hash",
+)
+_RETAINED_SUCCESSION_WARNING_TODO_IDS = 3
+
+
+def _compact_agent_item(item: Any) -> Any:
+    if not isinstance(item, dict):
+        return item
+    return {
+        key: item[key]
+        for key in _RETAINED_AGENT_ITEM_FIELDS
+        if key in item
+    }
+
+
+def _compact_nested_item_lists(
+    value: dict[str, Any],
+    *,
+    omitted_lanes: dict[str, int],
+    path: str,
+) -> dict[str, Any]:
+    compact: dict[str, Any] = {}
+    for key, child in value.items():
+        if isinstance(child, list) and key.endswith("items"):
+            if path == "todo_succession_warning":
+                todo_ids = [
+                    str(item.get("todo_id"))
+                    for item in child
+                    if isinstance(item, dict) and item.get("todo_id")
+                ][:_RETAINED_SUCCESSION_WARNING_TODO_IDS]
+                if todo_ids:
+                    compact["todo_ids"] = todo_ids
+            if child:
+                omitted_lanes[f"{path}.{key}"] = len(child)
+            continue
+        compact[key] = child
+    return compact
+
+
+def _compact_agent_todo_summary(summary: dict[str, Any]) -> dict[str, Any]:
+    compact: dict[str, Any] = {}
+    omitted_lanes: dict[str, int] = {}
+    for key, value in summary.items():
+        if isinstance(value, list):
+            limit = _RETAINED_AGENT_ITEM_LANES.get(key)
+            if limit is None:
+                if value:
+                    omitted_lanes[key] = len(value)
+                continue
+            compact[key] = [
+                _compact_agent_item(item)
+                for item in value[:limit]
+            ]
+            if len(value) > limit:
+                omitted_lanes[key] = len(value) - limit
+            continue
+        if isinstance(value, dict):
+            compact[key] = _compact_nested_item_lists(
+                value,
+                omitted_lanes=omitted_lanes,
+                path=key,
+            )
+            continue
+        compact[key] = value
+
+    compact["payload_compaction"] = {
+        "schema_version": QUOTA_CLI_TODO_SUMMARY_COMPACTION_SCHEMA_VERSION,
+        "retained_item_lanes": sorted(_RETAINED_AGENT_ITEM_LANES),
+        "omitted_lanes": omitted_lanes,
+        "full_detail_cold_path": QUOTA_CLI_TODO_SUMMARY_DETAIL_COMMAND,
+    }
+    return compact
+
+
+def compact_quota_should_run_cli_payload(
+    payload: dict[str, Any],
+) -> dict[str, Any]:
+    """Bound CLI-only todo diagnostics after the full decision is computed."""
+
+    summary = payload.get("agent_todo_summary")
+    if not isinstance(summary, dict):
+        return payload
+    compact = dict(payload)
+    compact["agent_todo_summary"] = _compact_agent_todo_summary(summary)
+    compact["todo_summary_projection"] = {
+        "schema_version": QUOTA_CLI_TODO_SUMMARY_COMPACTION_SCHEMA_VERSION,
+        "mode": "compact_hot_path",
+        "compacted_roles": ["agent"],
+        "detail_ref": QUOTA_CLI_TODO_SUMMARY_DETAIL_COMMAND,
+    }
+    return compact

loopx/control_plane/testing/cli_output_budget.py

@@ -126,21 +126,24 @@ class CliOutputCommandClassification:
         owner="quota guard",
         consumer_action="decide whether one agent may run",
         qualification_policy="absolute_hot_path",
-        cold_path="status, history, active state, and --include-scheduler-detail",
+        cold_path=(
+            "status, history, active state, --include-scheduler-detail, and "
+            "--include-todo-summary-detail"
+        ),
         semantic_json_keys=("interaction_contract", "scheduler_hint", "selected_todo"),
         markdown_anchor="# LoopX Quota Should Run",
         max_chars={
-            "small": {"json": 26_000, "markdown": 6_700},
-            "crowded": {"json": 40_000, "markdown": 7_800},
-            "multi_agent": {"json": 36_000, "markdown": 7_000},
+            "small": {"json": 20_000, "markdown": 6_700},
+            "crowded": {"json": 30_000, "markdown": 7_800},
+            "multi_agent": {"json": 23_000, "markdown": 7_000},
         },
         max_lines={
-            "small": {"json": 720, "markdown": 72},
-            "crowded": {"json": 1_050, "markdown": 78},
-            "multi_agent": {"json": 1_000, "markdown": 75},
+            "small": {"json": 520, "markdown": 72},
+            "crowded": {"json": 750, "markdown": 78},
+            "multi_agent": {"json": 650, "markdown": 75},
         },
         scale_axis="todo_count",
-        max_json_growth_chars_per_unit=400,
+        max_json_growth_chars_per_unit=300,
     ),
     CliOutputBudgetSpec(
         surface_id="loopx_turn_plan",
@@ -373,6 +376,16 @@ class CliOutputCommandClassification:
         max_chars={"json": 31_000, "markdown": 6_700},
         max_lines={"json": 820, "markdown": 72},
     ),
+    CliOutputModeVariantSpec(
+        variant_id="quota_should_run_todo_summary_detail",
+        parent_surface_id="quota_should_run",
+        command="quota should-run --include-todo-summary-detail",
+        output_formats=("json", "markdown"),
+        semantic_json_keys=("interaction_contract", "scheduler_hint", "selected_todo"),
+        markdown_anchor="# LoopX Quota Should Run",
+        max_chars={"json": 55_000, "markdown": 7_800},
+        max_lines={"json": 1_350, "markdown": 78},
+    ),
     CliOutputModeVariantSpec(
         variant_id="quota_should_run_turn_envelope",
         parent_surface_id="quota_should_run",

loopx/presentation/renderers/quota_markdown.py

@@ -708,6 +708,14 @@ def render_quota_should_run_markdown(payload: dict[str, Any]) -> str:
     if control_plane:
         lines.append(f"- control_plane: {control_plane_policy_summary(control_plane)}")
 
+    def compact_todo_identity(todo: dict[str, Any]) -> str:
+        parts: list[str] = []
+        for key in ("todo_id", "status", "priority", "task_class", "action_kind"):
+            value = str(todo.get(key) or "").strip()
+            if value:
+                parts.append(f"{key}={value}")
+        return " ".join(parts)
+
     def append_todo_summary(label: str, summary: dict[str, Any]) -> None:
         summary_parts = [
             f"open={summary.get('open_count')}",
@@ -718,23 +726,60 @@ def append_todo_summary(label: str, summary: dict[str, Any]) -> None:
             summary_parts.insert(2, f"unclaimed={summary.get('unclaimed_open_count', 0)}")
         if summary.get("monitor_due_count"):
             summary_parts.append(f"monitor_due={summary.get('monitor_due_count')}")
+        if summary.get("monitor_schedule_gap_count"):
+            summary_parts.append(
+                f"monitor_schedule_gap={summary.get('monitor_schedule_gap_count')}"
+            )
         if summary.get("completed_without_successor_count"):
             summary_parts.append(
                 f"succession_warning={summary.get('completed_without_successor_count')}"
             )
         lines.append(f"- {label}_summary: {' '.join(summary_parts)}")
+        for lane, suffix, limit in (
+            ("unclaimed_priority_open_items", "unclaimed_candidates", 3),
+            ("monitor_due_items", "monitor_due", 1),
+            (
+                "monitor_capability_blocked_due_items",
+                "monitor_capability_blocked_due",
+                2,
+            ),
+            ("monitor_schedule_gap_items", "monitor_schedule_gap", 1),
+        ):
+            lane_items = (
+                summary.get(lane)
+                if isinstance(summary.get(lane), list)
+                else []
+            )
+            identities = [
+                compact_todo_identity(item)
+                for item in lane_items[:limit]
+                if isinstance(item, dict) and item.get("todo_id")
+            ]
+            if identities:
+                lines.append(
+                    f"- {label}_{suffix}: {'; '.join(identities)}"
+                )
         succession_warning = (
             summary.get("todo_succession_warning")
             if isinstance(summary.get("todo_succession_warning"), dict)
             else {}
         )
         if succession_warning:
+            projected_todo_ids = (
+                succession_warning.get("todo_ids")
+                if isinstance(succession_warning.get("todo_ids"), list)
+                else []
+            )
             warning_items = (
                 succession_warning.get("items")
                 if isinstance(succession_warning.get("items"), list)
                 else []
             )
             todo_ids = [
+                str(todo_id)
+                for todo_id in projected_todo_ids[:3]
+                if str(todo_id or "").strip()
+            ] or [
                 str(item.get("todo_id"))
                 for item in warning_items[:3]
                 if isinstance(item, dict) and item.get("todo_id")
@@ -746,7 +791,17 @@ def append_todo_summary(label: str, summary: dict[str, Any]) -> None:
                 f"count={succession_warning.get('count')} "
                 f"todo_ids={todo_ids_text}"
             )
-        first_open = list(summary.get("first_open_items") if isinstance(summary.get("first_open_items"), list) else [])
+        first_open_key = (
+            "first_executable_items"
+            if label == "agent_todo"
+            and isinstance(summary.get("first_executable_items"), list)
+            else "first_open_items"
+        )
+        first_open = list(
+            summary.get(first_open_key)
+            if isinstance(summary.get(first_open_key), list)
+            else []
+        )
         if label == "user_todo" and isinstance(summary.get("user_action_items"), list):
             first_open.extend(
                 item
@@ -761,12 +816,16 @@ def append_todo_summary(label: str, summary: dict[str, Any]) -> None:
                 continue
             index = todo.get("index")
             suffix = f"[{index}]" if index is not None else ""
-            claim_suffix = (
-                f" claimed_by={todo.get('claimed_by')}"
-                if todo.get("claimed_by")
-                else ""
+            identity_suffix = (
+                f" todo_id={todo.get('todo_id')}"
+                if label == "agent_todo" and todo.get("todo_id")
+                else (
+                    f" claimed_by={todo.get('claimed_by')}"
+                    if todo.get("claimed_by")
+                    else ""
+                )
             )
-            lines.append(f"- {label}_next{suffix}: {text}{claim_suffix}")
+            lines.append(f"- {label}_next{suffix}: {text}{identity_suffix}")
         first_keys = {
             (
                 str(todo.get("todo_id") or ""),
@@ -1254,6 +1313,15 @@ def append_todo_summary(label: str, summary: dict[str, Any]) -> None:
     agent_todo_summary = (
         payload.get("agent_todo_summary") if isinstance(payload.get("agent_todo_summary"), dict) else {}
     )
+    selected_todo = (
+        payload.get("selected_todo")
+        if isinstance(payload.get("selected_todo"), dict)
+        else {}
+    )
+    selected_todo_id = str(selected_todo.get("todo_id") or "").strip()
+    lane_todo_id = str(agent_lane_next_action.get("todo_id") or "").strip()
+    if selected_todo and (not selected_todo_id or selected_todo_id != lane_todo_id):
+        lines.append(f"- selected_todo: {compact_todo_identity(selected_todo)}")
     if agent_todo_summary:
         append_todo_summary("agent_todo", agent_todo_summary)
     todo_write_hint = payload.get("todo_write_hint") if isinstance(payload.get("todo_write_hint"), dict) else {}