feat(todos): add an opt-in queue overview

docs/status-data-contract.md

@@ -2772,3 +2772,37 @@ The public examples under `examples/` are sanitized and can be used for demos.
 - Consumers should handle missing `run_history` as unavailable.
 - Consumers should treat `contract.ok=false` as a stronger signal than an empty
   attention queue.
+## Explicit Todo queue overview
+
+`loopx todo list --goal-id <goal-id> --summary` selects the versioned
+`todo_list_summary_v0` overview. It leaves ordinary list output and the existing
+`--thin` view unchanged. The overview is available in JSON and Markdown and
+helps a reader choose which Todo to inspect; it is not execution admission.
+
+The view reuses the canonical Todo query's role, status, Agent scope and stable
+ordering. It shows at most eight records per role (sixteen for both roles).
+`--limit` may lower that per-role cap. Agent-scoped hot reads retain their
+existing omission of done records; explicit status/role reads retain their
+existing selection. Blocked/deferred work and Monitors do not become runnable
+merely because they appear in the overview.
+
+One `todos` array contains `todo_id`, `role`, `title`, `status`, `priority`,
+`claimed_by` and `title_truncated`. Titles are bounded to 180 Unicode code
+points; canonical identities are retained. Markdown marks truncated titles
+with `*`. `todo list --goal-id <goal-id> --todo-id <todo-id>` restores complete
+requirements through the existing exact-read owner. `--summary` is mutually
+exclusive with `--thin` and `--todo-id`.
+
+`todo_count` and `matched_todo_count` cover the complete query match before
+display limits, `returned_todo_count` is the number displayed, and
+`omitted_todo_count` is their difference. `todo_list_summary_projection`
+declares the version, per-role/text limits and full-detail routes. The compact
+`authority_read` retains provider identity/revision/cursor and the canonical
+read/fallback facts. The existing `projection_envelope` distinguishes complete
+source coverage from display truncation. An unavailable source remains an
+error, never a successful empty queue.
+
+The new `todo_list_summary` output-budget variant has separate JSON/Markdown
+ceilings. Existing default and thin budgets remain unchanged. App and Lark
+callers keep their current full/exact views; this opt-in CLI projection does
+not switch consumers, create writes, or replace quota/lease/acceptance checks.

loopx/cli_commands/todo_argument_validation.py

@@ -80,6 +80,7 @@
     ("--agent-id", "agent_id"),
     ("--limit", "todo_limit"),
     ("--thin", "todo_thin"),
+    ("--summary", "todo_summary"),
     ("--state-file", "state_file"),
     ("--execute", "execute"),
     ("--provider-revision", "provider_revision"),
@@ -142,7 +143,7 @@
 # The same strict subset drives validation and action help. Complex lifecycle
 # commands retain full help until their conditional grammar is centralized.
 TODO_ACTION_FIELDS = {
-    "list": frozenset({"role", "todo_id", "status", "agent_id", "todo_limit", "todo_thin", "state_file", "project", "dry_run"}),
+    "list": frozenset({"role", "todo_id", "status", "agent_id", "todo_limit", "todo_thin", "todo_summary", "state_file", "project", "dry_run"}),
     "receipt": frozenset({"operation_id"}),
     "result-read": frozenset({"todo_id"}),
     "plan": frozenset({"text", "agent_id", "project"}),
@@ -689,6 +690,13 @@ def validate_shared_todo_options(args: argparse.Namespace) -> None:
         )
     if args.todo_thin and args.todo_command != "list":
         raise ValueError("--thin is supported only by todo list")
+    if getattr(args, "todo_summary", False):
+        if args.todo_command != "list":
+            raise ValueError("--summary is supported only by todo list")
+        if args.todo_thin:
+            raise ValueError("todo list accepts either --summary or --thin")
+        if args.todo_id:
+            raise ValueError("Exact Todo reads return full requirements; remove --summary or omit --todo-id for a queue overview")
 
 
 def validate_capability_gap_options(args: argparse.Namespace) -> None:

loopx/cli_commands/todo_list.py

@@ -38,6 +38,7 @@ def handle_todo_list_command(
             agent_id=args.agent_id,
             limit=args.todo_limit,
             thin=bool(args.todo_thin),
+            **({"summary": True} if getattr(args, "todo_summary", False) else {}),
             project=Path(args.project).expanduser() if args.project else None,
             state_file=Path(args.state_file).expanduser() if args.state_file else None,
             runtime_root_arg=runtime_root_arg,

loopx/cli_commands/todo_registration.py

@@ -472,6 +472,17 @@ def register_todo_command(
             "lower but not expand that bound."
         ),
     )
+    todo_parser.add_argument(
+        "--summary",
+        dest="todo_summary",
+        action="store_true",
+        help=(
+            "For todo list, show a versioned queue overview with at most eight "
+            "items per role; --limit can lower that bound. Use an exact "
+            "--todo-id read for complete requirements. Cannot combine with "
+            "--thin or --todo-id."
+        ),
+    )
     todo_parser.add_argument("--project", help="Project root. Defaults to the registry goal repo.")
     todo_parser.add_argument("--state-file", help="Active goal state path. Defaults to the registry goal state_file.")
     todo_parser.add_argument("--dry-run", action="store_true", help="Preview the active-state edit without writing.")

loopx/control_plane/testing/cli_output_budget.py

@@ -605,6 +605,20 @@ class CliOutputCommandClassification:
         max_chars={"json": 20_000, "markdown": 6_000},
         max_lines={"json": 600, "markdown": 60},
     ),
+    CliOutputModeVariantSpec(
+        variant_id="todo_list_summary",
+        parent_surface_id="todo_list",
+        command="todo list --summary",
+        output_formats=("json", "markdown"),
+        semantic_json_keys=(
+            "todos", "summary", "matched_todo_count", "returned_todo_count",
+            "omitted_todo_count", "authority_read", "todo_list_summary_projection",
+            "projection_envelope",
+        ),
+        markdown_anchor="# LoopX Todo List Summary",
+        max_chars={"json": 12_000, "markdown": 8_000},
+        max_lines={"json": 320, "markdown": 40},
+    ),
 )
 
 

loopx/control_plane/todos/list_readback.py

@@ -32,6 +32,7 @@ def list_goal_todos(
     runtime_root_arg: str | None = None,
     limit: int | None = None,
     thin: bool = False,
+    summary: bool = False,
     read_scope: Literal["active", "completed_history"] = "active",
     json_references: bool = False,
 ) -> dict[str, Any]:
@@ -44,6 +45,10 @@ def list_goal_todos(
         raise ValueError("todo_id must use the public token shape todo_<letters-digits-underscore-hyphen>")
     if normalized_todo_id and thin:
         raise ValueError("Exact Todo reads return full requirements; remove --thin or omit --todo-id for a bounded list")
+    if summary and thin:
+        raise ValueError("todo list accepts either --summary or --thin")
+    if normalized_todo_id and summary:
+        raise ValueError("Exact Todo reads return full requirements; remove --summary or omit --todo-id for a queue overview")
     normalized_agent_id = normalize_todo_claimed_by(agent_id) if agent_id else None
     if agent_id and not normalized_agent_id:
         raise ValueError("agent_id must be a public-safe agent token such as codex-main-control")
@@ -81,14 +86,16 @@ def list_goal_todos(
     from ..runtime.time import now_utc
 
     now = now_utc()
-    params = {"kind": "query",
+    params: dict[str, Any] = {"kind": "query",
         "source": {"runtime_root": str(runtime_root.expanduser().resolve(strict=False)), "goal_id": goal_id},
         "rollout_events": list(rollout_events), "rollout_event_source": rollout_events.source,
         "observed_at": now.timestamp(), "evaluated_at": now.isoformat(),
         "query": {"goal_id": goal_id, "role": role, "status": normalize_todo_status(status),
             "todo_id": normalized_todo_id, "agent_id": normalized_agent_id, "limit": limit,
             "read_scope": read_scope, "json_references": json_references, "thin": thin,
             "state_file": str(resolved_state_file), "project": str(resolved_project) if resolved_project else None}}
+    if summary:
+        params["query"]["summary"] = True
     result = project_todo_read(params)
     if "payload" not in result:
         failure = result["canonical"]

loopx/control_plane/todos/markdown.py

@@ -145,12 +145,44 @@ def _render_thin_todo_list(payload: dict[str, Any]) -> str:
     return "\n".join(lines)
 
 
+def _render_queue_summary(payload: dict[str, Any]) -> str:
+    projection = payload["todo_list_summary_projection"]
+    authority = payload.get("authority_read") or {}
+    lines = ["# LoopX Todo List Summary", "",
+             f"- goal_id: `{payload.get('goal_id')}`",
+             f"- shown: `{payload.get('returned_todo_count')}` / `{payload.get('matched_todo_count')}`; omitted: `{payload.get('omitted_todo_count')}`",
+             f"- item_limit_per_role: `{projection['item_limit_per_role']}`", ""]
+
+    def cell(value: Any) -> str:
+        return " ".join(str(value if value is not None else "—").split()).replace("|", "\\|")
+
+    if payload["todos"]:
+        lines.extend(["| Role | Priority | Status | Claimed by | Todo | ID |",
+                      "| --- | --- | --- | --- | --- | --- |"])
+        for item in payload["todos"]:
+            cells = [cell(item.get(key)) for key in (
+                "role", "priority", "status", "claimed_by", "title", "todo_id")]
+            if item.get("title_truncated"):
+                cells[4] += " *"
+            lines.append("| " + " | ".join(cells) + " |")
+    else:
+        lines.append("No matching Todos.")
+    if any(item.get("title_truncated") for item in payload["todos"]):
+        lines.extend(["", r"\* Title truncated; the exact read retains the full requirement."])
+    lines.extend(["", f"- source: `{authority.get('source_authority')}` @ `{authority.get('provider_revision')}`",
+                  f"- full_detail: `{projection['full_detail_cold_paths'][0]}`",
+                  f"- readback: {projection['readback']}"])
+    return "\n".join(lines)
+
+
 def render_todo_markdown(payload: dict[str, Any]) -> str:
     if payload.get("todo_detail_projection"):
         return _render_todo_detail(payload)
     if payload.get("command") == "project-markdown":
         return _render_todo_projection(payload)
     if payload.get("command") == "list":
+        if payload.get("summary"):
+            return _render_queue_summary(payload)
         if payload.get("thin"):
             return _render_thin_todo_list(payload)
 

loopx/control_plane/todos/read_presentation.ts

@@ -31,14 +31,18 @@ function compactThinValue(value: unknown): unknown {
   return value;
 }
 
+function displayItems(items: unknown[], hideDone: boolean): unknown[] {
+  return hideDone ? items.filter(row => !dict(row) || row.status !== "done") : items;
+}
+
 export function compactTodoReadSummary(summary: JsonObject, role: string, limit: number | null): JsonObject {
   const result: JsonObject = {}, lanes: JsonObject = {};
   let omittedLanes = 0, omittedDicts = 0;
   for (const [key, value] of Object.entries(summary)) {
     if (Array.isArray(value)) {
       const cap = limit ?? (layout._RETAINED_LANE_LIMITS as Record<string, number>)[key];
       if (cap === undefined) {omittedLanes += Number(nonempty(value)); continue;}
-      const retained = limit === null && key === "items" ? value.filter(row => !dict(row) || row.status !== "done") : value;
+      const retained = displayItems(value, limit === null && key === "items");
       result[key] = retained.slice(0, cap).map(row => compactItem(row));
       if (retained.length > cap) lanes[key] = {shown: cap, total: retained.length};
     } else if (dict(value)) {
@@ -114,3 +118,36 @@ export function compactThinTodoReadSummary(summary: JsonObject, details: {
       omitted_nonempty_dict_count: omittedDicts, full_detail_cold_path: "todo list without --thin or active state"};
     return fields;
 }
+
+/** An opt-in queue overview; selection, full requirements and admission stay upstream. */
+export function compactTodoQueueSummary(payload: JsonObject, hot: boolean): JsonObject {
+  const limit = Math.min(8, typeof payload.explicit_limit === "number" ? payload.explicit_limit : 8);
+  const projection = dict(payload.todo_list_projection) ? payload.todo_list_projection : {};
+  const matched = Number(projection.matched_todo_count ?? payload.todo_count ?? 0);
+  const todos: JsonObject[] = [];
+  for (const role of ["user", "agent"]) {
+    const group = payload[`${role}_todos`];
+    if (!dict(group) || !Array.isArray(group.items)) continue;
+    for (const row of displayItems(group.items, hot).filter(dict).slice(0, limit)) {
+      const full = stripPythonWhitespace(String(row.title || row.text || ""));
+      todos.push({todo_id: row.todo_id, role, title: compactText(full), status: row.status,
+        priority: row.priority ?? null, claimed_by: row.claimed_by ?? null,
+        title_truncated: Array.from(full).length > 180});
+    }
+  }
+  const authority = dict(payload.authority_read) ? payload.authority_read : {};
+  return {ok: payload.ok, dry_run: payload.dry_run, read_only: payload.read_only,
+    command: payload.command, goal_id: payload.goal_id, role: payload.role,
+    status_filter: payload.status_filter, source: payload.source,
+    ...(payload.agent_id_filter === undefined ? {} : {agent_id_filter: payload.agent_id_filter}),
+    ...(payload.unfiltered_todo_count === undefined ? {} : {unfiltered_todo_count: payload.unfiltered_todo_count}),
+    todo_count: matched, matched_todo_count: matched, returned_todo_count: todos.length,
+    omitted_todo_count: Math.max(0, matched - todos.length), summary: true, todos,
+    authority_read: {source_authority: authority.source_authority, provider_revision: authority.provider_revision,
+      cursor: authority.cursor, decision_read_from_provider: authority.decision_read_from_provider,
+      legacy_fallback_used: authority.legacy_fallback_used},
+    todo_list_summary_projection: {schema_version: "todo_list_summary_v0", item_limit_per_role: limit,
+      item_text_limit: 180, counts_cover_full_match: true,
+      full_detail_cold_paths: ["todo list --goal-id <goal-id> --todo-id <todo-id>", "todo list without --summary"],
+      readback: "Queue overview only. Read the exact Todo for complete requirements; quota owns execution admission."}};
+}

loopx/control_plane/todos/read_transaction.ts

@@ -1,13 +1,14 @@
 /** Canonical provider IO and full-source summaries share one read transaction. */
 import type {JsonObject} from "../effect_program.ts";
-import {requireJsonObject} from "../runtime_decode.ts";
+import {requireBoolean, requireJsonObject} from "../runtime_decode.ts";
+import {EffectRuntimeRequestError} from "../effect_runtime_errors.ts";
 import {readCanonicalSnapshot} from "../coordination/canonical_snapshot_page.ts";
 import {readLocalCoordinationTodoSource} from "../coordination/local_authority_read.ts";
 import {LOCAL_COORDINATION_TODO_SNAPSHOT_PAGE_REQUEST_SCHEMA,
   LOCAL_COORDINATION_TODO_LIST_RESULT_SCHEMA} from "../coordination/coordination_state_contract.generated.ts";
 import {projectCanonicalTodoFields, selectTodoReadFields, projectTodoReadGroup, TODO_READ_RESULT} from "./read_projection.ts";
 import {canonicalTodoReadItems, structureCanonicalTodo} from "./read_codec.ts";
-import {compactTodoReadSummary, compactThinTodoRead, todoReadProjectionContract} from "./read_presentation.ts";
+import {compactTodoReadSummary, compactThinTodoRead, compactTodoQueueSummary, todoReadProjectionContract} from "./read_presentation.ts";
 import {projectTodoContextPage} from "./context_projection.ts";
 import {sealProjectionEnvelope, PROJECTION_ENVELOPE_SEAL_REQUEST} from "../projection_envelope.ts";
 
@@ -57,6 +58,8 @@ export async function readHistoricalTodoProjection(input: JsonObject): Promise<J
 
 async function todoListPayload(input: JsonObject, original: JsonObject[], fields: JsonObject, canonical: JsonObject): Promise<JsonObject> {
   const query = requireJsonObject(input.query, "Todo list query");
+  const summary = query.summary === undefined ? false : requireBoolean(query.summary, "Todo summary view");
+  if (summary && (query.thin || query.todo_id)) throw new EffectRuntimeRequestError("Todo summary cannot be combined with thin or exact reads");
   const sourceName = "file_authority";
   const roles: unknown[] = query.role == null ? ["user", "agent"] : [query.role];
   const select = async (source: JsonObject) => selectTodoReadFields({...input, fields: source.fields,
@@ -92,7 +95,7 @@ async function todoListPayload(input: JsonObject, original: JsonObject[], fields
   const matchedCount = todos.length;
   const hot = query.read_scope !== "completed_history" && query.agent_id && query.limit == null &&
     query.role == null && query.status == null && query.todo_id == null;
-  if (hot) {
+  if (hot && !summary) {
     const summaries = requireJsonObject(projection.summaries, "Todo summaries");
     for (const [key, summary] of Object.entries(summaries)) summaries[key] = compactTodoReadSummary(
       requireJsonObject(summary, "Todo summary"), key.replace(/_todos$/, ""), null);
@@ -126,7 +129,8 @@ async function todoListPayload(input: JsonObject, original: JsonObject[], fields
   Object.assign(payload, projection.summaries);
   // The existing formatter is composed here, so a public read never calls back
   // across runtimes for fields of the result it just received.
-  const result = query.todo_id ? projectTodoContextPage({detail_payload: payload}) : query.thin ? compactThinTodoRead(payload) :
+  const result = summary ? compactTodoQueueSummary(payload, Boolean(hot)) :
+    query.todo_id ? projectTodoContextPage({detail_payload: payload}) : query.thin ? compactThinTodoRead(payload) :
     query.json_references ? projectTodoContextPage({list_payload: payload}) : payload;
   const shownCount = query.todo_id ? Number(result.todo !== null) : (result.todos as JsonObject[]).length;
   const observed = String(input.read_observed_at);

tests/control_plane/test_cli_output_budget.py

@@ -689,6 +689,8 @@ def _mode_variant_commands(
         + ["todo", "list", "--goal-id", GOAL_ID, "--limit", "3"],
         "todo_list_thin": common
         + ["todo", "list", "--goal-id", GOAL_ID, "--thin"],
+        "todo_list_summary": common
+        + ["todo", "list", "--goal-id", GOAL_ID, "--summary"],
     }
 
 
@@ -747,6 +749,7 @@ def test_manifest_covers_the_declared_agent_facing_surface_set() -> None:
         "heartbeat_prompt_full",
         "todo_list_limited",
         "todo_list_thin",
+        "todo_list_summary",
     }
     assert set(CLI_OUTPUT_MODE_VARIANT_BY_ID) == expected_variants
     assert manifest["mode_variant_count"] == len(expected_variants)

tests/control_plane/test_todo_list_read_transport.py

@@ -83,7 +83,7 @@ def run(*tail, format_name="json", expected_exit=0, isolated=False):
 def test_unmigrated_query_variants_refuse_without_changing_source(read_cli):
     _, run, _, state = read_cli
     before = state.read_bytes()
-    for options in [(), ("--todo-id", OWN), ("--limit", "1"), ("--thin",)]:
+    for options in [(), ("--todo-id", OWN), ("--limit", "1"), ("--thin",), ("--summary",)]:
         result = run(*options, expected_exit=1)
         assert result["error_code"] == "canonical_authority_required"
         assert result["legacy_fallback_used"] is False