perf(context): read scoped User bodies from one snapshot

docs/project-agent-todo-contract.md

@@ -1393,3 +1393,16 @@ inventory，不依赖摘要条数或 Markdown。补 evidence 不会重新设置
 显式执行 proof 仍严格校验。Monitor 规划/观察、ownership、routing、capability 与终态
 操作不在此 intent 内。缺失 display 不阻止 canonical 更新；提交后的正常投影恢复
 托管 Todo 展示。不支持的操作不会回退 Markdown，也不自动切换 provider 或 promotion。
+
+### Scoped User context source reads
+
+Inline work context obtains its scoped open User Todos in one canonical query.
+The existing typed reader filters scope and status before restoring complete
+selected bodies from that same snapshot. Its internal `full_text` read option
+changes presentation only; ordinary CLI inventories remain bounded, and exact
+reads, authority failures and execution admission retain their existing rules.
+It does not cache an earlier revision or grant permission from a read.
+
+内联工作上下文从同一份 canonical 快照筛选当前 Agent 的开放 User Todo，并恢复完整
+正文，避免清单之后再逐条重读。内部 `full_text` 仅影响正文展示；普通 CLI 清单仍有界，
+不缓存旧版本，也不改变精确读取、权限拒绝或执行准入规则。

loopx/control_plane/todos/list_readback.py

@@ -32,10 +32,13 @@ def list_goal_todos(
     runtime_root_arg: str | None = None,
     limit: int | None = None,
     thin: bool = False,
+    full_text: bool = False,
     summary: bool = False,
     read_scope: Literal["active", "completed_history"] = "active",
     json_references: bool = False,
 ) -> dict[str, Any]:
+    if full_text and (thin or json_references or summary):
+        raise ValueError("Full Todo body reads cannot use compact presentation")
     if read_scope not in {"active", "completed_history"}:
         raise ValueError("Todo read_scope must be active or completed_history")
     if read_scope == "completed_history" and (role != "agent" or status != "done"):
@@ -93,6 +96,7 @@ def list_goal_todos(
         "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,
+            **({"full_text": True} if full_text else {}),
             "state_file": str(resolved_state_file), "project": str(resolved_project) if resolved_project else None}}
     if summary:
         params["query"]["summary"] = True
@@ -106,6 +110,8 @@ def list_goal_todos(
         raise LocalCoordinationAuthorityUnavailable(str(failure.get("reason") or "canonical Todo authority is unavailable"),
             code=code, payload=payload)
     payload = result["payload"]
+    if full_text and payload.get("full_text") is not True:
+        raise ValueError("Complete Todo body delivery is unsupported by this runtime")
     if "canonical_acceptance" in result:
         from .goal_todo_projection import observe_todo_acceptance
         payload = observe_todo_acceptance(payload, result["canonical_acceptance"],

loopx/control_plane/todos/read_transaction.ts

@@ -58,7 +58,10 @@ 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 fullText = query.full_text === undefined ? false : requireBoolean(query.full_text, "full_text");
   const summary = query.summary === undefined ? false : requireBoolean(query.summary, "Todo summary view");
+  if (fullText && (query.thin === true || query.json_references === true || summary))
+    throw new EffectRuntimeRequestError("Full Todo body reads cannot use compact presentation");
   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];
@@ -86,15 +89,29 @@ async function todoListPayload(input: JsonObject, original: JsonObject[], fields
     }
   }
   let todos = projection.todos as JsonObject[];
-  if (query.todo_id && todos.length === 1) {
+  // Restore only already-selected bodies. Scope, status, archive and limits
+  // remain the existing owner's decisions; all bodies share this source head.
+  if (fullText) {
+    const identity = (row: JsonObject) => JSON.stringify([row.todo_id, row.role, row.archive_state ?? "active"]);
+    const bodies = new Map<string, JsonObject | null>();
+    for (const row of original) {
+      const key = identity(row);
+      bodies.set(key, bodies.has(key) ? null : row);
+    }
+    for (const item of todos) {
+      const source = bodies.get(identity(item));
+      if (!source) throw new Error("Todo detail cannot recover an unambiguous full source body");
+      item.text = String(source.text ?? "");
+    }
+  } else if (query.todo_id && todos.length === 1) {
     const item = todos[0]!;
     const matching = original.filter(row => row.todo_id === query.todo_id && row.role === item.role &&
       (row.archive_state ?? "active") === (item.archive_state ?? "active"));
     if (matching.length !== 1) throw new Error("Todo detail cannot recover an unambiguous full source body");
     item.text = String(matching[0]!.text ?? "");
   }
   const matchedCount = todos.length;
-  const hot = query.read_scope !== "completed_history" && query.agent_id && query.limit == null &&
+  const hot = !fullText && query.read_scope !== "completed_history" && query.agent_id && query.limit == null &&
     query.role == null && query.status == null && query.todo_id == null;
   if (hot && !summary) {
     const summaries = requireJsonObject(projection.summaries, "Todo summaries");
@@ -104,7 +121,7 @@ async function todoListPayload(input: JsonObject, original: JsonObject[], fields
   }
   const payload: JsonObject = {ok: true, dry_run: true, read_only: true, command: "list", goal_id: query.goal_id,
     role: query.role ?? "all", status_filter: query.status ?? null, source: sourceName, todo_count: matchedCount,
-    todos, state_file: query.state_file, project: query.project ?? null,
+    todos, ...(fullText ? {full_text: true} : {}), state_file: query.state_file, project: query.project ?? null,
     authority_read: {source_authority: canonical.source_authority, provider_revision: canonical.provider_revision,
       cursor: canonical.cursor, todo_read_model: canonical.todo_read_model,
       decision_read_from_provider: true, legacy_fallback_used: false}};

loopx/control_plane/work_items/context_readback.py

@@ -110,23 +110,13 @@ def _source_content(read, *, registry_path, runtime_root, goal_id, todo_id, agen
 
 def _user_context(*, registry_path, runtime_root, goal_id, agent_id):
     # Reuse the canonical scope owner (global/blocks_agent/bound_agent and
-    # legacy scoping). Exact reads restore full text after scope filtering.
+    # legacy scoping). The typed owner restores full text after scope filtering
+    # from this same canonical snapshot, not a separate read for each body.
     inventory = list_goal_todos(registry_path=registry_path, goal_id=goal_id,
         role="user", status="open", agent_id=agent_id,
-        runtime_root_arg=str(runtime_root))
-    records = []
-    revision = (inventory.get("authority_read") or {}).get("provider_revision")
-    for row in inventory.get("todos", []):
-        detail = list_goal_todos(registry_path=registry_path, goal_id=goal_id,
-            role="user", status="open", agent_id=agent_id, todo_id=row["todo_id"],
-            runtime_root_arg=str(runtime_root))
-        if not detail.get("matched") or detail.get("ambiguous"):
-            raise ValueError("User context changed during source read; rerun guard")
-        if (detail.get("authority_read") or {}).get("provider_revision") != revision:
-            raise ValueError("User context source revision changed; rerun guard")
-        records.append(detail["todo"])
+        runtime_root_arg=str(runtime_root), full_text=True)
     return {"source": inventory["source"], "authority_read": inventory.get("authority_read"),
-        "todos": records}
+        "todos": inventory["todos"]}
 
 
 def attach_work_context(payload: dict[str, Any], *, registry_path: Path,

tests/control_plane/test_selected_work_requirements.py

@@ -100,6 +100,44 @@ def guard(*extra, expected_exit=0):
         assert state.read_text() == state_text
         return
 
+    # The old implementation performs inventory plus one whole-source read per
+    # scoped User Todo. The optimized path must return identical exact bodies
+    # from one read, without changing scope or canonical authority.
+    from loopx.control_plane.work_items import context_readback
+    original_read = context_readback.list_goal_todos
+    expected_users = [original_read(registry_path=registry, runtime_root_arg=str(runtime),
+        goal_id="requirements-goal", role="user", status="open", agent_id="agent-a",
+        todo_id=todo)["todo"] for todo in ("todo_gate", "todo_user_action", "todo_global_action")]
+    bounded = original_read(registry_path=registry, runtime_root_arg=str(runtime),
+        goal_id="requirements-goal", role="user", status="open", agent_id="agent-a")
+    assert bounded["todos"][0]["text"] != gate_text, "ordinary inventories remain bounded"
+    reads = []
+
+    def counted_read(**options):
+        reads.append(options)
+        return original_read(**options)
+
+    monkeypatch.setattr(context_readback, "list_goal_todos", counted_read)
+    direct = context_readback._user_context(registry_path=registry, runtime_root=runtime,
+        goal_id="requirements-goal", agent_id="agent-a")
+    assert direct["todos"] == expected_users
+    assert len(reads) == 1, "scoped full User bodies need one canonical query, not N+1"
+    assert direct["authority_read"]["decision_read_from_provider"] is True
+    from loopx.control_plane.todos import goal_todo_projection
+    typed_read = goal_todo_projection.project_todo_read
+
+    def old_delivery(options):
+        result = typed_read(options)
+        if "payload" in result:
+            result["payload"].pop("full_text", None)
+        return result
+
+    with monkeypatch.context() as patch:
+        patch.setattr(goal_todo_projection, "project_todo_read", old_delivery)
+        with pytest.raises(ValueError, match="unsupported by this runtime"):
+            context_readback._user_context(registry_path=registry, runtime_root=runtime,
+                goal_id="requirements-goal", agent_id="agent-a")
+
     packet = guard()
     channel = packet["interaction_contract"]["agent_channel"]
     users = channel["work_context"]["user_todos"]["todos"]

tests/control_plane_ts/todo_read.test.ts

@@ -91,6 +91,18 @@ for (const provider of ["file", "sqlite"] as const) {
     assert.equal((payload.todo as JsonObject).resume_ready, true);
     assert.equal((payload.authority_read as JsonObject).provider_revision, committed.provider_revision);
     assert.equal((await store.loadAuthority()).status, "loaded");
+    const fullQuery = {...request, query: {...request.query, todo_id: null,
+      json_references: false, full_text: true}};
+    const full = (await readTodo(fullQuery)).payload as JsonObject;
+    assert.equal(full.full_text, true);
+    assert.equal((full.todos as JsonObject[])[0].text, body);
+    assert.equal((full.authority_read as JsonObject).provider_revision, committed.provider_revision);
+    const bounded = (await readTodo({...fullQuery, query: {...fullQuery.query, full_text: false}})).payload as JsonObject;
+    assert.equal(bounded.full_text, undefined);
+    assert.notEqual((bounded.todos as JsonObject[])[0].text, body);
+    await assert.rejects(() => readTodo({...fullQuery, query: {...fullQuery.query, thin: true}}), /compact presentation/);
+    await assert.rejects(() => readTodo({...fullQuery, query: {...fullQuery.query, summary: true}}), /compact presentation/);
+    await assert.rejects(() => readTodo({...fullQuery, query: {...fullQuery.query, full_text: "true"}}), /full_text/);
     const laterTodos = [row("todo_waiting", {status: "blocked", text: "Changed current requirement"}), todos.find(item => item.todo_id === "todo_archived")!]
       .sort((a, b) => String(a.todo_id).localeCompare(String(b.todo_id)));
     const changed = {...projection, todos: laterTodos,