From 8dd7134d9f4d6278ec6f26c356deb82e3c146f30 Mon Sep 17 00:00:00 2001 From: shangzh0 <97216392+shangzh0@users.noreply.github.com> Date: Sat, 31 Oct 2026 19:53:07 +0000 Subject: [PATCH] perf(context): read scoped User bodies from one snapshot Signed-off-by: shangzh0 <97216392+shangzh0@users.noreply.github.com> --- docs/project-agent-todo-contract.md | 13 +++++++ loopx/control_plane/todos/list_readback.py | 6 +++ loopx/control_plane/todos/read_transaction.ts | 26 +++++++++++-- .../work_items/context_readback.py | 18 ++------- .../test_selected_work_requirements.py | 38 +++++++++++++++++++ tests/control_plane_ts/todo_read.test.ts | 11 ++++++ 6 files changed, 94 insertions(+), 18 deletions(-) diff --git a/docs/project-agent-todo-contract.md b/docs/project-agent-todo-contract.md index 1c06f8d9ae..75e312054f 100644 --- a/docs/project-agent-todo-contract.md +++ b/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 清单仍有界， +不缓存旧版本，也不改变精确读取、权限拒绝或执行准入规则。 diff --git a/loopx/control_plane/todos/list_readback.py b/loopx/control_plane/todos/list_readback.py index e52b836eff..78ee35c7b2 100644 --- a/loopx/control_plane/todos/list_readback.py +++ b/loopx/control_plane/todos/list_readback.py @@ -32,9 +32,12 @@ def list_goal_todos( runtime_root_arg: str | None = None, limit: int | None = None, thin: bool = False, + full_text: bool = False, read_scope: Literal["active", "completed_history"] = "active", json_references: bool = False, ) -> dict[str, Any]: + if full_text and (thin or json_references): + 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"): @@ -88,6 +91,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}} result = project_todo_read(params) if "payload" not in result: @@ -99,6 +103,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"], diff --git a/loopx/control_plane/todos/read_transaction.ts b/loopx/control_plane/todos/read_transaction.ts index a27de0371f..86002455b3 100644 --- a/loopx/control_plane/todos/read_transaction.ts +++ b/loopx/control_plane/todos/read_transaction.ts @@ -1,6 +1,7 @@ /** 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, @@ -57,6 +58,9 @@ export async function readHistoricalTodoProjection(input: JsonObject): Promise { const query = requireJsonObject(input.query, "Todo list query"); + const fullText = query.full_text === undefined ? false : requireBoolean(query.full_text, "full_text"); + if (fullText && (query.thin === true || query.json_references === true)) + throw new EffectRuntimeRequestError("Full Todo body reads cannot use compact presentation"); const sourceName = "file_authority"; const roles: unknown[] = query.role == null ? ["user", "agent"] : [query.role]; const select = async (source: JsonObject) => selectTodoReadFields({...input, fields: source.fields, @@ -82,7 +86,21 @@ 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(); + 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")); @@ -90,7 +108,7 @@ async function todoListPayload(input: JsonObject, original: JsonObject[], fields 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) { const summaries = requireJsonObject(projection.summaries, "Todo summaries"); @@ -100,7 +118,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}}; diff --git a/loopx/control_plane/work_items/context_readback.py b/loopx/control_plane/work_items/context_readback.py index 35faa71fd8..9bcfea8d13 100644 --- a/loopx/control_plane/work_items/context_readback.py +++ b/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, diff --git a/tests/control_plane/test_selected_work_requirements.py b/tests/control_plane/test_selected_work_requirements.py index b8b817bdeb..dd47fa448c 100644 --- a/tests/control_plane/test_selected_work_requirements.py +++ b/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"] diff --git a/tests/control_plane_ts/todo_read.test.ts b/tests/control_plane_ts/todo_read.test.ts index a6a28cb2ec..dee14a1614 100644 --- a/tests/control_plane_ts/todo_read.test.ts +++ b/tests/control_plane_ts/todo_read.test.ts @@ -91,6 +91,17 @@ 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, 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,